continuum-recipes

Continuum Recipes Skill

Common, ready-to-paste patterns. Each is verified against framework 0.2.0.

1. RAG-augmented agent

from continuum.agent import BaseAgent, AgentRunner
from continuum.agent.config import AgentConfig

retrieved_docs = await my_retriever.search(query)
rag_text = "\n\n".join(d.text for d in retrieved_docs[:5])

agent = BaseAgent(
    name="rag-agent",
    instructions="Answer ONLY using the PROVIDED CONTEXT. If unsure, say so.",
    config=AgentConfig(rag_context=rag_text, require_context=True),
)
resp = await AgentRunner().run(agent, query, user_id="u1")

2. Plan-and-execute (orchestrator + executor)

from pydantic import BaseModel
from typing import Literal

class ToolStep(BaseModel):
    step_id: str
    tool_name: str
    parameters: dict
    instruction: str
    depends_on: list[str] | None = None

class ExecutionPlan(BaseModel):
    intent: Literal["search", "checkout", "support", "other"]
    respond_directly: bool = False
    direct_response: str | None = None
    steps: list[ToolStep] = []
    user_context: str | None = None
    response_instructions: str = "Be concise."

orchestrator = BaseAgent(
    name="orchestrator",
    instructions=("Analyze the request and emit an ExecutionPlan as JSON. "
                  "If you can answer directly, set respond_directly=true."),
    output_schema=ExecutionPlan,
    model="gpt-4o-mini",
)
executor = BaseAgent(
    name="executor",
    instructions="Execute the plan steps in order using the available tools.",
    mcp_servers=[mcp],
    model="gpt-4o-mini",
)

runner = AgentRunner(agent_registry={"orchestrator": orchestrator, "executor": executor})
plan_resp = await runner.run(orchestrator, user_msg, session_id=sid, user_id=uid)
plan: ExecutionPlan = plan_resp.structured_output

if plan.respond_directly:
    return plan.direct_response
exec_resp = await runner.run(executor, format_plan_for_executor(plan),
                             session_id=sid, user_id=uid)
return exec_resp.content