Menu
Kailash Kaizen (Python) — MANDATORY for AI agents/RAG/signatures. Custom LLM agents BLOCKED.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Baseado na classificação ocupacional SOC
Claude Code architecture — artifact design, context, agentic patterns. For CC audit/build.
Kailash Core SDK — workflows, 110+ nodes, runtime, async, cycles, MCP, OpenTelemetry. Use for WorkflowBuilder + connections + runtime patterns.
Kailash DataFlow — MANDATORY for DB/CRUD/bulk/migrations/multi-tenancy. Raw SQL/ORMs BLOCKED.
Kailash Nexus — MANDATORY for HTTP/API/CLI/MCP unified deployment. Direct FastAPI/Flask BLOCKED.
Kailash Kaizen (Python) — MANDATORY for AI agents/RAG/signatures. Custom LLM agents BLOCKED.
Kailash MCP — server/client/tools/resources/auth/transports for AI agent integration.
| 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