ahzan-agentforge

Production-grade AI agent framework — the Next.js for AI agents. Python SDK.

AgentForge owns the agent loop — LLM calls, tool execution, state checkpointing, crash recovery, cost governance, rollback, and full run tracing. You write tools and prompts. Everything else is handled.

Also available as a TypeScript package: @ahzan-agentforge/core

Install

pip install ahzan-agentforge

With LLM providers:

pip install ahzan-agentforge[anthropic]   # Anthropic Claude
pip install ahzan-agentforge[openai]      # OpenAI GPT
pip install ahzan-agentforge[all]         # All providers + Redis

Quick Start

import asyncio
from agentforge import define_agent, define_tool, create_llm
from pydantic import BaseModel

class OrderInput(BaseModel):
    order_id: str

class OrderOutput(BaseModel):
    id: str
    status: str
    total: float

get_order = define_tool(
    name="get_order",
    description="Fetch order details by ID",
    input_schema=OrderInput,
    output_schema=OrderOutput,
    execute=lambda inp: OrderOutput(id=inp.order_id, status="shipped", total=49.99),
)

agent = define_agent(
    name="support-agent",
    tools=[get_order],
    llm=create_llm(provider="openai", model="gpt-4o"),
    max_steps=15,
    system_prompt="You are a customer support agent. Look up orders and help customers.",
)

async def main():
    result = await agent.run(task="Order #4521 arrived damaged")
    print(result.status)   # 'completed'
    print(result.output)

asyncio.run(main())

Features

• Execution engine with checkpointing and crash recovery
• Pydantic schema validation on tool inputs and outputs
• LLM providers: Anthropic Claude, OpenAI GPT, Ollama (local)
• Token-level streaming via agent.stream()
• State stores: in-memory (dev) and Redis (production)
• Long-term memory: in-memory and PgVector with embedding retrieval
• Budget governor with token/cost limits and model pricing
• Autonomy policy with per-tool allow/deny/escalate rules
• Rollback with compensating actions
• Multi-agent coordination: pipeline, parallel, supervisor, debate patterns
• OpenTelemetry observability — every decision traced
• MockLLM and TestHarness for testing without API calls

LLM Providers

from agentforge import create_llm

llm = create_llm(provider="anthropic", model="claude-sonnet-4-20250514")
llm = create_llm(provider="openai", model="gpt-4o")
llm = create_llm(provider="ollama", model="llama3.1")  # local, zero cost

Streaming

async for event in agent.stream(task="Help customer"):
    if event.type == "llm_token":
        print(event.content, end="", flush=True)
    elif event.type == "tool_start":
        print(f"\nCalling {event.tool_name}...")
    elif event.type == "done":
        print(f"\nResult: {event.result.output}")

Multi-Agent Coordination

from agentforge import pipeline, parallel, supervisor, debate

result = await pipeline([researcher, writer, editor], task="Write a report")
result = await supervisor(manager, {"researcher": researcher, "writer": writer}, task="Complete project")
result = await parallel([agent1, agent2], task="Analyze data", merge_fn=merge_results)
result = await debate([optimist, pessimist], task="Evaluate proposal", judge=judge)

Testing

from agentforge import create_mock_llm, create_test_harness

harness = create_test_harness(
    agent=my_agent_config,
    llm=create_mock_llm(responses=[
        {"tool_calls": [{"name": "get_order", "input": {"order_id": "4521"}}]},
        {"text": "Order found and ticket created."},
    ]),
)

result = await harness.run(task="Handle order issue")
assert result.status == "completed"
assert len(result.tool_calls("get_order")) == 1

API Parity with TypeScript

Requirements

• Python >= 3.10
• Redis (optional, for persistent state)
• PostgreSQL + pgvector (optional, for long-term memory)

License

MIT