Menu
Kailash Kaizen (Python) — MANDATORY for AI agents/RAG/signatures. Custom LLM agents BLOCKED.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Based on SOC occupation classification
Claude Code architecture — artifact design, context, agentic patterns. For CC audit/build.
Conformance Walk — freeze-then-judge on the source→delivered→live axis: one cw_core + Source/Delivered/Live adapter families, coverage vs pass-rate, discrete verdicts. Use for any testable surface.
Kailash security (Python) — validation, secrets, injection, authn/z. Hardcoded secrets BLOCKED.
/onboard procedure: read roster + team-memory + posture + claims + codify lease + rules-changed for a new operator joining a multi-operator COC repo.
/certify procedure: brief → probe → gate at 100%; loops failed questions until pass. NO Claude-assistance during gate phase. Curated bank, not LLM-generated.
/ecosystem-init procedure — write the D6 ecosystem-config, run the disclosure scan before write, establish genesis via runEnrollmentCeremony, scaffold non-Kailash STACK.md.
| name | kaizen |
| description | Kailash Kaizen (Python) — MANDATORY for AI agents/RAG/signatures. Custom LLM agents BLOCKED. |
Kaizen is a production-ready AI agent framework built on Kailash Core SDK that provides signature-based programming and multi-agent coordination.
Kaizen enables building sophisticated AI agents with:
.kaizen/ directoryfrom kaizen.core.base_agent import BaseAgent
from kaizen.signatures import Signature, InputField, OutputField
from dataclasses import dataclass
# Define agent signature (type-safe interface)
class SummarizeSignature(Signature):
text: str = InputField(description="Text to summarize")
summary: str = OutputField(description="Generated summary")
# Define configuration
@dataclass
class SummaryConfig:
llm_provider: str = os.environ.get("LLM_PROVIDER", "openai")
model: str = os.environ["LLM_MODEL"]
temperature: float = 0.7
# Create agent with signature
class SummaryAgent(BaseAgent):
def __init__(self, config: SummaryConfig):
super().__init__(
config=config,
signature=SummarizeSignature()
)
# Execute
agent = SummaryAgent(SummaryConfig())
result = agent.run(text="Long text here...")
print(result['summary'])
from kaizen_agents.patterns.pipeline import Pipeline
# Ensemble: Multi-perspective collaboration
pipeline = Pipeline.ensemble(
agents=[code_expert, data_expert, writing_expert, research_expert],
synthesizer=synthesis_agent,
discovery_mode="a2a", # A2A semantic matching
top_k=3 # Select top 3 agents
)
# Execute - automatically selects best agents for task
result = pipeline.run(task="Analyze codebase", input="repo_path")
# Router: Intelligent task delegation
router = Pipeline.router(
agents=[code_agent, data_agent, writing_agent],
routing_strategy="semantic" # A2A-based routing
)
# Blackboard: Iterative problem-solving
blackboard = Pipeline.blackboard(
agents=[solver, analyzer, optimizer],
controller=controller,
max_iterations=10,
discovery_mode="a2a"
)
For in-depth documentation, see the kaizen package (docs/):
Core Guides:
Reference Documentation:
kaizen.llm.LlmClient + four-axis LlmDeployment + 24 presets + from_env() precedence + wire-send dispatch. Load first when touching LlmDeployment, LlmClient.embed()/complete(), wire_protocols/*, or adding a new wire-send method. Spec: specs/kaizen-llm-deployments.md.Pipeline Patterns (9 Composable Patterns):
AgentManifest with [agent] and [governance] TOML sectionsGovernanceManifest with risk_level, suggested_posture, budgetintrospect_agent() for runtime metadata extraction (Python API only, NOT MCP)deploy() / deploy_local() for local FileRegistry or remote CARE PlatformFileRegistry with atomic writes and path traversal preventionvalidate_dag() with iterative DFS cycle detection (max_agents=1000)check_schema_compatibility() with JSON Schema structural subtyping and type wideningestimate_cost() with historical data projection and confidence levelsCatalogMCPServer with 11 tools: Discovery (4), Deployment (3), Application (2), Governance (2)python -m kaizen.mcp.catalog_serverBudgetTracker with two-phase reserve/record, threshold callbacks, on_record() APIPostureBudgetIntegration links budget to posture state machineEnvelopeTracker with atomic recording, child allocation, reclamationEnvelopeSplitter for stateless ratio-based budget divisionEnvelopeEnforcer middleware with gradient zones (AutoApproved/Flagged/Held/Blocked)ContextScope tree with parent traversal and child mergeScopeProjection glob patterns (allow/deny with deny precedence)DataClassification 5-level clearance filteringMessageRouter with 8-step validationDeadLetterStore bounded ring buffer for undeliverable messagesAgentFactory with 8-check spawn preconditionsPlanValidator structural + envelope validationPlanExecutor with gradient rules (G1-G8)Located in the package source:
09-performance-optimization-guide.md) - Caching (10-100x speedup), parallel execution06-specialist-system-guide.md) - Claude Code-style specialists and skills00-native-tools-guide.md) - TAOD loop tool integration01-runtime-abstraction-guide.md) - Multi-runtime support02-local-kaizen-adapter-guide.md) - TAOD loop implementation03-memory-provider-guide.md) - Memory provider interface04-multi-llm-routing-guide.md) - Intelligent LLM selection05-unified-agent-api-guide.md) - Simplified 2-line agent creation07-task-skill-tools-guide.md) - Subagent spawning08-claude-code-parity-tools-guide.md) - 7 parity toolsSignatures define type-safe interfaces for agents:
Foundation for all Kaizen agents:
1. Hooks System - Event-driven observability framework
2. Checkpoint System - Persistent state management
3. Interrupt Mechanism - Graceful shutdown and execution control
4. Memory System - 3-tier hierarchical storage
5. Planning Agents - Structured workflow orchestration
6. Meta-Controller Routing - Intelligent task delegation
For 100+ agent distributed systems:
Use Kaizen when you need to:
Use Pipeline Patterns When:
from kaizen.core.base_agent import BaseAgent
from dataflow import DataFlow
class DataAgent(BaseAgent):
def __init__(self, config, db: DataFlow):
self.db = db
super().__init__(config=config, signature=MySignature())
from kaizen.core.base_agent import BaseAgent
from nexus import Nexus
# Deploy agents via API/CLI/MCP
agent_workflow = create_agent_workflow()
app = Nexus()
app.register("agent", agent_workflow.build())
app.start() # Agents available via all channels
from kaizen.core.base_agent import BaseAgent
from kailash.workflow.builder import WorkflowBuilder
# Embed agents in workflows
workflow = WorkflowBuilder()
workflow.add_node("KaizenAgent", "agent1", {
"agent": my_agent,
"input": "..."
})
As of v2.5.0, provider configuration follows an explicit over implicit model. Structured output config is separated from provider-specific settings.
| Field | Purpose | Example |
|---|---|---|
response_format | Structured output config (json_schema, json_object) | {"type": "json_schema", "json_schema": {}} |
provider_config | Provider-specific operational settings only | {"api_version": "2024-10-21"} |
structured_output_mode | Controls auto-generation: "auto" (deprecated), "explicit", "off" | "explicit" |
from kaizen.core.config import BaseAgentConfig
from kaizen.core.structured_output import create_structured_output_config
# Explicit mode (recommended)
config = BaseAgentConfig(
llm_provider="openai",
model=os.environ["LLM_MODEL"],
response_format=create_structured_output_config(MySignature(), strict=True),
structured_output_mode="explicit",
)
# Azure with provider-specific settings (separate from response_format)
config = BaseAgentConfig(
llm_provider="azure",
model=os.environ["LLM_MODEL"],
response_format={"type": "json_object"},
provider_config={"api_version": "2024-10-21"},
structured_output_mode="explicit",
)
| Canonical | Legacy (deprecated) |
|---|---|
AZURE_ENDPOINT | AZURE_OPENAI_ENDPOINT, AZURE_AI_INFERENCE_ENDPOINT |
AZURE_API_KEY | AZURE_OPENAI_API_KEY, AZURE_AI_INFERENCE_API_KEY |
AZURE_API_VERSION | AZURE_OPENAI_API_VERSION |
Legacy vars emit DeprecationWarning. Use resolve_azure_env() from kaizen.nodes.ai.azure_detection for canonical-first resolution.
provider_config -- use response_formatstructured_output_mode="explicit"AZURE_BACKEND explicitlykaizen.core.prompt_utils is the single source of truth for signature-based prompt generation:
generate_prompt_from_signature(signature) -- builds system prompt from signature fieldsjson_prompt_suffix(output_fields) -- returns JSON format instructions for Azure json_object compatibilityFor detailed configuration patterns, see:
response_format for structured output (not provider_config)structured_output_mode="explicit" for new agentsprovider_configGovernedSupervisor with 3-layer progressive API (2-param simple -> 8-param configured -> 9 governance subsystems)AccountabilityTracker -- D/T/R addressing, policy source chainBudgetTracker -- reclamation, predictive warnings, reallocationCascadeManager -- monotonic envelope tightening, BFS terminationClearanceEnforcer + ClassificationAssigner -- data classification (C0-C4), regex pre-filterDerelictionDetector -- insufficient tightening detectionBypassManager -- time-limited emergency overrides with anti-stackingVacancyManager -- orphan detection, grandparent auto-designationAuditTrail -- EATP hash chain with hmac.compare_digest()EnvelopeAllocator -> EnvelopeSplitter, ScopeBridge -> ScopedContextkaizen-l3-overview -- L3 autonomy primitives, L3Runtime integration, EATP event system
L3Runtime convenience class wiring all 5 subsystems (Factory->Enforcer, Factory->Router, Factory->Context, Enforcer->Plan)L3EventBus pub/sub for 15 governance event types across all primitivesEatpTranslator converts L3 events into EATP audit records with severity classificationkaizen-agents-security -- Security patterns for governance
_ReadOnlyView proxiesmath.isfinite() on all numeric paths)Composition wrappers add cross-cutting concerns (governance, monitoring, streaming) around a BaseAgent without modifying it. WrapperBase enforces a canonical stacking order and duplicate detection.
Canonical stacking order (innermost to outermost):
BaseAgent -> L3GovernedAgent -> MonitoredAgent -> StreamingAgent
WrapperBase rejects duplicate wrappers (DuplicateWrapperError) and out-of-order stacking (WrapperOrderError). Every wrapper proxies get_parameters() and to_workflow() to the inner agent. The innermost property walks the full stack to the non-wrapper agent.
Key files:
src/kaizen_agents/wrapper_base.py) -- WrapperBase with stack ordering + duplicate detectionsrc/kaizen_agents/governed_agent.py) -- L3GovernedAgent with ConstraintEnvelope enforcement (Financial, Operational, Temporal, Data Access, Communication, Posture ceiling). Rejects BEFORE LLM cost is incurred. Uses _ProtectedInnerProxy to block governance bypass via .inner._inner.src/kaizen_agents/monitored_agent.py) -- MonitoredAgent with CostTracker, budget enforcement via BudgetExhaustedError, NaN/Inf defense on budget valuessrc/kaizen_agents/streaming_agent.py) -- StreamingAgent with run_stream() async iterator, typed StreamEvent events, buffer overflow protection, timeout enforcement. Falls back to batch when provider lacks StreamingProvider.src/kaizen_agents/events.py) -- Frozen dataclass events: TextDelta, ToolCallStart, ToolCallEnd, TurnComplete, BudgetExhausted, ErrorEvent, StreamBufferOverflowsrc/kaizen_agents/supervisor_wrapper.py) -- SupervisorWrapper for task delegation to worker pool via LLMBased routingBuilding a wrapper stack:
from kaizen.core.base_agent import BaseAgent
from kaizen_agents.governed_agent import L3GovernedAgent
from kaizen_agents.monitored_agent import MonitoredAgent
from kaizen_agents.streaming_agent import StreamingAgent
from kaizen_agents.events import TextDelta, TurnComplete
from kailash.trust.envelope import ConstraintEnvelope, FinancialConstraint
# Stack innermost to outermost
agent = MyAgent(config=config)
governed = L3GovernedAgent(agent, envelope=ConstraintEnvelope(
financial=FinancialConstraint(budget_limit=10.0)
))
monitored = MonitoredAgent(governed, budget_usd=5.0)
streaming = StreamingAgent(monitored)
# Stream typed events
async for event in streaming.run_stream(prompt="analyze this"):
match event:
case TextDelta(text=t): print(t, end="")
case TurnComplete(text=t): print(f"\n[Done: {t[:50]}]")
SupervisorWrapper -- delegates tasks to a worker pool using LLM-based routing:
from kaizen_agents.supervisor_wrapper import SupervisorWrapper
from kaizen_agents.patterns.llm_routing import LLMBased
supervisor = SupervisorWrapper(inner_agent, workers=[w1, w2], routing=LLMBased())
result = await supervisor.run_async(task="complex task")
SPEC-02 defines runtime_checkable protocols in kaizen.providers.base for structural capability discovery. Providers satisfy protocols structurally -- no explicit inheritance needed.
| Protocol | Key Method | Purpose |
|---|---|---|
StreamingProvider | stream_chat() -> StreamEvent | Token-by-token streaming |
ToolCallingProvider | chat_with_tools(messages, tools) | Native function calling |
StructuredOutputProvider | chat_structured(messages, schema) | JSON schema structured outputs |
AsyncLLMProvider | chat_async(messages) | Async chat completions |
ProviderCapability enum: CHAT_SYNC, CHAT_ASYNC, CHAT_STREAM, TOOLS, STRUCTURED_OUTPUT, EMBEDDINGS, VISION, AUDIO, REASONING_MODELS, BYOK.
Use get_provider_for_model(model) from kaizen.providers.registry to resolve a model string to a provider instance. Use isinstance(provider, StreamingProvider) for capability checks.
LLMBased from kaizen_agents.patterns.llm_routing scores agent capabilities against task requirements using Kaizen signatures (not keyword matching or dispatch tables).
from kaizen_agents.patterns.llm_routing import LLMBased
routing = LLMBased(config=config) # config optional; falls back to .env defaults
score = await routing.score("analyze revenue data", agent_capability)
best = await routing.select_best("analyze revenue data", [agent1, agent2, agent3])
score() returns [0.0, 1.0]. Accepts Capability dataclasses (.name + .description) or plain strings. select_best() returns the highest-scoring candidate or None when empty.
Three convergence SPECs have shipped on the feat/spec04-baseagent-slim branch:
SPEC-02 (Provider Split) -- The provider monolith (kaizen.nodes.ai.ai_providers) is now split into per-provider modules under kaizen/providers/. See kaizen-multi-provider for the updated registry, protocols, and CostTracker.
kaizen.providers.base -- ProviderCapability enum (10 members), 5 runtime-checkable protocolskaizen.providers.registry -- ProviderRegistry with 14 provider entries and prefix-dispatch model detectionkaizen.providers.cost -- CostTracker with thread-safe accumulationkaizen.nodes.ai.ai_providers re-exports all public namesSPEC-05 (Delegate Facade) -- Delegate is now a composition facade wrapping AgentLoop -> [L3GovernedAgent] -> [MonitoredAgent]. See kaizen-delegate for the updated API surface.
ConstructorIOError -- raised on outbound IO in __init__ToolRegistryCollisionError -- raised on duplicate tool name registrationrun_sync() refuses under a running event loop with an actionable error messagemcp_servers= stores configs, connects on first run().core_agent, .signature, .model read-only propertiesSPEC-10 (Multi-Agent) -- 11 deprecated agent subclasses (SupervisorAgent, WorkerAgent, CoordinatorAgent, PipelineStageAgent, etc.) now emit DeprecationWarning. Composition patterns accept plain BaseAgent instances. max_total_delegations cap (default 20) with DelegationCapExceeded exception.
For Kaizen-specific questions, invoke:
kaizen-specialist - Kaizen framework implementationtesting-specialist - Agent testing strategies skill - When to use Kaizen vs other frameworks