# Table of Contents - [Documentation | Docs | Kayba](#documentation-docs-kayba) - [How ACE Works | Docs | Kayba](#how-ace-works-docs-kayba) - [Claude Code | Docs | Kayba](#claude-code-docs-kayba) - [Insight Levels | Docs | Kayba](#insight-levels-docs-kayba) - [Integration Pattern | Docs | Kayba](#integration-pattern-docs-kayba) - [Async Learning | Docs | Kayba](#async-learning-docs-kayba) - [Quick Start | Docs | Kayba](#quick-start-docs-kayba) - [Prompt Engineering | Docs | Kayba](#prompt-engineering-docs-kayba) - [Update Operations | Docs | Kayba](#update-operations-docs-kayba) - [LangChain | Docs | Kayba](#langchain-docs-kayba) - [Installation | Docs | Kayba](#installation-docs-kayba) - [Branching & Parallelism | Docs | Kayba](#branching-parallelism-docs-kayba) - [Three Roles | Docs | Kayba](#three-roles-docs-kayba) - [Browser-Use | Docs | Kayba](#browser-use-docs-kayba) - [Opik Observability | Docs | Kayba](#opik-observability-docs-kayba) - [API Reference | Docs | Kayba](#api-reference-docs-kayba) - [Error Handling | Docs | Kayba](#error-handling-docs-kayba) - [Full Pipeline | Docs | Kayba](#full-pipeline-docs-kayba) - [Core Concepts | Docs | Kayba](#core-concepts-docs-kayba) - [Building Custom Steps | Docs | Kayba](#building-custom-steps-docs-kayba) - [The Skillbook | Docs | Kayba](#the-skillbook-docs-kayba) - [LiteLLM | Docs | Kayba](#litellm-docs-kayba) - [Testing | Docs | Kayba](#testing-docs-kayba) - [Overview | Docs | Kayba](#overview-docs-kayba) - [Quick Start | Docs | Kayba](#quick-start-docs-kayba) - [Execution Model | Docs | Kayba](#execution-model-docs-kayba) - [Documentation | Kayba](#documentation-kayba) - [API Reference | Docs | Kayba](#api-reference-docs-kayba) - [Overview | Docs | Kayba](#overview-docs-kayba) - [Documentation | Docs | Kayba](#documentation-docs-kayba) - [Documentation | Docs | Kayba](#documentation-docs-kayba) --- # Documentation | Docs | Kayba Page not found ============== This documentation page could not be found. [Back to docs](https://www.kayba.ai/docs) --- # How ACE Works | Docs | Kayba How ACE Works ============= **Agentic Context Engineering (ACE)** enables AI agents to learn from their own execution feedback. Instead of updating model weights (expensive, slow, opaque), ACE evolves a **skillbook** of strategies based on what actually works. > **Research** > > ACE was introduced in [_Agentic Context Engineering: Evolving Contexts for Self-Improving Language Models_](https://arxiv.org/abs/2510.04618) > by researchers at Stanford University and SambaNova Systems. The Learning Loop ----------------- Three collaborative roles share the same base LLM: 1. The **Agent** executes a task using strategies from the skillbook 2. The **Environment** evaluates the result (correct/incorrect, feedback) 3. The **Reflector** analyzes what worked and what failed 4. The **SkillManager** updates the skillbook with new strategies The **Skillbook** accumulates strategies across runs, making every subsequent agent call smarter. Three Roles ----------- | Role | Responsibility | Key Class | | --- | --- | --- | | **Agent** | Executes tasks using skillbook strategies | `Agent` | | **Reflector** | Analyzes execution results (what worked, what failed) | `Reflector` | | **SkillManager** | Transforms reflections into skillbook updates | `SkillManager` | All three roles use the same LLM — the intelligence comes from the specialized prompts each role receives. See [Three Roles](https://kayba.ai/docs/concepts/roles) for details on each role's inputs and outputs. Two Architecture Patterns ------------------------- ### Full ACE Pipeline Use when building a new agent from scratch. All three roles participate. The Agent produces answers, the Environment evaluates them, and the learning pipeline updates the skillbook. from ace_next import ACE, Agent, Reflector, SkillManager, LiteLLMClient, SimpleEnvironment llm = LiteLLMClient(model="gpt-4o-mini") runner = ACE.from_roles( agent=Agent(llm), reflector=Reflector(llm), skill_manager=SkillManager(llm), environment=SimpleEnvironment(), ) results = runner.run(samples, epochs=3) ### Integration Pattern Use when wrapping an existing agent (browser-use, LangChain, Claude Code). No ACE Agent — the external framework handles execution. ACE only learns from the results. Three steps: **INJECT** skillbook context, **EXECUTE** with external agent, **LEARN** from results. from ace_next import BrowserUse runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) results = runner.run("Find the top post on Hacker News") See [Integration Pattern](https://kayba.ai/docs/guides/integration) for building custom integrations. How It Compares --------------- | Approach | Updates | Speed | Interpretability | | --- | --- | --- | --- | | **Fine-tuning** | Model weights | Slow (hours) | Low (opaque) | | **RAG** | External documents | Medium | Medium | | **ACE** | Skillbook context | Fast (real-time) | High (readable strategies) | ACE strategies are human-readable, auditable, and transferable between models. Performance ----------- | Benchmark | Improvement | Notes | | --- | --- | --- | | AppWorld Agent | **+17.1 pp** | Complex multi-step tasks with tool use | | FiNER (Finance) | **+8.6 pp** | Financial reasoning tasks | | Adaptation Latency | **\-86.9%** | vs. existing context-adaptation methods | What to Read Next ----------------- * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — how strategies are stored and evolve * [Three Roles](https://kayba.ai/docs/concepts/roles) — Agent, Reflector, and SkillManager in detail * [Quick Start](https://kayba.ai/docs/getting-started/quick-start) — run your first agent --- # Claude Code | Docs | Kayba Claude Code Integration ======================= The `ClaudeCode` runner wraps the [Claude Code CLI](https://docs.anthropic.com/en/docs/claude-code) with ACE learning. The agent runs coding tasks in your project directory and learns strategies from each execution — improving code generation, debugging, and project-specific patterns over time. Quick Start ----------- from ace_next import ClaudeCode runner = ClaudeCode.from_model(working_dir="./my_project") results = runner.run("Add unit tests for utils.py") runner.save("coding_expert.json") Installation ------------ pip install ace-framework[claude-code] Prerequisites ------------- * Claude Code CLI installed and authenticated * A project directory with source code Parameters ---------- ### from\_model() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `working_dir` | `str` | `None` | Path to the project directory | | `ace_model` | `str` | `"gpt-4o-mini"` | Model for Reflector + SkillManager | | `ace_max_tokens` | `int` | `2048` | Max tokens for ACE LLM | | `ace_llm` | `LLMClientLike` | `None` | Pre-built LLM for ACE roles | ### from\_roles() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `reflector` | `ReflectorLike` | — | Reflector instance | | `skill_manager` | `SkillManagerLike` | — | SkillManager instance | | `working_dir` | `str` | `None` | Project directory | | `timeout` | `int` | `600` | Execution timeout (seconds) | | `model` | `str` | `None` | Claude model override | | `allowed_tools` | `list[str]` | `None` | Allowed Claude Code tools | | `skillbook_path` | `str` | `None` | Load saved skillbook | | `dedup_config` | `DeduplicationConfig` | `None` | Deduplication config | | `checkpoint_dir` | `str` | `None` | Checkpoint directory | Methods ------- results = runner.run(tasks, epochs=1) # Run with learning runner.save("path.json") # Save skillbook runner.wait_for_background() # Wait for async learning runner.get_strategies() # View learned strategies How It Works ------------ 1. **INJECT** — Skillbook strategies are written to `CLAUDE.md` at the project root 2. **EXECUTE** — Claude Code CLI runs the task in the project directory 3. **Extract trace** — ACE reads the Claude Code execution transcript 4. **LEARN** — Reflector analyzes the trace, SkillManager updates the skillbook The agent learns project-specific patterns like: * Code style and conventions * Common debugging approaches * Test patterns and frameworks used * Module structure and dependencies Running Multiple Tasks ---------------------- results = runner.run([\ "Add unit tests for utils.py",\ "Fix the bug in the login handler",\ "Refactor the database module to use connection pooling",\ ]) Resuming from a Saved Skillbook ------------------------------- runner = ClaudeCode.from_model( working_dir="./my_project", skillbook_path="coding_expert.json", ) What to Read Next ----------------- * [Integration Pattern](https://kayba.ai/docs/guides/integration) — how the INJECT/EXECUTE/LEARN pattern works * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — how learned strategies are stored --- # Insight Levels | Docs | Kayba Insight Levels ============== The ACE framework operates at three insight levels depending on what scope the Reflector analyzes. Overview -------- | Level | Reflector Scope | Feedback Source | Implementation | | --- | --- | --- | --- | | **Micro** | Single interaction | Environment (ground truth) | `ACE` runner with `TaskEnvironment` | | **Meso** | Full agent run | Execution trace (no ground truth) | Integration runners (`BrowserUse`, `LangChain`, `ClaudeCode`) | | **Macro** | Cross-run analysis | Pattern comparison across runs | Future enhancement | Micro-Level ----------- The Reflector receives the agent's output **and** environment feedback (ground truth, correctness). This is the most precise learning signal. Use when you have labeled data or a reliable evaluation function. from ace_next import ACE, Sample, SimpleEnvironment runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=SimpleEnvironment(), ) samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ ] runner.run(samples, epochs=3) Meso-Level ---------- The Reflector receives the full **execution trace** — the agent's reasoning steps, tool calls, actions, and outcomes — but no external ground truth. It learns from execution patterns rather than correctness evaluation. Use when wrapping external agents where you don't have labeled answers. from ace_next import BrowserUse # The browser-use agent produces a rich trace of actions runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) runner.run("Find the top post on Hacker News") The extracted trace includes: * Agent reasoning at each step * Browser actions (click, type, navigate) * Page observations * Success/failure of each action Macro-Level ----------- Cross-run pattern analysis — comparing strategies across multiple execution histories. Not yet implemented. What to Read Next ----------------- * [Three Roles](https://kayba.ai/docs/concepts/roles) — the roles involved at each level * [Integration Pattern](https://kayba.ai/docs/guides/integration) — meso-level integrations in practice * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — micro-level pipelines in practice --- # Integration Pattern | Docs | Kayba Integration Pattern =================== Use the integration pattern when you have an **existing agent** (browser-use, LangChain, Claude Code, or a custom framework) and want to add ACE learning on top. > **Full Pipeline vs Integration** > > The [Full Pipeline](https://kayba.ai/docs/guides/full-pipeline) > uses all three ACE roles. The integration pattern skips the ACE Agent — your external agent handles execution, and ACE only learns from the results. Three Steps ----------- Every integration follows the same pattern: 1. INJECT — Add skillbook strategies to the agent's context 2. EXECUTE — Run the external agent normally 3. LEARN — Reflector + SkillManager update the skillbook Using Built-In Runners ---------------------- ACE provides runners for popular frameworks. Each uses `from_model()` for quick setup or `from_roles()` for full control: ### Browser-Use ```python from ace_next import BrowserUse from langchain_openai import ChatOpenAI runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) results = runner.run(["Find top HN post", "Check weather in NYC"]) runner.save("browser_expert.json") ``` ### LangChain ```python from ace_next import LangChain runner = LangChain.from_model(your_chain, ace_model="gpt-4o-mini") results = runner.run([{"input": "Summarize this document"}]) runner.save("chain_expert.json") ``` ### Claude Code ```python from ace_next import ClaudeCode runner = ClaudeCode.from_model(working_dir="./my_project") results = runner.run(["Add tests for utils.py", "Fix the login bug"]) runner.save("code_expert.json") ``` Construction Patterns --------------------- All integration runners offer two construction paths: ### from\_model() — Quick Setup Builds ACE roles automatically from a model string: runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", # Model for Reflector + SkillManager skillbook_path="saved.json", # Optional: resume from saved skillbook ) ### from\_roles() — Full Control Provide pre-built role instances: from ace_next import Reflector, SkillManager, LiteLLMClient learning_llm = LiteLLMClient(model="gpt-4o-mini") runner = BrowserUse.from_roles( browser_llm=ChatOpenAI(model="gpt-4o"), reflector=Reflector(learning_llm), skill_manager=SkillManager(learning_llm), skillbook_path="saved.json", dedup_config=my_dedup_config, checkpoint_dir="./checkpoints", ) Common Options -------------- All integration runners share these parameters: | Parameter | Description | Default | | --- | --- | --- | | `skillbook` | Existing `Skillbook` instance | `None` (creates empty) | | `skillbook_path` | Path to load skillbook from | `None` | | `dedup_config` | Deduplication configuration | `None` | | `dedup_interval` | Samples between dedup runs | `10` | | `checkpoint_dir` | Directory for checkpoint files | `None` | | `checkpoint_interval` | Samples between checkpoints | `10` | Lifecycle Methods ----------------- All runners expose: runner.save("path.json") # Save skillbook runner.wait_for_background() # Wait for async learning runner.learning_stats # Background progress dict runner.skillbook # Current Skillbook instance runner.get_strategies() # Formatted strategies string Building a Custom Integration ----------------------------- For frameworks not covered by the built-in runners, you can compose a custom pipeline using steps. The pattern: **Execute Step** (runs your agent) + **ToTrace Step** (extracts learning signal) + **learning\_tail()** (standard learning pipeline). from pipeline import Pipeline from ace_next import Skillbook, Reflector, SkillManager, LiteLLMClient from ace_next.steps import learning_tail from ace_next.runners import ACERunner # Your custom execute step would implement the Step protocol # See the Pipeline Engine docs for details on building custom steps learning_llm = LiteLLMClient(model="gpt-4o-mini") skillbook = Skillbook() steps = [\ MyCustomExecuteStep(...),\ MyCustomToTraceStep(),\ *learning_tail(\ Reflector(learning_llm),\ SkillManager(learning_llm),\ skillbook,\ ),\ ] runner = ACERunner(pipeline=Pipeline(steps), skillbook=skillbook) See [Pipeline Engine: Building Custom Steps](https://kayba.ai/docs/pipeline/custom-steps) for the Step protocol. What to Read Next ----------------- * [LiteLLM Integration](https://kayba.ai/docs/integrations/litellm) — simplest self-improving agent * [Browser-Use Integration](https://kayba.ai/docs/integrations/browser-use) — browser automation details * [LangChain Integration](https://kayba.ai/docs/integrations/langchain) — chain/agent wrapping * [Claude Code Integration](https://kayba.ai/docs/integrations/claude-code) — coding tasks --- # Async Learning | Docs | Kayba Async Learning ============== By default, learning (Reflect, Tag, Update, Apply) runs synchronously after each sample. With async learning, the Agent returns immediately while learning continues in the background. Architecture ------------ * **Reflectors** run concurrently (safe — they only read the skillbook) * **SkillManager** runs sequentially (required — it writes to the skillbook) * The Agent uses whatever skillbook state is available (eventual consistency) Basic Usage ----------- Pass `wait=False` to `run()`: from ace_next import ACE runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=environment, ) # Agent returns fast — learning continues in background results = runner.run(samples, epochs=3, wait=False) # Use results immediately for r in results: print(r) # Wait before saving runner.wait_for_background() runner.save("learned.json") Monitoring Progress ------------------- stats = runner.learning_stats # {'active': 5, 'completed': 25} With ACELiteLLM --------------- from ace_next import ACELiteLLM, Sample, SimpleEnvironment agent = ACELiteLLM.from_model("gpt-4o-mini") samples = [Sample(question="...", context="", ground_truth="...")] results = agent.learn(samples, environment=SimpleEnvironment(), wait=False) # Agent is immediately available answer = agent.ask("New question") # Wait when you need to save agent.wait_for_background() agent.save("learned.json") Why This Architecture --------------------- | Component | Parallelizable? | Reason | | --- | --- | --- | | Reflector | Yes | Only reads the skillbook, produces independent analysis | | SkillManager | No | Writes to the skillbook, handles deduplication | This gives ~3x faster learning when the Reflector LLM calls run concurrently. What to Read Next ----------------- * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — synchronous pipeline setup * [Testing](https://kayba.ai/docs/guides/testing) — test async learning with MagicMock --- # Quick Start | Docs | Kayba Quick Start =========== Get a self-learning agent running in under a minute. Simplest Example ---------------- from ace_next import ACELiteLLM agent = ACELiteLLM.from_model("gpt-4o-mini") # Ask related questions — the agent learns patterns across them answer1 = agent.ask("If all cats are animals, is Felix (a cat) an animal?") answer2 = agent.ask("If all birds fly, can penguins (birds) fly?") print(f"Learned {len(agent.skillbook.skills())} strategies") # Save and reload later agent.save("my_agent.json") Choose Your Integration ----------------------- ### LiteLLM The simplest path. Supports 100+ LLM providers. ```python from ace_next import ACELiteLLM agent = ACELiteLLM.from_model("gpt-4o-mini") answer = agent.ask("Your question") agent.save("learned.json") ``` ### LangChain Wrap any LangChain Runnable (chains, agents, graphs) with learning. ```python from ace_next import LangChain runner = LangChain.from_model(your_chain, ace_model="gpt-4o-mini") results = runner.run([{"input": "Your task"}]) runner.save("chain_expert.json") ``` ### Browser-Use Browser automation that learns navigation patterns. ```python from ace_next import BrowserUse from langchain_openai import ChatOpenAI runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) results = runner.run("Find the top post on Hacker News") runner.save("browser_expert.json") ``` ### Claude Code Self-improving coding agent using the Claude Code CLI. ```python from ace_next import ClaudeCode runner = ClaudeCode.from_model(working_dir="./my_project") results = runner.run("Add unit tests for utils.py") runner.save("coding_expert.json") ``` Full Pipeline Example --------------------- For full control, use the three ACE roles directly: from ace_next import ( ACE, Agent, Reflector, SkillManager, LiteLLMClient, Sample, SimpleEnvironment, ) # Create LLM and roles llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(llm) reflector = Reflector(llm) skill_manager = SkillManager(llm) # Build the adaptive pipeline runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=SimpleEnvironment(), ) # Train on samples samples = [\ Sample(question="What is the capital of France?", context="", ground_truth="Paris"),\ Sample(question="What is 2 + 2?", context="", ground_truth="4"),\ ] results = runner.run(samples, epochs=2) print(f"Learned {len(runner.skillbook.skills())} strategies") runner.save("trained.json") Loading Saved Agents -------------------- from ace_next import ACELiteLLM # Resume from a saved skillbook agent = ACELiteLLM.from_model("gpt-4o-mini", skillbook_path="my_agent.json") answer = agent.ask("New question") # Uses previously learned strategies Trying Different Models ----------------------- from ace_next import ACELiteLLM # OpenAI agent = ACELiteLLM.from_model("gpt-4o-mini") # Anthropic agent = ACELiteLLM.from_model("claude-sonnet-4-5-20250929") # Google agent = ACELiteLLM.from_model("gemini-pro") # Local (Ollama) agent = ACELiteLLM.from_model("ollama/llama2") What to Read Next ----------------- * [How ACE Works](https://kayba.ai/docs/concepts/overview) — understand the three-role architecture * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — how strategies are stored and evolve * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — build custom ACE pipelines * [Integrations](https://kayba.ai/docs/integrations/index) — LangChain, Browser-Use, Claude Code --- # Prompt Engineering | Docs | Kayba Prompt Engineering ================== ACE uses specialized prompt templates for each role. The framework includes multiple prompt versions with different trade-offs. Default Prompts --------------- `ace_next` ships with v2.1 prompts built in. All three roles (`Agent`, `Reflector`, `SkillManager`) use them by default — no extra imports needed. > **Recommendation** > > The built-in v2.1 prompts work well out of the box. Only provide custom prompts when you need domain-specific instructions. Overriding Prompts ------------------ Pass a `prompt_template` string to any role constructor: from ace_next import Agent, Reflector, SkillManager, LiteLLMClient llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(llm, prompt_template="Your custom agent prompt ...") reflector = Reflector(llm, prompt_template="Your custom reflector prompt ...") skill_manager = SkillManager(llm, prompt_template="Your custom skill manager prompt ...") Template Variables ------------------ ### Agent Prompt | Variable | Description | | --- | --- | | `{skillbook}` | Current skillbook in TOON format | | `{question}` | The input question | | `{context}` | Additional context | | `{reflection}` | Optional reflection from a previous attempt | ### Reflector Prompt | Variable | Description | | --- | --- | | `{skillbook}` | Current skillbook in TOON format | | `{question}` | The original question | | `{agent_output}` | The agent's response | | `{ground_truth}` | Expected answer | | `{feedback}` | Environment feedback | ### SkillManager Prompt | Variable | Description | | --- | --- | | `{skillbook}` | Current skillbook in TOON format | | `{reflection}` | Reflector's analysis | | `{question_context}` | Description of the task domain | | `{progress}` | Current training progress | Custom Prompts -------------- You can provide your own prompt templates. They must include the required template variables: custom_agent_prompt = """ Skillbook: {skillbook} Question: {question} Context: {context} Generate a JSON response with: - reasoning: Your step-by-step thought process - skill_ids: List of skillbook IDs you used - final_answer: Your answer """ agent = Agent(llm, prompt_template=custom_agent_prompt) Formatting Skillbook for External Agents ---------------------------------------- Integration runners inject the skillbook into external agent prompts using a wrapper function: from ace_next import wrap_skillbook_context context = wrap_skillbook_context(skillbook) # Returns formatted strategies with usage instructions Troubleshooting --------------- | Problem | Solution | | --- | --- | | JSON parse failures | Increase `max_tokens`, use Instructor, or try v2.1 prompts | | Empty skill\_ids | Agent not citing skills — check skillbook has content | | Poor answer quality | Switch to v2.1 prompts or try a larger model | What to Read Next ----------------- * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — use prompts in a complete pipeline * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — what goes into `{skillbook}` --- # Update Operations | Docs | Kayba Update Operations ================= The SkillManager communicates changes to the skillbook through **update operations**. Each operation is a structured instruction to modify the skillbook in a specific way. Operation Types --------------- | Type | Description | Required Fields | | --- | --- | --- | | `ADD` | Create a new skill | `section`, `content` | | `UPDATE` | Modify an existing skill's content | `skill_id`, `content` | | `TAG` | Record a skill as helpful, harmful, or neutral | `skill_id`, `tag` | | `REMOVE` | Delete a skill from the skillbook | `skill_id` | Examples -------- ### ADD Adds a new strategy learned from experience: { "type": "ADD", "section": "Math Strategies", "content": "Break complex problems into smaller steps before computing" } ### UPDATE Refines an existing strategy: { "type": "UPDATE", "skill_id": "math-00001", "content": "Break complex problems into smaller steps. Verify each step before proceeding." } ### TAG Records whether a strategy helped or hurt: { "type": "TAG", "skill_id": "math-00001", "tag": "helpful" } Tags are one of: `helpful`, `harmful`, `neutral`. ### REMOVE Prunes a strategy that is consistently harmful: { "type": "REMOVE", "skill_id": "math-00003" } Update Batches -------------- The SkillManager emits operations as an `UpdateBatch` — one or more operations applied atomically: from ace_next import UpdateOperation, UpdateBatch batch = UpdateBatch(operations=[\ UpdateOperation(type="ADD", section="Debugging", content="Log inputs before errors"),\ UpdateOperation(type="TAG", skill_id="debug-00001", tag="helpful"),\ ]) skillbook.apply_update(batch) How Updates Flow ---------------- Agent cites skill_ids --> Reflector tags them --> SkillManager emits ADD/UPDATE/REMOVE 1. The **Agent** cites skill IDs it used in its reasoning 2. The **Reflector** classifies each cited skill as helpful/harmful/neutral (TAG operations) 3. The **SkillManager** may also ADD new strategies or UPDATE/REMOVE existing ones based on the reflection What to Read Next ----------------- * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — where operations are applied * [Three Roles](https://kayba.ai/docs/concepts/roles) — which role emits which operations --- # LangChain | Docs | Kayba LangChain Integration ===================== The `LangChain` runner wraps any LangChain Runnable (chains, `AgentExecutor`, LangGraph graphs) with ACE learning. The runner extracts execution traces and learns strategies from them. Installation ------------ pip install ace-framework[langchain] Quick Start ----------- from ace_next import LangChain runner = LangChain.from_model(your_chain, ace_model="gpt-4o-mini") results = runner.run([\ {"input": "Summarize this document"},\ {"input": "Extract key entities"},\ ]) runner.save("chain_expert.json") Parameters ---------- ### from\_model() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `runnable` | `Any` | — | LangChain Runnable (chain, AgentExecutor, graph) | | `ace_model` | `str` | `"gpt-4o-mini"` | Model for Reflector + SkillManager | | `ace_max_tokens` | `int` | `2048` | Max tokens for ACE LLM | | `ace_llm` | `LLMClientLike` | `None` | Pre-built LLM for ACE roles | ### from\_roles() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `runnable` | `Any` | — | LangChain Runnable | | `reflector` | `ReflectorLike` | — | Reflector instance | | `skill_manager` | `SkillManagerLike` | — | SkillManager instance | | `skillbook_path` | `str` | `None` | Load saved skillbook | | `output_parser` | `Callable` | `None` | Custom output extraction | | `dedup_config` | `DeduplicationConfig` | `None` | Deduplication config | | `checkpoint_dir` | `str` | `None` | Checkpoint directory | Methods ------- results = runner.run(inputs, epochs=1) # Run with learning results = runner.invoke(single_input) # Single input convenience runner.save("path.json") # Save skillbook runner.wait_for_background() # Wait for async learning How It Works ------------ 1. **INJECT** — Skillbook strategies are added to the chain input 2. **EXECUTE** — LangChain runs the chain normally 3. **Extract trace** — ACE extracts intermediate steps, tool calls, and reasoning 4. **LEARN** — Reflector analyzes the trace, SkillManager updates the skillbook The runner handles simple chains, `AgentExecutor` (with `intermediate_steps`), and LangGraph graphs automatically. Input Types ----------- The runner accepts any input your chain expects: # String input runner.run(["What is ACE?"]) # Dict input runner.run([{"input": "query", "context": "..."}]) # Message list runner.run([[HumanMessage(content="Hello")]]) Resuming from a Saved Skillbook ------------------------------- runner = LangChain.from_model( your_chain, ace_model="gpt-4o-mini", skillbook_path="chain_expert.json", ) What to Read Next ----------------- * [Integration Pattern](https://kayba.ai/docs/guides/integration) — how the INJECT/EXECUTE/LEARN pattern works * [Insight Levels](https://kayba.ai/docs/concepts/insight-levels) — meso-level learning from traces * [Opik Observability](https://kayba.ai/docs/integrations/opik) — monitor chain execution costs --- # Installation | Docs | Kayba Installation ============ For Users --------- ### pip ```bash pip install ace-framework ``` ### With extras ```bash pip install ace-framework[all] # All optional features pip install ace-framework[instructor] # Structured outputs (Instructor) pip install ace-framework[langchain] # LangChain integration pip install ace-framework[browser-use] # Browser automation pip install ace-framework[claude-code] # Claude Code CLI integration pip install ace-framework[observability] # Opik monitoring + cost tracking pip install ace-framework[deduplication] # Skill deduplication (embeddings) pip install ace-framework[transformers] # Local model support ``` For Contributors ---------------- ### UV (Recommended) ```bash git clone https://github.com/kayba-ai/agentic-context-engine cd agentic-context-engine uv sync # Installs everything (10-100x faster than pip) ``` ### pip ```bash git clone https://github.com/kayba-ai/agentic-context-engine cd agentic-context-engine pip install -e . ``` Requirements ------------ * **Python 3.12** * An API key for your LLM provider API Key Setup ------------- Set one of these environment variables depending on your provider: export OPENAI_API_KEY="sk-..." export ANTHROPIC_API_KEY="sk-ant-..." export GOOGLE_API_KEY="AIza..." Or create a `.env` file: echo "OPENAI_API_KEY=sk-..." > .env from dotenv import load_dotenv load_dotenv() # Loads from .env Supported Providers ------------------- ACE uses [LiteLLM](https://docs.litellm.ai/) for model access, supporting 100+ providers: | Provider | Model Example | Env Variable | | --- | --- | --- | | OpenAI | `gpt-4o-mini` | `OPENAI_API_KEY` | | Anthropic | `claude-sonnet-4-5-20250929` | `ANTHROPIC_API_KEY` | | Google | `gemini-pro` | `GOOGLE_API_KEY` | | Ollama (local) | `ollama/llama2` | — | | AWS Bedrock | `bedrock/anthropic.claude-v2` | AWS credentials | | Azure | `azure/gpt-4` | `AZURE_API_KEY` | Verify Installation ------------------- from ace_next import ACELiteLLM agent = ACELiteLLM.from_model("gpt-4o-mini") print(agent.ask("Hello!")) What to Read Next ----------------- * [Quick Start](https://kayba.ai/docs/getting-started/quick-start) — build your first self-learning agent * [How ACE Works](https://kayba.ai/docs/concepts/overview) — understand the architecture --- # Branching & Parallelism | Docs | Kayba Branching & Parallelism ======================= A `Branch` runs multiple pipelines in parallel on the same input, then merges their outputs before the next step. It is itself a step — it satisfies `StepProtocol` and can be used anywhere a step is expected. * * * What is a Branch? ----------------- The branch forks the context to both child pipelines, runs them in parallel, and joins (merges) the results before `Summarize` runs. The join is implicit — any step after a `Branch` waits for all branches to complete. * * * Creating branches ----------------- Two equivalent APIs: ### Fluent shorthand ```python pipe = ( Pipeline() .then(Tokenize()) .branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=MergeStrategy.RAISE_ON_CONFLICT, ) .then(Summarize()) ) ``` ### Explicit Branch ```python from pipeline import Branch, MergeStrategy branch_step = Branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=MergeStrategy.RAISE_ON_CONFLICT, ) pipe = ( Pipeline() .then(Tokenize()) .then(branch_step) .then(Summarize()) ) ``` * * * Merge strategies ---------------- When all branches complete, their output contexts must be merged back into one. The `merge` parameter controls how conflicts are resolved. ### `RAISE_ON_CONFLICT` (default) Raises `ValueError` if two branches write different values to the same named field. Disjoint fields pass through without conflict. pipe = ( Pipeline() .then(Tokenize()) .branch( Pipeline().then(Uppercase()), # provides: {"upper_tokens"} Pipeline().then(Reverse()), # provides: {"reversed_tokens"} merge=MergeStrategy.RAISE_ON_CONFLICT, ) ) # Works — branches write to different fields results = pipe.run([StepContext(sample="hello world")]) > **Tip** > > In practice, branches that write disjoint fields (which is the common case) never conflict and the merge is a no-op. ### `LAST_WRITE_WINS` The last branch's value wins for every conflicting field. Simple but lossy. pipe = Pipeline().then(Tokenize()).branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=MergeStrategy.LAST_WRITE_WINS, ) ### `NAMESPACED` Each branch's output is stored at `metadata["branch_0"]`, `metadata["branch_1"]`, etc. No conflict is possible. pipe = Pipeline().then(Tokenize()).branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=MergeStrategy.NAMESPACED, ) results = pipe.run([StepContext(sample="hello world")]) meta = results[0].output.metadata print(meta["branch_0"]) # context from Uppercase branch print(meta["branch_1"]) # context from Reverse branch ### Custom merge function For full control, pass a callable: def priority_merge(ctxs: list[StepContext]) -> StepContext: """First branch wins for all fields.""" base = ctxs[0] merged_meta = dict(base.metadata) for ctx in ctxs[1:]: for k, v in ctx.metadata.items(): merged_meta.setdefault(k, v) # first writer wins return base.replace(metadata=MappingProxyType(merged_meta)) pipe = Pipeline().then(Tokenize()).branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=priority_merge, ) * * * How context flows through branches ---------------------------------- 1. All branches receive the **same frozen context** — no copy needed since `StepContext` is immutable 2. Each branch runs its pipeline and returns a **new** context 3. The merge function receives the list of output contexts and returns a single merged context 4. The merged context is passed to the next step in the outer pipeline Immutability is what makes this safe. No branch can corrupt another branch's input. * * * Execution model --------------- ### Sync Branches run in a `ThreadPoolExecutor` with `max_workers=len(pipelines)`: ```python # All branches get their own thread with ThreadPoolExecutor(max_workers=len(self.pipelines)) as executor: futures = [executor.submit(p, ctx) for p in self.pipelines] results = [f.result() for f in futures] return self._merge_fn(results) ``` ### Async Branches run via `asyncio.gather`. Sync child pipelines are wrapped with `asyncio.to_thread`: ```python results = await asyncio.gather( *[run_child(p) for p in self.pipelines], return_exceptions=True, ) return self._merge_fn(results) ``` In both cases, all branches run to completion even if one fails. * * * Error handling in branches -------------------------- When one or more branches fail, a `BranchError` is raised — but only after **all** branches have completed: from pipeline.errors import BranchError try: results = pipe.run(contexts) except BranchError as e: print(f"{len(e.failures)} branch(es) failed") for failure in e.failures: print(f" - {type(failure).__name__}: {failure}") * `BranchError.failures` contains one exception per failed branch * Successful branches are not lost — their contexts are still available * `SampleResult.failed_at` is set to `"Branch"` and `SampleResult.cause` carries the inner exception See [Error Handling](https://kayba.ai/docs/pipeline/error-handling) for the full error model. * * * Contract inference ------------------ A `Branch` computes its own `requires` and `provides` from the union of its children: branch = Branch( Pipeline().then(Uppercase()), # requires: {"tokens"}, provides: {"upper_tokens"} Pipeline().then(Reverse()), # requires: {"tokens"}, provides: {"reversed_tokens"} ) # Inferred: # branch.requires = {"tokens"} # branch.provides = {"upper_tokens", "reversed_tokens"} The outer pipeline validates against these aggregated contracts at construction time, so nesting branches inside pipelines works seamlessly. --- # Three Roles | Docs | Kayba Three Roles =========== ACE uses three collaborative roles that share the same base LLM. Each role has a specialized prompt that focuses it on a specific part of the learning loop. Agent ----- **Produces answers** using the current skillbook. The Agent receives a question, context, and the skillbook's strategies, then generates a reasoned answer citing which skills it used. from ace_next import Agent, LiteLLMClient llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(llm) output = agent.generate( question="What is 2+2?", context="Show your work", skillbook=skillbook, reflection=None, # Optional: reflection from a previous attempt ) ### AgentOutput | Field | Type | Description | | --- | --- | --- | | `final_answer` | `str` | The generated answer | | `reasoning` | `str` | Step-by-step reasoning | | `skill_ids` | `List[str]` | Skillbook strategies cited | | `raw` | `Dict` | Raw LLM response | Reflector --------- **Analyzes execution outcomes** — what worked, what failed, and why. The Reflector receives the agent's output, the environment's feedback, and the skillbook. It produces an analysis of the outcome and tags each cited skill as helpful, harmful, or neutral. from ace_next import Reflector reflector = Reflector(llm) reflection = reflector.reflect( question="What is 2+2?", agent_output=output, skillbook=skillbook, ground_truth="4", feedback="Correct!", ) ### ReflectorOutput | Field | Type | Description | | --- | --- | --- | | `reasoning` | `str` | Analysis of the outcome | | `error_identification` | `str` | What went wrong (if anything) | | `root_cause_analysis` | `str` | Why it went wrong | | `correct_approach` | `str` | What should have been done | | `key_insight` | `str` | Main lesson learned | | `skill_tags` | `List[SkillTag]` | `(skill_id, tag)` pairs | ### Reflector Modes | Mode | Description | | --- | --- | | `SIMPLE` | Single-pass analysis (default) | | `RECURSIVE` | Multi-pass with code execution in a REPL loop | SkillManager ------------ **Transforms reflections into skillbook updates.** The SkillManager takes the Reflector's analysis and decides which operations to apply to the skillbook — adding new strategies, updating existing ones, or removing harmful ones. from ace_next import SkillManager skill_manager = SkillManager(llm) sm_output = skill_manager.update_skills( reflection=reflection, skillbook=skillbook, question_context="Math problems", progress="3/5 correct", ) # Apply the updates skillbook.apply_update(sm_output.update) ### SkillManagerOutput | Field | Type | Description | | --- | --- | --- | | `update` | `UpdateBatch` | Batch of update operations to apply | | `consolidation_ops` | `List` | Deduplication operations (if enabled) | Shared LLM ---------- All three roles share the same LLM instance. The intelligence comes from the specialized prompts, not from using different models: from ace_next import Agent, Reflector, SkillManager, LiteLLMClient llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(llm) # Same LLM reflector = Reflector(llm) # Same LLM skill_manager = SkillManager(llm) # Same LLM You can optionally use a cheaper model for the learning roles (Reflector + SkillManager) while keeping a stronger model for the Agent: agent_llm = LiteLLMClient(model="gpt-4o") learning_llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(agent_llm) reflector = Reflector(learning_llm) skill_manager = SkillManager(learning_llm) What to Read Next ----------------- * [Insight Levels](https://kayba.ai/docs/concepts/insight-levels) — micro, meso, and macro analysis scopes * [Update Operations](https://kayba.ai/docs/concepts/updates) — the operations the SkillManager emits * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — wire the roles together --- # Browser-Use | Docs | Kayba Browser-Use Integration ======================= The `BrowserUse` runner wraps [browser-use](https://github.com/browser-use/browser-use) with ACE learning. The agent automates browser tasks and learns strategies from each run — improving navigation, element selection, and error recovery over time. Installation ------------ pip install ace-framework[browser-use] Quick Start ----------- from ace_next import BrowserUse from langchain_openai import ChatOpenAI runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) results = runner.run("Find the top post on Hacker News") runner.save("browser_expert.json") Parameters ---------- ### from\_model() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `browser_llm` | `Any` | — | LLM for browser-use execution | | `ace_model` | `str` | `"gpt-4o-mini"` | Model for Reflector + SkillManager | | `ace_max_tokens` | `int` | `2048` | Max tokens for ACE LLM | | `ace_llm` | `LLMClientLike` | `None` | Pre-built LLM for ACE roles | ### from\_roles() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `browser_llm` | `Any` | — | LLM for browser-use execution | | `reflector` | `ReflectorLike` | — | Reflector instance | | `skill_manager` | `SkillManagerLike` | — | SkillManager instance | | `skillbook_path` | `str` | `None` | Load saved skillbook | | `browser` | `Browser` | `None` | browser-use Browser instance | | `agent_kwargs` | `dict` | `None` | Extra kwargs for browser-use Agent | | `dedup_config` | `DeduplicationConfig` | `None` | Deduplication config | | `checkpoint_dir` | `str` | `None` | Checkpoint directory | Methods ------- results = runner.run(tasks, epochs=1) # Run with learning runner.save("path.json") # Save skillbook runner.wait_for_background() # Wait for async learning runner.get_strategies() # View learned strategies How It Works ------------ 1. **INJECT** — Skillbook strategies are added to the task prompt 2. **EXECUTE** — browser-use runs the task (navigation, clicks, form fills) 3. **Extract trace** — ACE extracts a chronological trace of agent thoughts, actions, and results 4. **LEARN** — Reflector analyzes the full trace, SkillManager updates the skillbook The extracted trace includes: * Agent reasoning at each step * Browser actions taken (click, type, navigate) * Page observations * Success/failure of each action Running Multiple Tasks ---------------------- results = runner.run([\ "Find the top post on Hacker News",\ "Search for ACE framework on GitHub",\ "Check the weather in NYC",\ ]) Example: Domain Checker ----------------------- from ace_next import BrowserUse from langchain_openai import ChatOpenAI runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", ) domains = ["example.com", "test.org", "sample.net"] for domain in domains: runner.run(f"Check if {domain} is available for registration") # After several runs, the agent learns: # - Which registrar sites to use # - How to navigate the domain search UI # - How to interpret availability results runner.save("domain_checker.json") Resuming from a Saved Skillbook ------------------------------- runner = BrowserUse.from_model( browser_llm=ChatOpenAI(model="gpt-4o"), ace_model="gpt-4o-mini", skillbook_path="browser_expert.json", ) What to Read Next ----------------- * [Integration Pattern](https://kayba.ai/docs/guides/integration) — how the INJECT/EXECUTE/LEARN pattern works * [The Skillbook](https://kayba.ai/docs/concepts/skillbook) — how learned strategies are stored * [Opik Observability](https://kayba.ai/docs/integrations/opik) — monitor browser automation costs --- # Opik Observability | Docs | Kayba Opik Observability ================== ACE integrates with [Opik](https://github.com/comet-ml/opik) for tracing, cost tracking, and performance monitoring. All Opik tracing is **explicit opt-in** — it is never auto-enabled just because the package is installed. Two independent tracing modes: 1. **Pipeline step** (`OpikStep`) — client-agnostic, logs one Opik trace per sample with ACE context fields. 2. **LiteLLM callback** (`register_opik_litellm_callback`) — LiteLLM-specific, tracks per-LLM-call tokens and costs. Installation ------------ pip install ace-framework[observability] Quick Start ----------- from ace_next import ACELiteLLM # Easiest: ACELiteLLM enables both tracing modes with one flag ace = ACELiteLLM.from_model("gpt-4o-mini", opik=True, opik_project="my-experiment") from ace_next import ( ACE, OpikStep, Agent, Reflector, SkillManager, LiteLLMClient, SimpleEnvironment, ) # Manual: Add OpikStep via extra_steps client = LiteLLMClient(model="gpt-4o-mini") runner = ACE.from_roles( agent=Agent(client), reflector=Reflector(client), skill_manager=SkillManager(client), environment=SimpleEnvironment(), extra_steps=[OpikStep(project_name="my-experiment")], ) # LLM-level cost tracking only (no pipeline traces) from ace_next import register_opik_litellm_callback registered = register_opik_litellm_callback(project_name="my-experiment") Starting the Opik Server ------------------------ ### Local (Docker) ```bash docker run -d -p 5173:5173 --name opik ghcr.io/comet-ml/opik:latest # View traces at http://localhost:5173 ``` ### Comet Cloud ```bash export COMET_API_KEY="your-api-key" # Traces appear at https://www.comet.com/opik ``` OpikStep -------- `OpikStep` is a terminal side-effect step that logs one Opik trace per sample. It reads context fields but never mutates them — safe to append to any pipeline. ### Parameters | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `project_name` | `str` | `"ace-framework"` | Opik project for organizing traces | | `tags` | `list[str]` | `None` | Extra tags attached to every trace | ### What Gets Logged Each trace includes: | Field | Source | | --- | --- | | **Input** | Question and context from the sample | | **Output** | Answer, reasoning, and skill IDs from `AgentOutput` | | **Metadata** | Epoch, step index, skill count, reflection insights, operation counts | | **Feedback scores** | Accuracy extracted from environment feedback (correct / incorrect) | ### Trace Hierarchy LLM Cost Tracking ----------------- `OpikStep` does **not** register the LiteLLM callback — the two tracing modes are independent. To get per-LLM-call cost tracking, call `register_opik_litellm_callback()` separately: from ace_next import register_opik_litellm_callback success = register_opik_litellm_callback(project_name="cost-tracking") # Returns True if registered, False if Opik unavailable Every LLM call is then automatically tracked with: * Input / output tokens * Model used * Cost per call * Latency When using `ACELiteLLM` with `opik=True`, both modes are enabled together automatically — no need to call `register_opik_litellm_callback()` manually. Environment Variables --------------------- | Variable | Description | Default | | --- | --- | --- | | `OPIK_PROJECT_NAME` | Project name for organizing traces | `ace-framework` | | `OPIK_DISABLED=true` | Disable all Opik tracing | Not set | | `OPIK_ENABLED=false` | Alternative way to disable tracing | Not set | | `OPIK_URL_OVERRIDE` | Custom Opik server URL | `http://localhost:5173/api` | | `OPIK_WORKSPACE` | Opik workspace name | `default` | Error Handling -------------- When using `ACELiteLLM` with `opik=True`, errors are **raised immediately**: * `ImportError` if the `opik` package is not installed * `RuntimeError` if the Opik client fails to initialize (bad config, disabled via env vars) This ensures you know immediately if tracing is broken, rather than discovering missing traces later. When using `OpikStep` directly via `extra_steps`, it soft-imports Opik and silently becomes a no-op if the package is absent — useful for pipelines that should work with or without observability. from ace_next import OPIK_AVAILABLE if OPIK_AVAILABLE: print("Opik tracing is available") Troubleshooting: `~/.opik.config` --------------------------------- The Opik SDK stores a global config file at `~/.opik.config` (created by `opik.configure()`). This file **overrides environment variables** and can cause silent failures if it contains stale settings. If traces aren't appearing, check: cat ~/.opik.config A correct config for Comet Cloud looks like: [opik] url_override = https://www.comet.com/opik/api/ workspace = your-workspace-name Common issues: * **Wrong URL**: `https://www.comet.com/api/` (missing `/opik/`) causes 404 errors * **Wrong workspace**: `workspace = default` instead of your actual workspace name * **Stale config**: Re-run `opik.configure()` or edit the file directly to fix Disabling Tracing ----------------- # In CI or tests OPIK_DISABLED=true pytest tests/ # Or via the alternative variable OPIK_ENABLED=false python my_script.py Full Example ------------ ### ACELiteLLM (easiest) ```python from ace_next import ACELiteLLM, Sample, SimpleEnvironment ace = ACELiteLLM.from_model("gpt-4o-mini", opik=True, opik_project="ace-training") samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ Sample(question="Capital of France?", context="", ground_truth="Paris"),\ ] results = ace.learn(samples, environment=SimpleEnvironment(), epochs=3) ace.save("trained.json") # View traces at http://localhost:5173 → project "ace-training" ``` ### ACE runner (manual) ```python from ace_next import ( ACE, Agent, Reflector, SkillManager, Skillbook, LiteLLMClient, SimpleEnvironment, Sample, OpikStep, register_opik_litellm_callback, ) client = LiteLLMClient(model="gpt-4o-mini") runner = ACE.from_roles( agent=Agent(client), reflector=Reflector(client), skill_manager=SkillManager(client), environment=SimpleEnvironment(), extra_steps=[OpikStep(project_name="ace-training")], ) # Optionally add LLM-level cost tracking register_opik_litellm_callback(project_name="ace-training") samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ Sample(question="Capital of France?", context="", ground_truth="Paris"),\ ] results = runner.run(samples, epochs=3) runner.save("trained.json") ``` What to Read Next ----------------- * [Integration Pattern](https://kayba.ai/docs/guides/integration) — how runners compose pipeline steps * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — building pipelines from scratch * [Async Learning](https://kayba.ai/docs/guides/async-learning) — background learning with cost monitoring --- # API Reference | Docs | Kayba API Reference ============= Quick reference for the most-used classes and functions in `ace_next`. Runners ------- ### ACELiteLLM Simple self-improving conversational agent. from ace_next import ACELiteLLM agent = ACELiteLLM.from_model("gpt-4o-mini") | Method | Description | | --- | --- | | `ask(question, context="")` | Generate an answer using the current skillbook | | `learn(samples, environment, epochs=1, *, wait=True)` | Run the full ACE learning pipeline | | `learn_from_feedback(feedback, ground_truth=None)` | Learn from the last `ask()` interaction | | `learn_from_traces(traces, epochs=1, *, wait=True)` | Learn from pre-recorded execution traces | | `save(path)` | Save skillbook to JSON | | `load(path)` | Load skillbook from JSON | | `enable_learning()` / `disable_learning()` | Toggle learning on/off | | `wait_for_background(timeout=None)` | Wait for async learning to finish | | `learning_stats` | Dict with background learning progress | | `get_strategies()` | Formatted string of current strategies | See [LiteLLM Integration](https://kayba.ai/docs/integrations/litellm) for full details. ### ACE Full adaptive pipeline (Agent + Reflector + SkillManager + Environment). from ace_next import ACE, Agent, Reflector, SkillManager, Skillbook, LiteLLMClient, SimpleEnvironment client = LiteLLMClient(model="gpt-4o-mini") runner = ACE.from_roles( agent=Agent(client), reflector=Reflector(client), skill_manager=SkillManager(client), environment=SimpleEnvironment(), skillbook=Skillbook(), ) results = runner.run(samples, epochs=3) | Method | Description | | --- | --- | | `run(samples, epochs=1, wait=True)` | Run adaptation loop, return `list[SampleResult]` | | `save(path)` | Save skillbook | | `wait_for_background(timeout=None)` | Wait for async learning | | `learning_stats` | Background learning progress | See [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) . ### BrowserUse Browser automation with learning. from ace_next import BrowserUse runner = BrowserUse.from_model(browser_llm=my_llm, ace_model="gpt-4o-mini") results = runner.run("Find the top post on Hacker News") See [Browser-Use Integration](https://kayba.ai/docs/integrations/browser-use) . ### LangChain Wrap LangChain Runnables with learning. from ace_next import LangChain runner = LangChain.from_model(my_chain, ace_model="gpt-4o-mini") results = runner.run([{"input": "Summarize this document"}]) See [LangChain Integration](https://kayba.ai/docs/integrations/langchain) . ### ClaudeCode Claude Code CLI with learning. from ace_next import ClaudeCode runner = ClaudeCode.from_model(working_dir="./project", ace_model="gpt-4o-mini") results = runner.run("Add unit tests for utils.py") See [Claude Code Integration](https://kayba.ai/docs/integrations/claude-code) . * * * Roles ----- ### Agent Produces answers using the current skillbook. from ace_next import Agent agent = Agent(llm) output = agent.generate( question="What is 2+2?", context="", skillbook=skillbook, reflection=None, # optional ) **AgentOutput fields:** | Field | Type | Description | | --- | --- | --- | | `final_answer` | `str` | The generated answer | | `reasoning` | `str` | Step-by-step reasoning | | `skill_ids` | `list[str]` | Skillbook strategies cited | | `raw` | `dict` | Raw LLM response | ### Reflector Analyzes what worked and what failed. from ace_next import Reflector reflector = Reflector(llm) reflection = reflector.reflect( question="What is 2+2?", agent_output=output, skillbook=skillbook, ground_truth="4", feedback="Correct!", ) **ReflectorOutput fields:** | Field | Type | Description | | --- | --- | --- | | `reasoning` | `str` | Analysis of the outcome | | `error_identification` | `str` | What went wrong | | `root_cause_analysis` | `str` | Why it went wrong | | `correct_approach` | `str` | What should have been done | | `key_insight` | `str` | Main lesson learned | | `extracted_learnings` | `list[ExtractedLearning]` | Learnings with evidence and justification | | `skill_tags` | `list[SkillTag]` | `(skill_id, tag)` pairs | | `raw` | `dict` | Raw LLM response | ### SkillManager Transforms reflections into skillbook updates. from ace_next import SkillManager skill_manager = SkillManager(llm) sm_output = skill_manager.update_skills( reflection=reflection, skillbook=skillbook, question_context="Math problems", progress="3/5 correct", ) # Apply the updates skillbook.apply_update(sm_output.update) Returns a `SkillManagerOutput` with an `.update` field (`UpdateBatch`) and `.raw` field. See [Roles](https://kayba.ai/docs/concepts/roles) for full details. * * * Skillbook --------- from ace_next import Skillbook skillbook = Skillbook() | Method / Property | Description | | --- | --- | | `add_skill(section, content, metadata=None)` | Add a strategy | | `apply_update(update_batch)` | Apply update operations | | `as_prompt()` | TOON format for LLM consumption | | `save_to_file(path)` | Save to JSON | | `Skillbook.load_from_file(path)` | Load from JSON | | `stats()` | Section count, skill count, tag totals | | `skills()` | List of all skills | See [The Skillbook](https://kayba.ai/docs/concepts/skillbook) . * * * Data Types ---------- ### Sample from ace_next import Sample sample = Sample( question="What is 2+2?", context="Show your work", ground_truth="4", ) ### EnvironmentResult from ace_next import EnvironmentResult result = EnvironmentResult( feedback="Correct!", ground_truth="4", metrics={"accuracy": 1.0}, ) ### UpdateOperation from ace_next import UpdateOperation op = UpdateOperation( type="ADD", section="Math", content="Break problems into smaller steps", skill_id="math-00001", ) Operations: `ADD`, `UPDATE`, `TAG`, `REMOVE`. See [Update Operations](https://kayba.ai/docs/concepts/updates) . ### DeduplicationConfig **Requires:** `pip install ace-framework[deduplication]` from ace_next import DeduplicationConfig config = DeduplicationConfig( enabled=True, embedding_model="text-embedding-3-small", similarity_threshold=0.85, ) * * * Environments ------------ Extend `TaskEnvironment` to provide evaluation feedback: from ace_next import TaskEnvironment, EnvironmentResult class MyEnvironment(TaskEnvironment): def evaluate(self, sample, agent_output): correct = sample.ground_truth.lower() in agent_output.final_answer.lower() return EnvironmentResult( feedback="Correct!" if correct else "Incorrect", ground_truth=sample.ground_truth, ) A built-in `SimpleEnvironment` uses substring matching and is included for quick testing. * * * LLM Clients ----------- ### LiteLLMClient from ace_next import LiteLLMClient client = LiteLLMClient(model="gpt-4o-mini", temperature=0.0, max_tokens=2048) response = client.complete("Hello") Supports all [LiteLLM providers](https://docs.litellm.ai/) (OpenAI, Anthropic, Google, Ollama, etc.). ### InstructorClient Wraps any LLM client with Pydantic validation for more reliable structured outputs. **Requires:** `pip install ace-framework[instructor]` from ace_next import InstructorClient, LiteLLMClient client = InstructorClient(LiteLLMClient(model="ollama/gemma3:1b")) * * * Patterns -------- ### SubRunner Abstract base class for steps that run an internal `Pipeline` in a loop. Satisfies `StepProtocol` — can be placed directly in any pipeline. from ace_next.core import SubRunner Subclasses implement five template methods plus `__call__`: | Method | Signature | Description | | --- | --- | --- | | `_build_inner_pipeline` | `(**kwargs) -> Pipeline` | Return the step sequence for one iteration | | `_build_initial_context` | `(**kwargs) -> StepContext` | Return the context for the first iteration | | `_is_done` | `(ctx) -> bool` | Return `True` when the loop should stop | | `_extract_result` | `(ctx) -> Any` | Pull the final result from the terminal context | | `_accumulate` | `(ctx) -> StepContext` | Build the next iteration's context from the current one | | `_on_timeout` | `(last_ctx, iteration, **kwargs) -> Any` | Called when `max_iterations` is reached. Default raises `RuntimeError`. | | `run_loop` | `(**kwargs) -> Any` | Execute the loop (called by `__call__`, also usable standalone) | | `__call__` | `(ctx) -> StepContext` | `StepProtocol` entry point — map outer context to `run_loop` result | | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `max_iterations` | `int` | `20` | Maximum loop iterations before `_on_timeout` fires | **Example:** from ace_next.core import SubRunner from pipeline import Pipeline, StepContext class RefineRunner(SubRunner): requires = frozenset({"draft"}) provides = frozenset({"refined"}) def __init__(self, scorer, improver, threshold=0.9): super().__init__(max_iterations=10) self.scorer = scorer self.improver = improver self.threshold = threshold def _build_inner_pipeline(self, **kw): return Pipeline([ScoreStep(self.scorer), ImproveStep(self.improver)]) def _build_initial_context(self, **kw): return RefineContext(text=kw["draft"]) def _is_done(self, ctx): return ctx.score >= self.threshold def _extract_result(self, ctx): return ctx.text def _accumulate(self, ctx): return ctx.replace(iteration=ctx.iteration + 1) def _on_timeout(self, last_ctx, iteration, **kwargs): return last_ctx.text # best effort def __call__(self, ctx): result = self.run_loop(draft=ctx.metadata["draft"]) return ctx.replace(metadata=MappingProxyType({**ctx.metadata, "refined": result})) **Canonical implementation:** `RRStep` in `ace_next/rr/` — the Recursive Reflector's REPL loop. See [Building Custom Steps](https://kayba.ai/docs/pipeline/custom-steps) for the full guide. * * * Observability ------------- ### OpikStep Append to any pipeline for automatic tracing and cost tracking: from ace_next import OpikStep OpikStep(project_name="my-experiment", tags=["training"]) ### register\_opik\_litellm\_callback Standalone LLM cost tracking without pipeline traces: from ace_next import register_opik_litellm_callback register_opik_litellm_callback(project_name="my-experiment") See [Opik Observability](https://kayba.ai/docs/integrations/opik) . * * * Prompts ------- The default prompts are v2.1 (built into `ace_next`). Pass a custom template via `prompt_template`: agent = Agent(llm, prompt_template="Custom prompt with {skillbook}, {question}, {context}") reflector = Reflector(llm, prompt_template="Custom reflector prompt ...") skill_manager = SkillManager(llm, prompt_template="Custom skill manager prompt ...") See [Prompt Engineering](https://kayba.ai/docs/guides/prompts) . --- # Error Handling | Docs | Kayba Error Handling ============== The pipeline engine guarantees that every sample produces a `SampleResult` — nothing is dropped silently. One failing sample never blocks others. Retry logic is the responsibility of individual steps, not the pipeline. * * * SampleResult ------------ Every sample that enters `run()` produces exactly one `SampleResult`: @dataclass class SampleResult: sample: Any # the original input output: StepContext | None # final context (None if failed) error: Exception | None # the exception (None if succeeded) failed_at: str | None # step class name where error occurred cause: Exception | None = None # inner exception for BranchError | Field | On success | On failure | | --- | --- | --- | | `sample` | original input | original input | | `output` | final `StepContext` | `None` | | `error` | `None` | the exception | | `failed_at` | `None` | class name of the failing step (e.g. `"Tokenize"`) | | `cause` | `None` | inner exception when `failed_at == "Branch"` | > **Background steps** > > For steps after an `async_boundary`, `output` and `error` may still be `None` when `run()` returns. Call `pipe.wait_for_background()` to block until all background work completes and results are finalized. * * * Construction-time errors ------------------------ These are caught **before any data flows** — they surface immediately when you build the pipeline. ### PipelineOrderError Raised when a step requires a field that is produced by a **later** step in the pipeline: from pipeline import Pipeline from pipeline.errors import PipelineOrderError class Uppercase: requires = frozenset({"tokens"}) provides = frozenset({"upper_tokens"}) def __call__(self, ctx): ... class Tokenize: requires = frozenset() provides = frozenset({"tokens"}) def __call__(self, ctx): ... try: Pipeline().then(Uppercase()).then(Tokenize()) except PipelineOrderError as e: print(e) # Uppercase requires {"tokens"} but it is provided by a later step > **Tip** > > `PipelineOrderError` is always a bug — reorder your steps. Fields not produced by **any** step in the pipeline are treated as external inputs and do not trigger this error. ### PipelineConfigError Raised for invalid pipeline wiring: **Multiple async boundaries:** from pipeline.errors import PipelineConfigError class StepA: requires = frozenset() provides = frozenset({"a"}) async_boundary = True def __call__(self, ctx): ... class StepB: requires = frozenset({"a"}) provides = frozenset({"b"}) async_boundary = True def __call__(self, ctx): ... try: Pipeline().then(StepA()).then(StepB()) except PipelineConfigError: print("Only one async_boundary per pipeline is allowed") **Async boundary inside a Branch child:** try: Pipeline().branch( Pipeline().then(StepA()), # async_boundary = True inside branch ) except PipelineConfigError: print("async_boundary inside Branch children is not allowed") * * * Runtime errors -------------- ### Foreground failures When a step before the `async_boundary` (or in a pipeline with no boundary) raises an exception, the pipeline catches it per-sample and records it in the `SampleResult`: class Boom: requires = frozenset() provides = frozenset() def __call__(self, ctx): raise RuntimeError(f"Failed on {ctx.sample!r}") pipe = Pipeline().then(Tokenize()).then(Boom()) results = pipe.run([\ StepContext(sample="good"),\ StepContext(sample="also good"),\ ]) for r in results: if r.error: print(f"Sample '{r.sample}' failed at {r.failed_at}: {r.error}") else: print(f"Sample '{r.sample}' succeeded") Sample 'good' failed at Boom: Failed on 'good' Sample 'also good' failed at Boom: Failed on 'also good' Each sample is processed independently — one failure does not prevent others from running. ### Background failures When a step **after** the `async_boundary` raises, the caller has already moved on. The exception is captured and attached to the `SampleResult` in-place: pipe = Pipeline().then(Tokenize()).then(BrokenBackgroundStep()) results = pipe.run(samples) # results returned immediately — background still running pipe.wait_for_background(timeout=10.0) # Now check for background failures for r in results: if r.error: print(f"Background failure at {r.failed_at}: {r.error}") ### BranchError When one or more branch pipelines fail, a `BranchError` is raised with the full list of failures: from pipeline.errors import BranchError results = pipe.run(contexts) for r in results: if isinstance(r.error, BranchError): print(f"{len(r.error.failures)} branch(es) failed:") for f in r.error.failures: print(f" {type(f).__name__}: {f}") elif r.error: print(f"Step failure at {r.failed_at}: {r.error}") All branches run to completion before `BranchError` is raised — no branch is cancelled when another fails. The `SampleResult.cause` field carries the inner exception from the failing branch. * * * Inspecting results ------------------ The standard pattern after `run()`: results = pipe.run(contexts) pipe.wait_for_background() # if using async_boundary succeeded = [r for r in results if r.error is None] failed = [r for r in results if r.error is not None] print(f"{len(succeeded)} succeeded, {len(failed)} failed") for r in failed: print(f" Sample: {r.sample}") print(f" Failed at: {r.failed_at}") print(f" Error: {r.error}") * * * Background monitoring --------------------- ### `wait_for_background()` Blocks until all background tasks complete: # Wait indefinitely pipe.wait_for_background() # Wait with timeout — raises TimeoutError if not done pipe.wait_for_background(timeout=30.0) Completed threads are removed from the tracking list after this call. ### `background_stats()` Returns a snapshot of background task progress. Thread-safe — can be called from any thread while the pipeline is running: stats = pipe.background_stats() print(stats) # {'active': 2, 'completed': 8} --- # Full Pipeline | Docs | Kayba Full Pipeline Guide =================== This guide walks through building a complete ACE pipeline from scratch — choosing components, defining an environment, running training, and saving results. Components ---------- A full pipeline needs four things: 1. **LLM Client** — the language model powering all three roles 2. **Three Roles** — Agent, Reflector, SkillManager 3. **Environment** — evaluates agent outputs 4. **Samples** — training data with questions and ground truth Step 1: Create the LLM Client ----------------------------- from ace_next import LiteLLMClient llm = LiteLLMClient(model="gpt-4o-mini") For robust JSON parsing with small models, wrap with Instructor (requires `pip install ace-framework[instructor]`): from ace_next import LiteLLMClient, wrap_with_instructor llm = wrap_with_instructor(LiteLLMClient(model="ollama/gemma3:1b")) Step 2: Create the Roles ------------------------ from ace_next import Agent, Reflector, SkillManager agent = Agent(llm) reflector = Reflector(llm) skill_manager = SkillManager(llm) Optionally use a cheaper model for learning: agent_llm = LiteLLMClient(model="gpt-4o") learning_llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(agent_llm) reflector = Reflector(learning_llm) skill_manager = SkillManager(learning_llm) Step 3: Define an Environment ----------------------------- The environment evaluates agent outputs. Extend `TaskEnvironment` and implement `evaluate()`: from ace_next import TaskEnvironment, EnvironmentResult class MathEnvironment(TaskEnvironment): def evaluate(self, sample, agent_output): correct = str(sample.ground_truth).lower() in str(agent_output.final_answer).lower() return EnvironmentResult( feedback="Correct!" if correct else f"Incorrect. Expected: {sample.ground_truth}", ground_truth=sample.ground_truth, metrics={"accuracy": 1.0 if correct else 0.0}, ) Or use the built-in `SimpleEnvironment` for basic ground-truth matching: from ace_next import SimpleEnvironment environment = SimpleEnvironment() Step 4: Prepare Samples ----------------------- from ace_next import Sample samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ Sample(question="Capital of France?", context="", ground_truth="Paris"),\ Sample(question="Who wrote Hamlet?", context="", ground_truth="Shakespeare"),\ ] Step 5: Build and Run the Pipeline ---------------------------------- from ace_next import ACE runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=environment, ) results = runner.run(samples, epochs=3) Step 6: Save the Skillbook -------------------------- runner.save("trained.json") print(f"Learned {len(runner.skillbook.skills())} strategies") Complete Example ---------------- from ace_next import ( ACE, Agent, Reflector, SkillManager, LiteLLMClient, Sample, SimpleEnvironment, ) # LLM and roles llm = LiteLLMClient(model="gpt-4o-mini") agent = Agent(llm) reflector = Reflector(llm) skill_manager = SkillManager(llm) # Pipeline runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=SimpleEnvironment(), ) # Training data samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ Sample(question="Capital of France?", context="", ground_truth="Paris"),\ ] # Train and save results = runner.run(samples, epochs=3) runner.save("trained.json") Checkpoints ----------- Save the skillbook automatically during long training runs: runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=environment, checkpoint_dir="./checkpoints", checkpoint_interval=10, # Save every 10 samples ) This creates: * `ace_checkpoint_10.json`, `ace_checkpoint_20.json`, etc. * `ace_latest.json` (always the most recent) Deduplication ------------- Prevent duplicate skills from accumulating (requires `pip install ace-framework[deduplication]`): from ace_next import DeduplicationConfig, DeduplicationManager dedup = DeduplicationManager(DeduplicationConfig( enabled=True, embedding_model="text-embedding-3-small", similarity_threshold=0.85, )) runner = ACE.from_roles( ..., dedup_manager=dedup, dedup_interval=10, ) Custom Prompts -------------- The default prompts are v2.1 and work well out of the box. You can pass your own templates via the `prompt_template` parameter: agent = Agent(llm, prompt_template="Your custom agent prompt with {skillbook}, {question}, {context}") reflector = Reflector(llm, prompt_template="Your custom reflector prompt ...") skill_manager = SkillManager(llm, prompt_template="Your custom skill manager prompt ...") See [Prompt Engineering](https://kayba.ai/docs/guides/prompts) for template variables and more examples. Testing Without API Calls ------------------------- Use a mock to test pipeline wiring without making real LLM calls. Any object satisfying the `LLMClientLike` protocol (with `complete()` and `complete_structured()` methods) works: from unittest.mock import MagicMock mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "test", "final_answer": "4", "skill_ids": []}' agent = Agent(mock_llm) reflector = Reflector(mock_llm) skill_manager = SkillManager(mock_llm) Observability ------------- Add Opik tracing to any pipeline via `extra_steps` (requires `pip install ace-framework[observability]`): from ace_next import ACE, OpikStep, register_opik_litellm_callback runner = ACE.from_roles( agent=agent, reflector=reflector, skill_manager=skill_manager, environment=environment, extra_steps=[OpikStep(project_name="my-experiment")], ) # Optionally add per-LLM-call cost tracking register_opik_litellm_callback(project_name="my-experiment") See [Opik Observability](https://kayba.ai/docs/integrations/opik) for full details. What to Read Next ----------------- * [Async Learning](https://kayba.ai/docs/guides/async-learning) — parallel Reflector execution * [Prompt Engineering](https://kayba.ai/docs/guides/prompts) — customize prompt templates * [Integration Pattern](https://kayba.ai/docs/guides/integration) — wrap existing agents instead * [Opik Observability](https://kayba.ai/docs/integrations/opik) — monitor costs and traces --- # Core Concepts | Docs | Kayba Core Concepts ============= The pipeline engine is built on four foundational concepts: the **Step protocol**, the **StepContext**, the **contract system**, and the **Pipeline** compositor. Understanding these gives you the mental model for everything else. * * * StepProtocol ------------ A Step is any Python object that satisfies the `StepProtocol` — a structural (duck-typed) interface. No base class is required. from collections.abc import Set as AbstractSet from typing import Protocol, runtime_checkable @runtime_checkable class StepProtocol(Protocol): requires: AbstractSet[str] # metadata keys this step reads provides: AbstractSet[str] # metadata keys this step writes def __call__(self, ctx: StepContext) -> StepContext: ... Any object with `requires`, `provides`, and a `__call__` method is a valid step: class Tokenize: requires = frozenset() # no dependencies provides = frozenset({"tokens", "word_count"}) def __call__(self, ctx: StepContext) -> StepContext: tokens = str(ctx.sample).split() return ctx.replace( metadata=MappingProxyType({ **ctx.metadata, "tokens": tokens, "word_count": len(tokens), }) ) Key details: * **`AbstractSet[str]`** accepts both `set` and `frozenset`. Steps can use plain set literals — the pipeline normalizes them to `frozenset` at construction time. * **`@runtime_checkable`** lets the pipeline use `isinstance(step, StepProtocol)` at construction time to catch missing attributes early, rather than failing at call time. * **`Pipeline` and `Branch`** both satisfy this protocol, so they can be nested wherever a step is expected. * * * StepContext ----------- `StepContext` is the data carrier passed from step to step. It is a **frozen dataclass** — steps never mutate the incoming context. from dataclasses import dataclass, field from types import MappingProxyType @dataclass(frozen=True) class StepContext: sample: Any = None metadata: MappingProxyType = field( default_factory=lambda: MappingProxyType({}) ) def replace(self, **changes) -> "StepContext": return dataclasses.replace(self, **changes) The engine only reads `sample` and `metadata`. All domain-specific fields are added by subclassing. ### The `.replace()` pattern Steps create new contexts — they never mutate the incoming one: def __call__(self, ctx: StepContext) -> StepContext: result = process(ctx.sample) return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "result": result}) ) This is the only way to "modify" a context. `frozen=True` makes mutation a hard error at runtime rather than a subtle bug. ### Metadata auto-coercion If a caller passes a plain `dict` as metadata, `StepContext.__post_init__` automatically wraps it in `MappingProxyType`, ensuring mutation is always a runtime error: # Both of these produce identical immutable metadata: ctx = StepContext(sample="hello", metadata={"key": "value"}) ctx = StepContext(sample="hello", metadata=MappingProxyType({"key": "value"})) ### Subclassing for domain fields Applications subclass `StepContext` to add named fields for concepts shared across their pipelines: @dataclass(frozen=True) class MLContext(StepContext): # Shared configuration model_config: dict | None = None # Produced by steps (None until the providing step runs) predictions: list | None = None scores: dict | None = None report: str | None = None Use **named fields** for data shared across multiple steps in the pipeline. Use **`metadata`** for integration-specific or step-specific transient data that doesn't warrant a dedicated field. > **When to use which** > > * Named field: `predictions`, `scores` — shared by multiple steps, type-checkable > * Metadata: `metadata["debug_log"]`, `metadata["cache_key"]` — step-specific, doesn't pollute the class ### Why immutability? * **Branch safety** — All branches receive the same frozen context. No deep copy is needed since no branch can mutate what it receives. * **Thread safety** — Steps running concurrently (via `workers` or `Branch`) can safely share context objects. * **Debugging** — Each step returns a new context, creating a clear trace of data transformations. * * * Contracts: requires and provides -------------------------------- Every step declares: * **`requires`** — the set of field names it reads from the context * **`provides`** — the set of field names it writes to the context The pipeline validates these at construction time. ### How validation works When you build a pipeline with `.then()`, the engine checks step ordering immediately: pipe = ( Pipeline() .then(Tokenize()) # provides: {"tokens", "word_count"} .then(Uppercase()) # requires: {"tokens"} ✓ — Tokenize provides it ) If a step requires a field that a **later** step provides, the pipeline raises `PipelineOrderError`: # This raises PipelineOrderError at construction time: pipe = Pipeline().then(Uppercase()).then(Tokenize()) # ↑ Uppercase requires "tokens", but Tokenize (which provides it) comes after ### External inputs vs internal dependencies Fields not produced by any step in the pipeline are treated as **external inputs** — they must be present in the initial `StepContext` passed to `run()`. These do not trigger ordering errors: class ScoreStep: requires = frozenset({"predictions"}) # external input provides = frozenset({"scores"}) # No error — "predictions" is expected to come from the initial context pipe = Pipeline().then(ScoreStep()) ### Contract inference for nested pipelines When a `Pipeline` is used as a step inside another pipeline, its `requires` and `provides` are computed automatically from its inner steps: inner = Pipeline().then(Tokenize()).then(Uppercase()) # Inferred automatically: # inner.requires = frozenset() — Tokenize needs nothing external # inner.provides = frozenset({"tokens", "word_count", "upper_tokens"}) outer = Pipeline().then(inner).then(Summarize()) # Summarize's requirements validated against inner.provides The inference algorithm: 1. Walk steps in order, tracking what has been provided so far 2. `requires` = fields needed by steps that no earlier step provides (external dependencies) 3. `provides` = union of all fields any step writes > **All Branch children always run** > > The contract system assumes all `Branch` children execute. There is no concept of conditional branches where only some children run — all branches always run. If a branch provides a field that a later step requires, validation passes; if that branch were to not run, the pipeline would fail at runtime. * * * Pipeline -------- A `Pipeline` is an ordered list of steps that runs sequentially for a single input. It satisfies the `StepProtocol`, so it can be nested inside other pipelines. ### Building a pipeline Two equivalent forms: ### Fluent builder (preferred) ```python pipe = ( Pipeline() .then(Tokenize()) .then(Uppercase()) .then(Summarize()) ) ``` ### Constructor list ```python pipe = Pipeline([\ Tokenize(),\ Uppercase(),\ Summarize(),\ ]) ``` Both validate step ordering at construction time. The fluent builder validates **after each `.then()` call**, giving precise error messages about which step caused the violation. ### The `.branch()` shorthand Instead of manually creating a `Branch`, use the fluent shorthand: pipe = ( Pipeline() .then(Tokenize()) .branch( Pipeline().then(Uppercase()), Pipeline().then(Reverse()), merge=MergeStrategy.RAISE_ON_CONFLICT, ) .then(Summarize()) ) This is equivalent to `.then(Branch(...))`. ### Nesting A pipeline used as a step is a black box — the outer pipeline sees only its aggregated `requires` and `provides`: preprocessing = Pipeline().then(Tokenize()).then(Uppercase()) postprocessing = Pipeline().then(Summarize()).then(FormatStep()) full = Pipeline().then(preprocessing).then(postprocessing) > **Inner pipeline as a fan-out step** > > A step receives one context and must return one context — but nothing prevents it from internally expanding to multiple sub-inputs: > > class MultiSearchStep: > requires = frozenset() > provides = frozenset({"search_results"}) > > def __call__(self, ctx: StepContext) -> StepContext: > queries = generate_queries(ctx.sample) > sub_ctxs = [StepContext(sample=q) for q in queries] > sub_pipe = Pipeline().then(FetchStep()) > results = sub_pipe.run(sub_ctxs, workers=len(queries)) > merged = merge_results(results) > return ctx.replace( > metadata=MappingProxyType({**ctx.metadata, "search_results": merged}) > ) > > > From the outer pipeline's perspective, `MultiSearchStep` is a single step. The fan-out is an internal implementation detail. --- # Building Custom Steps | Docs | Kayba Building Custom Steps ===================== This guide covers everything you need to create your own pipeline steps — from the minimal contract to advanced patterns like dependency injection, async execution, and testing. * * * The step contract ----------------- Any Python object with `requires`, `provides`, and `__call__` is a valid step. No base class needed. from types import MappingProxyType from pipeline import StepContext class MyStep: requires = frozenset({"input_field"}) provides = frozenset({"output_field"}) def __call__(self, ctx: StepContext) -> StepContext: result = process(ctx.metadata["input_field"]) return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "output_field": result}) ) Rules: * `requires` and `provides` can be `set` or `frozenset` — the pipeline normalizes to `frozenset` * `__call__` receives a `StepContext` and must return a `StepContext` * Never mutate the incoming context — always use `.replace()` * * * Sync vs async steps ------------------- ### Sync ```python class ComputeStep: requires = frozenset({"data"}) provides = frozenset({"result"}) def __call__(self, ctx: StepContext) -> StepContext: result = expensive_computation(ctx.metadata["data"]) return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "result": result}) ) ``` ### Async ```python class FetchStep: requires = frozenset({"url"}) provides = frozenset({"response"}) async def __call__(self, ctx: StepContext) -> StepContext: async with aiohttp.ClientSession() as session: resp = await session.get(ctx.metadata["url"]) data = await resp.json() return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "response": data}) ) ``` Use async steps for I/O-bound work (HTTP requests, API calls, file I/O). The pipeline detects and handles both transparently. * * * Dependency injection -------------------- Steps that need external collaborators receive them via `__init__`. The `__call__` method stays stateless — it only uses `self.*` for injected dependencies and `ctx` for data. class ScoringStep: requires = frozenset({"predictions"}) provides = frozenset({"scores"}) def __init__(self, scorer, threshold: float = 0.5): self.scorer = scorer self.threshold = threshold def __call__(self, ctx: StepContext) -> StepContext: raw_scores = self.scorer.evaluate(ctx.metadata["predictions"]) filtered = {k: v for k, v in raw_scores.items() if v >= self.threshold} return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "scores": filtered}) ) This makes testing easy — inject mocks: pipe = Pipeline().then(ScoringStep(scorer=mock_scorer, threshold=0.8)) * * * Declaring concurrency --------------------- Two optional class attributes control how a step participates in concurrent execution: ### `async_boundary` Marks the foreground/background split point. Everything from this step onward runs in a background thread: class AnalyzeStep: requires = frozenset({"data"}) provides = frozenset({"analysis"}) async_boundary = True # background from here def __call__(self, ctx: StepContext) -> StepContext: ... See [Execution Model — Async Boundary](https://kayba.ai/docs/pipeline/execution#async-boundary-fire-and-forget-background) for details. ### `max_workers` Controls the per-step-class thread pool size for background execution: class ParallelAnalyzeStep: requires = frozenset({"data"}) provides = frozenset({"analysis"}) async_boundary = True max_workers = 4 # up to 4 concurrent analyses def __call__(self, ctx: StepContext) -> StepContext: ... Default is `max_workers = 1` (serialized). > **Warning** > > Steps that write shared state (e.g. updating an external database or accumulating results into a shared object) must use `max_workers = 1` to avoid race conditions. * * * Subclassing StepContext ----------------------- When `metadata` becomes unwieldy, subclass `StepContext` to add named fields: from dataclasses import dataclass @dataclass(frozen=True) class MLContext(StepContext): predictions: list | None = None scores: dict | None = None report: str | None = None Steps write to named fields using `.replace()`: class PredictStep: requires = frozenset() provides = frozenset({"predictions"}) def __init__(self, model): self.model = model def __call__(self, ctx: MLContext) -> MLContext: preds = self.model.predict(ctx.sample) return ctx.replace(predictions=preds) > **When to subclass** > > * **Named fields**: Data shared across multiple steps that benefits from type checking > * **Metadata**: Step-specific or integration-specific transient data (e.g. `metadata["cache_key"]`) > > The `requires`/`provides` validation works on attribute names, so it's subclass-agnostic. A step declaring `requires = {"predictions"}` works with any context subclass that has a `predictions` attribute. * * * Testing steps ------------- ### Unit test — step in isolation from types import MappingProxyType from pipeline import StepContext def test_tokenize_splits_words(): step = Tokenize() ctx = StepContext(sample="hello world") result = step(ctx) assert result.metadata["tokens"] == ["hello", "world"] assert result.metadata["word_count"] == 2 def test_uppercase_transforms_tokens(): step = Uppercase() ctx = StepContext( metadata=MappingProxyType({"tokens": ["hello", "world"]}) ) result = step(ctx) assert result.metadata["upper_tokens"] == ["HELLO", "WORLD"] ### Protocol compliance from pipeline import StepProtocol def test_step_satisfies_protocol(): step = Tokenize() assert isinstance(step, StepProtocol) assert hasattr(step, "requires") assert hasattr(step, "provides") assert callable(step) ### Pipeline integration test from pipeline import Pipeline, StepContext def test_full_pipeline(): pipe = Pipeline().then(Tokenize()).then(Uppercase()) results = pipe.run([StepContext(sample="hello world")]) assert len(results) == 1 assert results[0].error is None assert results[0].output.metadata["upper_tokens"] == ["HELLO", "WORLD"] * * * Common patterns --------------- ### Map-reduce step A step that internally fans out to multiple sub-inputs: class MultiSearchStep: requires = frozenset() provides = frozenset({"search_results"}) def __call__(self, ctx: StepContext) -> StepContext: queries = generate_queries(ctx.sample) # 1 → N sub_ctxs = [StepContext(sample=q) for q in queries] sub_pipe = Pipeline().then(FetchStep()) results = sub_pipe.run(sub_ctxs, workers=len(queries)) # parallel merged = merge_results(results) # N → 1 return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "search_results": merged}) ) From the outer pipeline's perspective, this is a black box that takes one context and returns one. ### Logging / observability step A pass-through step that logs without modifying data: class LogStep: requires = frozenset() provides = frozenset() def __init__(self, logger): self.logger = logger def __call__(self, ctx: StepContext) -> StepContext: self.logger.info(f"Processing sample: {ctx.sample}") self.logger.debug(f"Metadata keys: {list(ctx.metadata.keys())}") return ctx # pass through unchanged ### Retry wrapper A step that wraps another step with retry logic: import time class RetryStep: def __init__(self, inner, max_retries: int = 3, delay: float = 1.0): self.inner = inner self.max_retries = max_retries self.delay = delay self.requires = inner.requires self.provides = inner.provides def __call__(self, ctx: StepContext) -> StepContext: for attempt in range(self.max_retries): try: return self.inner(ctx) except Exception: if attempt == self.max_retries - 1: raise time.sleep(self.delay * (attempt + 1)) Usage: pipe = Pipeline().then(RetryStep(FlakyAPIStep(), max_retries=3)) --- # The Skillbook | Docs | Kayba The Skillbook ============= The **Skillbook** is ACE's knowledge store — a structured collection of strategies that the agent has learned from experience. Each strategy is called a **skill**. What Is a Skill? ---------------- A skill is a single strategy entry with: | Field | Description | | --- | --- | | `id` | Unique identifier (e.g., `math-00001`) | | `section` | Category grouping (e.g., `"Math Strategies"`) | | `content` | The strategy text | | `helpful` | Times this skill contributed to a correct answer | | `harmful` | Times this skill led to an incorrect answer | | `neutral` | Times cited but had no clear effect | Example skill: { "id": "math-00001", "section": "Math Strategies", "content": "Break complex problems into smaller steps before computing", "helpful": 5, "harmful": 0, "neutral": 1 } Skill Lifecycle --------------- Skills go through four stages: 1. **Created** — the SkillManager adds a new skill after a reflection 2. **Tagged** — each time the Agent cites a skill, the Reflector tags it as helpful, harmful, or neutral 3. **Updated** — the SkillManager may refine a skill's content based on new learnings 4. **Removed** — skills that are consistently harmful get pruned These correspond to four [update operations](https://kayba.ai/docs/concepts/updates) : `ADD`, `TAG`, `UPDATE`, `REMOVE`. TOON Format ----------- When the skillbook is included in LLM prompts, it uses **TOON** (Token-Oriented Object Notation) — a compact format that saves 16-62% tokens compared to JSON: skillbook.as_prompt() # TOON format for LLM consumption skills[3]{id section content helpful harmful neutral}: math-00001 Math Strategies Break complex problems into smaller steps 5 0 1 math-00002 Math Strategies Verify answers by working backwards 3 1 0 logic-00001 Logic Check edge cases before concluding 2 0 0 For human debugging, use the string representation: str(skillbook) # Markdown format for readability Sections -------- Skills are organized into sections. Sections emerge naturally from the SkillManager's categorization during learning: from ace_next import Skillbook skillbook = Skillbook() # Add skills to specific sections skillbook.add_skill( section="Math Strategies", content="Break complex problems into smaller steps", metadata={"helpful": 5, "harmful": 0, "neutral": 1}, ) Persistence ----------- # Save skillbook.save_to_file("strategies.json") # Load skillbook = Skillbook.load_from_file("strategies.json") Statistics ---------- stats = skillbook.stats() # {"sections": 3, "skills": 15, "tags": {"helpful": 45, "harmful": 5, "neutral": 10}} Deduplication ------------- As the skillbook grows, similar skills can accumulate. The `DeduplicationManager` detects and consolidates them using embedding similarity: from ace_next import DeduplicationConfig, DeduplicationManager config = DeduplicationConfig( enabled=True, embedding_model="text-embedding-3-small", similarity_threshold=0.85, within_section_only=True, ) dedup = DeduplicationManager(config) When used with a runner, deduplication runs automatically at a configurable interval: runner = ACE.from_roles( ..., dedup_manager=dedup, dedup_interval=10, # Every 10 samples ) Insight Source Tracing ---------------------- Each skill tracks where it came from — which sample, epoch, and step produced it. Query this with: sources = skillbook.source_map() # skill_id -> source info summary = skillbook.source_summary() # Aggregated statistics What to Read Next ----------------- * [Update Operations](https://kayba.ai/docs/concepts/updates) — how ADD, UPDATE, TAG, REMOVE work * [Three Roles](https://kayba.ai/docs/concepts/roles) — which role creates, tags, and updates skills * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — see the skillbook in action --- # LiteLLM | Docs | Kayba LiteLLM Integration =================== `ACELiteLLM` is the simplest way to get a self-improving agent. It bundles Agent, Reflector, SkillManager, and Skillbook into a single class with `ask()` and `learn()` methods. Quick Start ----------- from ace_next import ACELiteLLM agent = ACELiteLLM.from_model("gpt-4o-mini") # Ask questions — learns patterns across them answer = agent.ask("If all cats are animals, is Felix (a cat) an animal?") # Save and reload agent.save("learned.json") Parameters ---------- ### from\_model() | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"gpt-4o-mini"` | LiteLLM model identifier | | `max_tokens` | `int` | `2048` | Max tokens for responses | | `temperature` | `float` | `0.0` | Sampling temperature | | `api_key` | `str` | `None` | API key (or use env variable) | | `base_url` | `str` | `None` | Custom API endpoint | | `skillbook_path` | `str` | `None` | Path to load saved skillbook | | `environment` | `TaskEnvironment` | `None` | Evaluation environment | | `dedup_config` | `DeduplicationConfig` | `None` | Skill deduplication config | | `is_learning` | `bool` | `True` | Enable/disable learning | | `opik` | `bool` | `False` | Enable Opik observability (pipeline traces + LiteLLM per-call cost tracking) | | `opik_project` | `str` | `"ace-framework"` | Opik project name for organizing traces | | `opik_tags` | `list[str]` | `None` | Tags applied to every Opik trace | Methods ------- ### ask() Direct agent call using the current skillbook: answer = agent.ask("Your question", context="Optional context") ### learn() Run the full ACE learning pipeline over samples: from ace_next import Sample, SimpleEnvironment samples = [\ Sample(question="What is 2+2?", context="", ground_truth="4"),\ ] results = agent.learn(samples, environment=SimpleEnvironment(), epochs=3) ### learn\_from\_feedback() Learn from the last `ask()` interaction: agent.ask("What is the capital of France?") agent.learn_from_feedback(feedback="Correct!", ground_truth="Paris") ### learn\_from\_traces() Learn from pre-recorded execution traces: results = agent.learn_from_traces(traces, epochs=1) ### Lifecycle agent.save("path.json") # Save skillbook agent.load("path.json") # Load skillbook agent.enable_learning() # Turn on learning agent.disable_learning() # Turn off learning agent.wait_for_background() # Wait for async learning agent.learning_stats # Background progress agent.skillbook # Current Skillbook agent.get_strategies() # Formatted strategies Using a Cheaper Learning Model ------------------------------ Use a strong model for the Agent and a cheaper one for learning: from ace_next import ACELiteLLM, Agent, Reflector, SkillManager, LiteLLMClient agent_llm = LiteLLMClient(model="gpt-4o") learning_llm = LiteLLMClient(model="gpt-4o-mini") ace = ACELiteLLM( llm=agent_llm, agent=Agent(agent_llm), reflector=Reflector(learning_llm), skill_manager=SkillManager(learning_llm), ) Deduplication ------------- Prevent duplicate skills from accumulating: from ace_next import DeduplicationConfig agent = ACELiteLLM.from_model( "gpt-4o-mini", dedup_config=DeduplicationConfig( enabled=True, embedding_model="text-embedding-3-small", similarity_threshold=0.85, ), ) Supported Providers ------------------- Any model supported by [LiteLLM](https://docs.litellm.ai/) : # OpenAI agent = ACELiteLLM.from_model("gpt-4o-mini") # Anthropic agent = ACELiteLLM.from_model("claude-sonnet-4-5-20250929") # Google agent = ACELiteLLM.from_model("gemini-pro") # Local (Ollama) agent = ACELiteLLM.from_model("ollama/llama2") # Custom endpoint agent = ACELiteLLM.from_model("gpt-4o-mini", base_url="https://your-endpoint.com") Opik Observability ------------------ Enable tracing and cost tracking with a single flag: ace = ACELiteLLM.from_model("gpt-4o-mini", opik=True, opik_project="my-experiment") # Both tracing modes are enabled: # 1. Pipeline traces (OpikStep) — one trace per sample with ACE context # 2. LiteLLM callback — per-LLM-call token/cost tracking results = ace.learn(samples, environment=SimpleEnvironment(), epochs=3) # View traces at http://localhost:5173 → project "my-experiment" See [Opik Observability](https://kayba.ai/docs/integrations/opik) for full details, environment variables, and manual setup. What to Read Next ----------------- * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — for more control over the pipeline * [Async Learning](https://kayba.ai/docs/guides/async-learning) — background learning with `wait=False` * [Opik Observability](https://kayba.ai/docs/integrations/opik) — monitor costs and traces --- # Testing | Docs | Kayba Testing ======= Running Tests ------------- ### pytest (recommended) ```bash uv run pytest # All tests uv run pytest -m unit # Unit tests only uv run pytest -m integration # Integration tests only uv run pytest tests/test_skillbook.py # Specific file uv run pytest -v # Verbose output ``` ### unittest ```bash python -m unittest discover -s tests python -m unittest discover -s tests -v # Verbose ``` Testing Without API Calls ------------------------- Use a mock LLM to test pipeline wiring without making real API calls. Any object with `complete()` and `complete_structured()` methods satisfies the `LLMClientLike` protocol: from unittest.mock import MagicMock from ace_next import Agent, Reflector, SkillManager mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "test", "final_answer": "4", "skill_ids": []}' agent = Agent(mock_llm) reflector = Reflector(mock_llm) skill_manager = SkillManager(mock_llm) Unit Testing ------------ ### Testing the Skillbook from ace_next import Skillbook def test_add_skill(): skillbook = Skillbook() skill = skillbook.add_skill( section="Test", content="Test strategy", metadata={"helpful": 0, "harmful": 0, "neutral": 0}, ) assert len(skillbook.skills()) == 1 assert skill.content == "Test strategy" def test_save_load(tmp_path): skillbook = Skillbook() skillbook.add_skill(section="Test", content="Strategy") path = str(tmp_path / "test.json") skillbook.save_to_file(path) loaded = Skillbook.load_from_file(path) assert len(loaded.skills()) == 1 ### Testing the Agent from unittest.mock import MagicMock from ace_next import Agent, Skillbook def test_agent_generate(): mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "2+2=4", "final_answer": "4", "skill_ids": []}' agent = Agent(mock_llm) output = agent.generate( question="What is 2+2?", context="", skillbook=Skillbook(), ) assert output.final_answer is not None assert output.reasoning is not None ### Testing Reflector and SkillManager from unittest.mock import MagicMock from ace_next import Agent, Reflector, SkillManager, Skillbook def make_mock_llm(): mock = MagicMock() mock.complete.return_value = '{"reasoning": "test", "final_answer": "4", "skill_ids": []}' return mock def test_reflector(): mock_llm = make_mock_llm() reflector = Reflector(mock_llm) agent = Agent(mock_llm) output = agent.generate(question="Test", context="", skillbook=Skillbook()) reflection = reflector.reflect( question="Test", agent_output=output, skillbook=Skillbook(), ground_truth="expected", feedback="Correct", ) assert reflection.key_insight is not None def test_skill_manager(): sm = SkillManager(make_mock_llm()) # ... similar pattern with reflection input Integration Testing ------------------- ### End-to-End Learning Cycle from unittest.mock import MagicMock from ace_next import ( ACE, Agent, Reflector, SkillManager, Sample, SimpleEnvironment, ) def test_full_learning_cycle(): mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "test", "final_answer": "answer", "skill_ids": []}' runner = ACE.from_roles( agent=Agent(mock_llm), reflector=Reflector(mock_llm), skill_manager=SkillManager(mock_llm), environment=SimpleEnvironment(), ) samples = [Sample(question="Test", context="", ground_truth="answer")] results = runner.run(samples, epochs=1) assert len(results) == 1 ### Testing Checkpoints def test_checkpoints(tmp_path): mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "test", "final_answer": "A", "skill_ids": []}' runner = ACE.from_roles( agent=Agent(mock_llm), reflector=Reflector(mock_llm), skill_manager=SkillManager(mock_llm), environment=SimpleEnvironment(), checkpoint_dir=str(tmp_path), checkpoint_interval=1, ) samples = [Sample(question="Q", context="", ground_truth="A")] runner.run(samples, epochs=1) # Check that checkpoint files were created checkpoints = list(tmp_path.glob("ace_*.json")) assert len(checkpoints) > 0 Common Test Patterns -------------------- ### Fixtures import pytest from unittest.mock import MagicMock from ace_next import Agent, Reflector, SkillManager, Skillbook @pytest.fixture def mock_llm(): mock = MagicMock() mock.complete.return_value = '{"reasoning": "test", "final_answer": "4", "skill_ids": []}' return mock @pytest.fixture def skillbook(): return Skillbook() @pytest.fixture def agent(mock_llm): return Agent(mock_llm) ### Mocking LLM Responses from unittest.mock import MagicMock def test_with_mock(): mock_llm = MagicMock() mock_llm.complete.return_value = '{"reasoning": "...", "final_answer": "4", "skill_ids": []}' agent = Agent(mock_llm) # ... CI Configuration ---------------- # .github/workflows/test.yml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v4 - run: uv sync - run: uv run pytest -v Code Quality ------------ uv run black ace/ tests/ examples/ # Format uv run mypy ace/ # Type check uv run pre-commit run --all-files # All hooks Troubleshooting --------------- | Problem | Solution | | --- | --- | | Import errors | Run `uv sync` to install all dependencies | | API key errors in tests | Use `MagicMock` for unit tests (see above) | | Flaky async tests | Increase timeout or use `wait_for_background()` | | Coverage too low | `--cov-fail-under=25` is the threshold | What to Read Next ----------------- * [Full Pipeline Guide](https://kayba.ai/docs/guides/full-pipeline) — what you're testing * [Async Learning](https://kayba.ai/docs/guides/async-learning) — testing async pipelines --- # Overview | Docs | Kayba Pipeline Engine =============== A generic, composable step runner for ordered and parallel data processing. * * * What is the Pipeline Engine? ---------------------------- The Pipeline Engine is a lightweight, domain-agnostic framework for composing processing steps into pipelines. It provides contract validation, immutable context passing, and built-in concurrency control — all in ~300 lines of pure Python with no external dependencies beyond the standard library. Everything composes from three primitives: **Sequential** — steps run one after another: **Branch** — fork, run in parallel, join: **Nesting** — a pipeline used as a step: Steps declare what data they read and write. The pipeline validates ordering at construction time — before any data flows — so wiring errors surface immediately, not at runtime. * * * Core Principles --------------- * **Three primitives** — Sequential steps, parallel branches, and nested pipelines cover every composition pattern * **Contracts** — Steps declare `requires` and `provides` fields; the pipeline validates ordering at construction time * **Immutable context** — Steps receive a frozen context and return a new one via `.replace()`, making concurrent execution safe by default * **Declared concurrency** — Parallelism is configured on the step (`max_workers`, `async_boundary`), not the pipeline * **Per-sample error isolation** — One failing sample never blocks others; every sample produces a result * * * Architecture at a Glance ------------------------ `Pipeline` and `Branch` both satisfy `StepProtocol` through structural typing — no inheritance required. This means a `Pipeline` can be used as a step inside another pipeline, and a `Branch` slots into any step position. | Concept | What it is | Threading | Data flow | | --- | --- | --- | --- | | **Step** | Single unit of work | Sync internally | Receives and returns `StepContext` | | **Pipeline** | Ordered chain of steps | `workers=N` across samples | Passes `StepContext` step-to-step | | **Branch** | Parallel fork/join | One thread per branch | Copies context in, merges outputs | | **Nested Pipeline** | Pipeline used as a step | Inherits parent threading | Same `StepContext` flow | * * * Async Boundary — Background Processing -------------------------------------- One of the engine's key features is the **async boundary**: a way to split a pipeline into foreground (fast return) and background (fire-and-forget) stages. Mark any step with `async_boundary = True` — the pipeline returns results immediately after the foreground steps, while everything from the boundary onward continues in background threads. Use `pipe.wait_for_background()` when you need the final results. This is critical for pipelines where early steps produce user-facing output quickly but later steps (analysis, logging, scoring) are slow and don't need to block the caller. See [Execution Model](https://kayba.ai/docs/pipeline/execution) for full details. * * * When to Use ----------- > **Good fit** > > * Ordered multi-step processing with explicit data dependencies > * Parallel fork/join patterns (multiple independent operations on the same data) > * Fire-and-forget background processing with `async_boundary` > * Any pipeline where you want construction-time contract validation > **Not designed for** > > * DAG scheduling with complex dependency graphs > * Distributed computing across multiple machines > * Stream processing with backpressure > * ETL pipelines requiring a data catalog * * * Installation ------------ The pipeline engine is included in the project with no extra dependencies: from pipeline import Pipeline, Branch, StepContext, MergeStrategy * * * What's Next ----------- * [**Quick Start**](https://kayba.ai/docs/pipeline/quick-start) — Build and run your first pipeline in under 30 lines * [**Core Concepts**](https://kayba.ai/docs/pipeline/core-concepts) — Understand Step, Context, and the contract system * [**Execution Model**](https://kayba.ai/docs/pipeline/execution) — Three types of async, workers, and background processing * [**Branching & Parallelism**](https://kayba.ai/docs/pipeline/branching) — Parallel fork/join with merge strategies * [**Error Handling**](https://kayba.ai/docs/pipeline/error-handling) — Per-sample isolation, SampleResult, and error types * [**Building Custom Steps**](https://kayba.ai/docs/pipeline/custom-steps) — Create your own steps with dependency injection * [**API Reference**](https://kayba.ai/docs/pipeline/api-reference) — Complete signatures for all public classes --- # Quick Start | Docs | Kayba Quick Start =========== Build and run your first pipeline in under 30 lines. * * * Define two steps ---------------- Every step needs three things: `requires`, `provides`, and a `__call__` method. from types import MappingProxyType from pipeline import Pipeline, StepContext class Tokenize: """Split text into tokens and count words.""" requires = frozenset() provides = frozenset({"tokens", "word_count"}) def __call__(self, ctx: StepContext) -> StepContext: tokens = str(ctx.sample).split() return ctx.replace( metadata=MappingProxyType({ **ctx.metadata, "tokens": tokens, "word_count": len(tokens), }) ) class Uppercase: """Convert tokens to uppercase.""" requires = frozenset({"tokens"}) provides = frozenset({"upper_tokens"}) def __call__(self, ctx: StepContext) -> StepContext: upper = [t.upper() for t in ctx.metadata["tokens"]] return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "upper_tokens": upper}) ) * * * Build and run ------------- Chain steps with `.then()` and run with a list of contexts: pipe = Pipeline().then(Tokenize()).then(Uppercase()) results = pipe.run([\ StepContext(sample="hello world"),\ StepContext(sample="pipeline engine demo"),\ ]) The pipeline validates ordering at construction time — if `Uppercase` came before `Tokenize`, you'd get a `PipelineOrderError` immediately, not at runtime. * * * Inspect results --------------- Every sample produces exactly one `SampleResult`: for r in results: if r.error: print(f"Failed at {r.failed_at}: {r.error}") else: print(f"Sample: {r.sample}") print(f"Tokens: {r.output.metadata['upper_tokens']}") print(f"Count: {r.output.metadata['word_count']}") Sample: hello world Tokens: ['HELLO', 'WORLD'] Count: 2 Sample: pipeline engine demo Tokens: ['PIPELINE', 'ENGINE', 'DEMO'] Count: 3 * * * Add parallelism with Branch --------------------------- Run independent steps simultaneously with `Branch`: from pipeline import MergeStrategy class Reverse: """Reverse each token.""" requires = frozenset({"tokens"}) provides = frozenset({"reversed_tokens"}) def __call__(self, ctx: StepContext) -> StepContext: rev = [t[::-1] for t in ctx.metadata["tokens"]] return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "reversed_tokens": rev}) ) pipe = ( Pipeline() .then(Tokenize()) .branch( Pipeline().then(Uppercase()), # runs in parallel Pipeline().then(Reverse()), # runs in parallel merge=MergeStrategy.RAISE_ON_CONFLICT, ) ) results = pipe.run([StepContext(sample="fork join")]) meta = results[0].output.metadata print(meta["upper_tokens"]) # ['FORK', 'JOIN'] print(meta["reversed_tokens"]) # ['krof', 'nioj'] Both branches write to different fields (`upper_tokens` vs `reversed_tokens`), so `RAISE_ON_CONFLICT` passes through without raising. * * * Fire-and-forget with async\_boundary ------------------------------------ Some steps are slow and don't need to block the caller. Mark a step with `async_boundary = True` to hand everything from that point onward to a background thread — `run()` returns immediately after the foreground steps. import time class SlowScore: """Expensive scoring that runs in the background.""" requires = frozenset({"tokens"}) provides = frozenset({"score"}) async_boundary = True # everything from here runs in background max_workers = 3 # up to 3 background threads def __call__(self, ctx: StepContext) -> StepContext: time.sleep(0.5) # simulate slow work score = ctx.metadata["word_count"] * 10 return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "score": score}) ) pipe = Pipeline().then(Tokenize()).then(SlowScore()) # Returns immediately — only Tokenize runs in the foreground results = pipe.run([\ StepContext(sample="hello world"),\ StepContext(sample="background processing demo"),\ ]) # Background scoring still running... print(pipe.background_stats()) # {'active': 2, 'completed': 0} # Block until background work finishes pipe.wait_for_background(timeout=10.0) # Now results are fully populated for r in results: print(f"{r.sample}: score={r.output.metadata['score']}") hello world: score=20 background processing demo: score=30 See [Execution Model](https://kayba.ai/docs/pipeline/execution) for the full concurrency model — `workers` vs `max_workers`, async steps, and boundary rules. * * * Try it interactively -------------------- All the examples on this page (and more) are available as a runnable Jupyter notebook: [:material-notebook: Open the Pipeline Demo Notebook](https://github.com/kayba-ai/agentic-context-engine/blob/main/examples/pipeline_ex/pipeline_demo.ipynb) { .md-button } * * * Next steps ---------- * [**Core Concepts**](https://kayba.ai/docs/pipeline/core-concepts) — Understand the contract system and how validation works * [**Execution Model**](https://kayba.ai/docs/pipeline/execution) — Learn about async steps, `async_boundary`, and workers * [**Branching & Parallelism**](https://kayba.ai/docs/pipeline/branching) — Deep dive into merge strategies and error handling * [**Building Custom Steps**](https://kayba.ai/docs/pipeline/custom-steps) — Dependency injection, testing, and common patterns --- # Execution Model | Docs | Kayba Execution Model =============== "Async" means three different things in this framework. They operate at different levels and solve different problems. Keeping them separate is key to understanding the concurrency model. * * * Three types of concurrency -------------------------- | Type | Level | Problem it solves | | --- | --- | --- | | **Async steps** | single step | Don't block the thread during I/O | | **`async_boundary`** | across samples | Start the next sample before the current one finishes | | **Branch parallelism** | within one sample | Run independent work simultaneously on the same data | Each mechanism is independent. They compose freely — you can have async steps inside branches, behind an `async_boundary`, run with multiple workers. * * * Entry points: `run()` and `run_async()` --------------------------------------- ### Sync ```python # For regular (non-async) callers results = pipe.run(contexts, workers=4) ``` ### Async ```python # For async callers (e.g. inside an async framework) results = await pipe.run_async(contexts, workers=4) ``` `run()` is a thin wrapper that calls `asyncio.run(self.run_async(...))`. Both accept the same parameters and return `list[SampleResult]`. * * * Workers: sample-level parallelism --------------------------------- The `workers` parameter on `run()` / `run_async()` controls how many samples are processed through foreground steps simultaneously: import time from pipeline import Pipeline, StepContext class SlowStep: requires = frozenset() provides = frozenset({"result"}) def __call__(self, ctx: StepContext) -> StepContext: time.sleep(0.1) # Simulate expensive work return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "result": "done"}) ) pipe = Pipeline().then(SlowStep()) samples = [StepContext(sample=f"s{i}") for i in range(6)] # Sequential: 6 × 0.1s ≈ 0.6s results = pipe.run(samples, workers=1) # Parallel: 0.1s (all 6 run at once) results = pipe.run(samples, workers=6) Under the hood, `workers` creates an `asyncio.Semaphore` — at most N samples flow through the foreground steps at any given time. * * * Async steps — non-blocking I/O ------------------------------ A step that makes network calls (HTTP requests, API calls, subprocess) can be defined as a coroutine to avoid blocking the thread: ### Sync step ```python class FetchStep: requires = frozenset() provides = frozenset({"response"}) def __call__(self, ctx: StepContext) -> StepContext: response = requests.get(ctx.sample) # blocks the thread return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "response": response}) ) ``` ### Async step ```python class FetchStep: requires = frozenset() provides = frozenset({"response"}) async def __call__(self, ctx: StepContext) -> StepContext: async with aiohttp.ClientSession() as session: response = await session.get(ctx.sample) # yields the thread return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "response": response}) ) ``` The pipeline detects async steps automatically via `asyncio.iscoroutinefunction` and awaits them. Sync steps are wrapped with `asyncio.to_thread()` so they're safe in an async context too. > **Note** > > Async steps are about **not blocking the thread**, not about parallelism. The pipeline is still sequential — it just yields the thread during I/O waits. * * * Async boundary — fire-and-forget background ------------------------------------------- **Problem:** Some steps are slow (e.g. LLM calls for analysis). Waiting for them before starting the next sample hurts throughput. **Solution:** A step declares `async_boundary = True`. Everything from that step onward runs in a background thread. The pipeline loop moves to the next sample immediately. class SlowScoreStep: requires = frozenset({"tokens"}) provides = frozenset({"score"}) async_boundary = True # hand off to background from here max_workers = 3 # up to 3 scoring threads in parallel def __call__(self, ctx: StepContext) -> StepContext: time.sleep(0.5) # Expensive scoring score = len(ctx.metadata["tokens"]) * 10 return ctx.replace( metadata=MappingProxyType({**ctx.metadata, "score": score}) ) > **Indigo** = foreground (returns immediately) · **Blue** = background (fire-and-forget) Multiple samples flow through this simultaneously — sample 2 starts its foreground steps while sample 1's background steps are still running. ### Using the boundary pipe = Pipeline().then(Tokenize()).then(Uppercase()).then(SlowScoreStep()) # run() returns immediately after foreground steps (Tokenize + Uppercase) results = pipe.run(samples, workers=4) # Background scoring continues — results not yet populated print(pipe.background_stats()) # {'active': 3, 'completed': 1} # Block until all background work finishes pipe.wait_for_background(timeout=30.0) # Now all SampleResult.output fields are fully populated for r in results: print(r.output.metadata["score"]) ### Background pool model Each step **class** has a single shared `ThreadPoolExecutor`: * `SlowScoreStep.max_workers = 3` means one pool of 3 threads for all `SlowScoreStep` instances, regardless of how many pipelines are running * The pool is created lazily at first use and persists for the process lifetime * If two users need different concurrency limits for the same step type, they should subclass > **Boundary rules** > > * **One boundary per pipeline.** If multiple steps declare `async_boundary = True`, the pipeline raises `PipelineConfigError` at construction time. > * **No boundary inside Branch children.** A boundary inside a branch child raises `PipelineConfigError`. Branch children always block until joined — detaching mid-branch is incoherent. > * **Nested pipeline boundary is ignored.** When a pipeline is used as a step inside another pipeline, `async_boundary` is warned and ignored — there is no "next sample" to move to from the outer pipeline's perspective. * * * `workers` vs `max_workers` — independent pools ---------------------------------------------- These two knobs control different thread pools and do not interact: | Knob | Pool | Controls | | --- | --- | --- | | `pipe.run(contexts, workers=N)` | foreground pool | How many samples run through pre-boundary steps simultaneously | | `step.max_workers = K` | background pool (per step class) | How many instances of that step run in the background simultaneously | A sample leaves the foreground pool when it crosses the `async_boundary` and enters the background step's pool. **Mental model:** `workers` controls throughput _into_ the pipeline; `max_workers` controls throughput _through_ each background step. > **Rate limits** > > `workers` and `max_workers` are independent pools, but total concurrent outbound calls = foreground calls + background calls. With `workers=4` and `max_workers=3`, up to 7 requests may be in-flight simultaneously. Account for this when configuring per-provider rate limits. * * * Rule of thumb ------------- | Question | Answer | | --- | --- | | Does the step wait on I/O? | `async def __call__` | | Do I want to process more samples while previous ones are still in background steps? | `async_boundary = True` on the handoff step | | Can two steps on the same sample run simultaneously? | [`Branch`](https://kayba.ai/docs/pipeline/branching) | | Do I want N samples going through the pipeline at the same time? | `workers=N` on `run()` | --- # Documentation | Kayba Documentation ============= Learn how to use the Kayba framework to build self-improving AI agents. Getting Started --------------- [Installation\ \ Read documentation →](https://kayba.ai/docs/getting-started/installation) [Quick Start\ \ Read documentation →](https://kayba.ai/docs/getting-started/quick-start) Concepts -------- [How ACE Works\ \ Read documentation →](https://kayba.ai/docs/concepts/overview) [The Skillbook\ \ Read documentation →](https://kayba.ai/docs/concepts/skillbook) [Three Roles\ \ Read documentation →](https://kayba.ai/docs/concepts/roles) [Insight Levels\ \ Read documentation →](https://kayba.ai/docs/concepts/insight-levels) [Update Operations\ \ Read documentation →](https://kayba.ai/docs/concepts/updates) Guides ------ [Full Pipeline\ \ Read documentation →](https://kayba.ai/docs/guides/full-pipeline) [Integration Pattern\ \ Read documentation →](https://kayba.ai/docs/guides/integration) [Prompt Engineering\ \ Read documentation →](https://kayba.ai/docs/guides/prompts) [Async Learning\ \ Read documentation →](https://kayba.ai/docs/guides/async-learning) [Testing\ \ Read documentation →](https://kayba.ai/docs/guides/testing) Pipeline Engine --------------- [Overview\ \ Read documentation →](https://kayba.ai/docs/pipeline/index) [Quick Start\ \ Read documentation →](https://kayba.ai/docs/pipeline/quick-start) [Core Concepts\ \ Read documentation →](https://kayba.ai/docs/pipeline/core-concepts) [Execution Model\ \ Read documentation →](https://kayba.ai/docs/pipeline/execution) [Branching & Parallelism\ \ Read documentation →](https://kayba.ai/docs/pipeline/branching) [Error Handling\ \ Read documentation →](https://kayba.ai/docs/pipeline/error-handling) [Building Custom Steps\ \ Read documentation →](https://kayba.ai/docs/pipeline/custom-steps) [API Reference\ \ Read documentation →](https://kayba.ai/docs/pipeline/api-reference) Integrations ------------ [Overview\ \ Read documentation →](https://kayba.ai/docs/integrations/index) [LiteLLM\ \ Read documentation →](https://kayba.ai/docs/integrations/litellm) [LangChain\ \ Read documentation →](https://kayba.ai/docs/integrations/langchain) [Browser-Use\ \ Read documentation →](https://kayba.ai/docs/integrations/browser-use) [Claude Code\ \ Read documentation →](https://kayba.ai/docs/integrations/claude-code) [Opik Observability\ \ Read documentation →](https://kayba.ai/docs/integrations/opik) API Reference ------------- [API Reference\ \ Read documentation →](https://kayba.ai/docs/api/index) --- # API Reference | Docs | Kayba API Reference ============= Complete reference for all public classes, methods, and enums in the pipeline engine. * * * `pipeline.context` ------------------ ### `StepContext` Frozen dataclass passed from step to step. The pipeline engine only reads `sample` and `metadata` — domain-specific fields are added by subclassing. @dataclass(frozen=True) class StepContext: sample: Any = None metadata: MappingProxyType = field( default_factory=lambda: MappingProxyType({}) ) | Method | Signature | Description | | --- | --- | --- | | `replace` | `(**changes: Any) -> StepContext` | Return a new context with the given fields replaced. Uses `dataclasses.replace` internally. | **Behavior:** * `metadata` is auto-coerced from `dict` to `MappingProxyType` in `__post_init__` * Subclasses inherit `.replace()` — it works on all fields including subclass-defined ones * * * `pipeline.protocol` ------------------- ### `StepProtocol` Structural protocol that every step (and Pipeline/Branch) must satisfy. @runtime_checkable class StepProtocol(Protocol): requires: AbstractSet[str] provides: AbstractSet[str] def __call__(self, ctx: StepContext) -> StepContext: ... | Attribute | Type | Description | | --- | --- | --- | | `requires` | `AbstractSet[str]` | Metadata keys the step reads | | `provides` | `AbstractSet[str]` | Metadata keys the step writes | | `__call__` | `(StepContext) -> StepContext` | Execute the step | **Notes:** * `AbstractSet[str]` accepts both `set` and `frozenset` * `@runtime_checkable` enables `isinstance(step, StepProtocol)` checks * * * ### `SampleResult` Outcome for one sample after the pipeline has run. @dataclass class SampleResult: sample: Any output: StepContext | None error: Exception | None failed_at: str | None cause: Exception | None = None | Field | Type | Description | | --- | --- | --- | | `sample` | `Any` | The original input sample | | `output` | `StepContext \| None` | Final context (`None` if any step failed) | | `error` | `Exception \| None` | The exception (`None` if succeeded) | | `failed_at` | `str \| None` | Class name of the step that raised (`None` if succeeded) | | `cause` | `Exception \| None` | Inner exception for `BranchError` failures (default `None`) | **Notes:** * Mutable — background threads update it in-place when background steps complete * For background steps, `output`/`error` may be `None` until `wait_for_background()` completes * * * `pipeline.pipeline` ------------------- ### `Pipeline` Ordered sequence of steps. Satisfies `StepProtocol` — can be nested inside other pipelines. #### Constructor Pipeline(steps: list | None = None) | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `steps` | `list \| None` | `None` | Optional initial list of steps | Validates step ordering and infers contracts at construction time. #### Attributes | Attribute | Type | Description | | --- | --- | --- | | `requires` | `frozenset[str]` | Fields the pipeline needs from external context (auto-inferred) | | `provides` | `frozenset[str]` | Fields the pipeline writes (auto-inferred, union of all steps) | #### Methods ##### `then` def then(self, step: object) -> Pipeline Append a step and return `self` for chaining. Validates ordering immediately. | Parameter | Type | Description | | --- | --- | --- | | `step` | `object` | Any object satisfying `StepProtocol` | **Returns:** `self` (for method chaining) **Raises:** `PipelineOrderError` if the step requires a field produced by a later step * * * ##### `branch` def branch( self, *pipelines: object, merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT, ) -> Pipeline Append a `Branch` step and return `self` for chaining. Shorthand for `.then(Branch(*pipelines, merge=merge))`. **Returns:** `self` (for method chaining) * * * ##### `run` def run( self, contexts: Iterable[StepContext], workers: int = 1, ) -> list[SampleResult] Process contexts through the pipeline (sync entry point). | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `contexts` | `Iterable[StepContext]` | — | Input contexts to process | | `workers` | `int` | `1` | Max concurrent samples in foreground steps | **Returns:** `list[SampleResult]` — one result per input context **Notes:** Calls `asyncio.run(self.run_async(...))` internally. For background steps, call `wait_for_background()` after this returns. * * * ##### `run_async` async def run_async( self, contexts: Iterable[StepContext], workers: int = 1, ) -> list[SampleResult] Async entry point. Use `await pipe.run_async(contexts)` from coroutine contexts. Same parameters and return type as `run()`. * * * ##### `__call__` def __call__(self, ctx: StepContext) -> StepContext Run all steps sequentially on a single context. Used when the pipeline is nested as a step inside another pipeline. **Notes:** `async_boundary` markers are ignored in this mode — all steps run to completion. * * * ##### `wait_for_background` def wait_for_background(self, timeout: float | None = None) -> None Block until all background tasks complete. | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `timeout` | `float \| None` | `None` | Max seconds to wait. `None` = wait indefinitely. | **Raises:** `TimeoutError` if timeout elapses before completion * * * ##### `background_stats` def background_stats(self) -> dict[str, int] Return a snapshot of background task progress. Thread-safe. **Returns:** `{"active": int, "completed": int}` * * * `pipeline.branch` ----------------- ### `MergeStrategy` Enum of built-in merge strategies for `Branch` outputs. class MergeStrategy(Enum): RAISE_ON_CONFLICT = "raise_on_conflict" LAST_WRITE_WINS = "last_write_wins" NAMESPACED = "namespaced" | Value | Behavior | | --- | --- | | `RAISE_ON_CONFLICT` | Raises `ValueError` if two branches write different values to the same named field. Metadata merges with last-writer-wins. | | `LAST_WRITE_WINS` | Last branch's value wins for every conflicting field. | | `NAMESPACED` | Each branch's output stored at `metadata["branch_N"]`. No conflict possible. | * * * ### `Branch` Runs multiple pipelines in parallel, then merges their outputs. Satisfies `StepProtocol`. #### Constructor Branch( *pipelines: object, merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT, ) | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `*pipelines` | `object` | — | Child pipelines to run in parallel (at least one required) | | `merge` | `MergeStrategy \| Callable` | `RAISE_ON_CONFLICT` | Merge strategy or custom `fn(list[StepContext]) -> StepContext` | **Raises:** `ValueError` if no pipelines are provided #### Attributes | Attribute | Type | Description | | --- | --- | --- | | `requires` | `frozenset[str]` | Union of all children's requires | | `provides` | `frozenset[str]` | Union of all children's provides | | `pipelines` | `list` | The child pipelines | #### Methods ##### `__call__` def __call__(self, ctx: StepContext) -> StepContext Sync fan-out via `ThreadPoolExecutor`. All branches run to completion before any failure is raised. **Raises:** `BranchError` if any branch fails * * * ##### `__call_async__` async def __call_async__(self, ctx: StepContext) -> StepContext Async fan-out via `asyncio.gather`. Sync children are wrapped with `asyncio.to_thread`. **Raises:** `BranchError` if any branch fails * * * `pipeline.errors` ----------------- ### `PipelineOrderError` class PipelineOrderError(Exception): ... A step requires a field that no earlier step provides (but a later step does). Raised at **construction time**. * * * ### `PipelineConfigError` class PipelineConfigError(Exception): ... Invalid pipeline wiring. Raised at **construction time**. Examples: * More than one `async_boundary = True` step in the same pipeline * An `async_boundary = True` step inside a `Branch` child * * * ### `BranchError` class BranchError(Exception): failures: list[BaseException] One or more branch pipelines failed. All branches always run to completion before this is raised. Raised at **runtime**. | Attribute | Type | Description | | --- | --- | --- | | `failures` | `list[BaseException]` | One exception per failed branch | * * * Step class attributes --------------------- Optional attributes a step class can declare to control pipeline behavior: | Attribute | Type | Default | Description | | --- | --- | --- | --- | | `requires` | `set[str] \| frozenset[str]` | _(required)_ | Metadata keys the step reads | | `provides` | `set[str] \| frozenset[str]` | _(required)_ | Metadata keys the step writes | | `async_boundary` | `bool` | `False` | Marks the foreground/background split point | | `max_workers` | `int` | `1` | Max concurrent background threads for this step class | --- # Overview | Docs | Kayba Integrations Overview ===================== ACE provides runners for popular agentic frameworks. Each runner adds self-improving learning to an existing agent with minimal code changes. Available Integrations ---------------------- | Runner | Framework | Input | Insight Level | | --- | --- | --- | --- | | [`ACELiteLLM`](https://kayba.ai/docs/integrations/litellm) | LiteLLM (100+ providers) | Questions | Micro | | [`LangChain`](https://kayba.ai/docs/integrations/langchain) | LangChain Runnables | Chain inputs | Meso | | [`BrowserUse`](https://kayba.ai/docs/integrations/browser-use) | browser-use | Task strings | Meso | | [`ClaudeCode`](https://kayba.ai/docs/integrations/claude-code) | Claude Code CLI | Task strings | Meso | | [OpenClaw](https://kayba.ai/docs/integrations/openclaw) | OpenClaw transcripts | JSONL trace files | Meso | | [MCP Server](https://kayba.ai/docs/integrations/mcp) | MCP (stdio) | Tool calls | Micro | | [Opik](https://kayba.ai/docs/integrations/opik) | Opik observability | — | Monitoring | The Pattern ----------- All integration runners follow the same three-step pattern: 1. INJECT — Add skillbook strategies to the agent's context 2. EXECUTE — Run the external agent normally 3. LEARN — Reflector + SkillManager update the skillbook Quick Construction ------------------ Every runner offers a `from_model()` factory that builds ACE roles automatically: from ace_next import BrowserUse, LangChain, ClaudeCode # Browser automation browser = BrowserUse.from_model(browser_llm=my_llm, ace_model="gpt-4o-mini") # LangChain chain/agent chain = LangChain.from_model(my_runnable, ace_model="gpt-4o-mini") # Claude Code CLI coder = ClaudeCode.from_model(working_dir="./project", ace_model="gpt-4o-mini") Shared Features --------------- All runners share these capabilities: * **Skillbook persistence** — `save()` / load via `skillbook_path` * **Checkpointing** — automatic saves during long runs * **Deduplication** — prevent duplicate skills * **Background learning** — `wait=False` for async learning * **Progress tracking** — `learning_stats` property Which Integration Should I Use? ------------------------------- * **Building a Q&A or reasoning agent?** Use [ACELiteLLM](https://kayba.ai/docs/integrations/litellm) * **Have an existing LangChain chain or agent?** Use [LangChain](https://kayba.ai/docs/integrations/langchain) * **Automating browser tasks?** Use [BrowserUse](https://kayba.ai/docs/integrations/browser-use) * **Running coding tasks with Claude Code?** Use [ClaudeCode](https://kayba.ai/docs/integrations/claude-code) * **Want to monitor costs and traces?** Add [Opik](https://kayba.ai/docs/integrations/opik) * **Learning from OpenClaw session transcripts?** Use [OpenClaw](https://kayba.ai/docs/integrations/openclaw) * **Exposing ACE as an MCP tool provider?** Use the [MCP Server](https://kayba.ai/docs/integrations/mcp) * **Using a different framework?** See the [Integration Guide](https://kayba.ai/docs/guides/integration) to build a custom runner --- # Documentation | Docs | Kayba Page not found ============== This documentation page could not be found. [Back to docs](https://kayba.ai/docs) --- # Documentation | Docs | Kayba Page not found ============== This documentation page could not be found. [Back to docs](https://kayba.ai/docs) ---