DEV Community

Thesius Code
Thesius Code

Posted on • Originally published at datanest-stores.pages.dev

AI Agent Framework

#ai #llm #machinelearning #python

AI Agent Framework

Build production-grade multi-agent systems that can reason, plan, use tools, and collaborate. This framework provides battle-tested patterns for orchestrating LLM-powered agents — from simple single-agent tool callers to complex multi-agent workflows with human oversight. Stop reinventing agent infrastructure and focus on your domain logic.

Key Features

  • Multi-Agent Orchestration — Route tasks between specialized agents with configurable delegation strategies (round-robin, capability-based, auction)
  • Tool Calling Engine — Register Python functions as tools with automatic schema generation, input validation, and retry logic
  • Persistent Memory — Short-term conversation buffers and long-term vector-backed memory with configurable retention policies
  • Planning Loops — ReAct, Plan-and-Execute, and Tree-of-Thought planning patterns with step-level observability
  • Human-in-the-Loop Gates — Configurable approval checkpoints before high-risk actions (API calls, data mutations, external communication)
  • Structured Output Parsing — Enforce JSON schemas on agent responses with automatic retry on malformed output
  • Conversation Branching — Fork and merge conversation threads for parallel agent exploration
  • Execution Tracing — Full trace logs for every agent step, tool call, and decision point

Quick Start 

from agent_framework import Agent, ToolRegistry, Orchestrator

# 1. Define tools
registry = ToolRegistry()

@registry.tool(description="Search the knowledge base for relevant documents")
def search_docs(query: str, top_k: int = 5) -> list[dict]:
    # Your retrieval logic here
    return [{"title": "Result", "content": "..."}]

@registry.tool(description="Send a formatted email to a recipient")
def send_email(to: str, subject: str, body: str) -> dict:
    # Requires human approval (configured below)
    return {"status": "sent", "to": to}

# 2. Create agents with different capabilities
researcher = Agent(
    name="researcher",
    system_prompt="You find and synthesize information from the knowledge base.",
    tools=[search_docs],
    model="gpt-4o",
)

writer = Agent(
    name="writer",
    system_prompt="You draft professional emails based on research findings.",
    tools=[send_email],
    model="gpt-4o",
    require_approval=["send_email"],  # Human gate on this tool
)

# 3. Orchestrate
orchestrator = Orchestrator(agents=[researcher, writer])
result = orchestrator.run("Research our Q3 metrics and email a summary to the team.")
print(result.final_output)
print(result.trace.summary())  # Shows all steps, tools called, tokens used
Enter fullscreen mode Exit fullscreen mode

Architecture 

┌─────────────────────────────────────────────┐
│                Orchestrator                  │
│  ┌─────────┐  ┌──────────┐  ┌───────────┐  │
│  │ Planner │  │ Router   │  │ Evaluator │  │
│  └────┬────┘  └────┬─────┘  └─────┬─────┘  │
│       │            │              │          │
│  ┌────▼────────────▼──────────────▼─────┐   │
│  │            Agent Pool                │   │
│  │  ┌─────┐  ┌─────┐  ┌─────┐         │   │
│  │  │ A_1 │  │ A_2 │  │ A_N │  ...    │   │
│  │  └──┬──┘  └──┬──┘  └──┬──┘         │   │
│  └─────┼────────┼────────┼─────────────┘   │
│        │        │        │                  │
│  ┌─────▼────────▼────────▼─────────────┐   │
│  │         Shared Services              │   │
│  │  Memory │ Tools │ Approval │ Trace   │   │
│  └──────────────────────────────────────┘   │
└─────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

The Planner decomposes tasks into subtasks. The Router assigns subtasks to the best-fit agent. The Evaluator checks if the final output satisfies the original request and triggers replanning if needed.

Usage Examples

ReAct Planning Loop 

from agent_framework import Agent, PlanningStrategy

agent = Agent(
    name="analyst",
    system_prompt="You analyze datasets and produce insights.",
    planning=PlanningStrategy.REACT,  # Thought -> Action -> Observation loop
    max_iterations=10,
)

result = agent.run("What were the top 3 revenue drivers last quarter?")
for step in result.trace.steps:
    print(f"[{step.type}] {step.content[:100]}")
Enter fullscreen mode Exit fullscreen mode

Multi-Agent Delegation 

from agent_framework import Orchestrator, DelegationStrategy

orchestrator = Orchestrator(
    agents=[researcher, writer, reviewer],
    strategy=DelegationStrategy.CAPABILITY_BASED,
    max_delegation_depth=3,  # Prevent infinite delegation loops
)
Enter fullscreen mode Exit fullscreen mode

Memory Configuration 

from agent_framework.memory import ConversationMemory, VectorMemory

agent = Agent(
    name="assistant",
    memory=[
        ConversationMemory(max_turns=20),           # Recent context
        VectorMemory(collection="long_term", top_k=5),  # Semantic recall
    ],
)
Enter fullscreen mode Exit fullscreen mode

Configuration 

# config.yaml
orchestrator:
  max_concurrent_agents: 4
  timeout_seconds: 120
  delegation_strategy: "capability_based"

agents:
  default_model: "gpt-4o"
  temperature: 0.1
  max_tokens: 4096
  retry_on_parse_failure: true
  max_retries: 3

memory:
  backend: "sqlite"            # sqlite | redis | postgres
  conversation_buffer_size: 20
  vector_store: "chromadb"
  embedding_model: "text-embedding-3-small"

approval:
  enabled: true
  timeout_seconds: 300         # Auto-reject after 5 minutes
  notify_channel: "slack"      # slack | email | console
  high_risk_tools:
    - "send_email"
    - "execute_sql"
    - "deploy_service"

tracing:
  enabled: true
  output: "logs/traces/"
  format: "json"               # json | opentelemetry
  include_prompts: false       # Set true only in dev (contains PII)
Enter fullscreen mode Exit fullscreen mode

Best Practices

  1. Start with a single agent — Add multi-agent orchestration only when a single agent demonstrably can't handle the task breadth.
  2. Gate destructive tools — Any tool that mutates state (DB writes, API calls, file deletions) should require human approval in production.
  3. Set iteration limits — Always configure max_iterations on planning loops to prevent runaway token consumption.
  4. Use structured outputs — Define Pydantic models for tool inputs/outputs to catch schema mismatches early.
  5. Log everything in development — Enable full tracing with include_prompts: true during development, disable in production.
  6. Test with deterministic seeds — Use temperature: 0 and fixed seeds during testing for reproducible agent behavior.
  7. Monitor token budgets — Set per-agent and per-orchestration token limits to avoid surprise costs during complex planning loops.

Troubleshooting

Problem Cause Fix
Agent loops forever without producing output Missing or too-high max_iterations Set max_iterations: 10 and add an evaluator to check completion
Tool calls fail with schema validation errors LLM generates malformed JSON arguments Enable retry_on_parse_failure: true and add examples to tool descriptions
Multi-agent tasks produce contradictory results Agents lack shared context Use shared VectorMemory or pass the orchestrator's scratchpad between agents
Human approval gate times out timeout_seconds too low or notification missed Increase timeout, add fallback notification channel, check Slack/email integration

This is 1 of 11 resources in the AI Builder Pro toolkit. Get the complete [AI Agent Framework] with all files, templates, and documentation for $59.

Get the Full Kit →

Or grab the entire AI Builder Pro bundle (11 products) for $169 — save 30%.

Get the Complete Bundle →

Related Articles

Top comments (0)