agentflow

Lightweight multi-agent AI pipeline framework. Define agents with decorators, give them tools, wire them into a DAG, and run independent stages in parallel, with built-in cost tracking, caching, timeouts, streaming, and observability.

• Tool / function calling: @tool turns any Python function into an LLM tool; agents run a bounded ReAct loop
• Parallel execution: agents with no inter-dependencies run concurrently; a failing level cancels its siblings instead of burning tokens
• Cost tracking: per-agent and per-pipeline USD cost, and the run tells you which models it could not price
• Cost budgets: budget_usd= aborts a run at a hard ceiling and hands back the results you already paid for
• Typed output that repairs itself: declare a Pydantic schema; agentflow prompts with it, validates, and asks the model to fix a bad reply
• Token streaming: LLM.astream() yields tokens for interactive UIs
• Decorator-based: define agents as plain async functions, no boilerplate
• LLM response caching: in-memory (or Redis) cache cuts cost on repeated runs
• Per-agent timeouts & retries: timeout= and pipeline-level retry with exponential backoff + jitter
• Conditional branching: skip agents dynamically based on upstream outputs
• Explainable DAG: pipe.explain() prints the resolved levels without calling an LLM
• Observability: lifecycle Hooks, an OpenTelemetry adapter, and structured JSON logs with run IDs
• Provider agnostic: any OpenAI-compatible API (OpenAI, Groq, Together, Ollama, vLLM, OpenRouter)
• Fully typed: ships py.typed; passes mypy --strict
• Minimal deps: only openai + pydantic

📖 Documentation → · Public API & stability contract · Design decisions

Install

pip install agentflowkit

# Optional: Redis cache backend
pip install "agentflowkit[redis]"

The showcase: earnings-call triage

One run of examples/earnings_triage.py: a six-agent diamond DAG where a tool-calling fetcher feeds three analysts running in parallel, a risk synthesizer enforces a typed Pydantic schema, and the whole run sits under a hard USD budget. Works against any OpenAI-compatible endpoint (zero API keys with Ollama, free tier on Groq).

pipe = Pipeline(llm=llm, budget_usd=0.25)      # hard cost ceiling per run
pipe.add(transcript_fetcher)                   # ReAct tools: transcript + consensus
pipe.add(financials_analyst, depends_on=["transcript_fetcher"])  # ┐
pipe.add(sentiment_analyst,  depends_on=["transcript_fetcher"])  # ├ run in parallel
pipe.add(competitor_scanner, depends_on=["transcript_fetcher"])  # ┘
pipe.add(risk_synthesizer,   depends_on=["financials_analyst",   # output_schema=
                                         "sentiment_analyst",    #   RiskAssessment
                                         "competitor_scanner"])
pipe.add(brief_writer,       depends_on=["risk_synthesizer"])    # gets the validated dict

Representative output (python examples/earnings_triage.py with gpt-4o-mini):

━━━ Run 1: cold (real LLM calls) ━━━
  ▶ transcript_fetcher  (level 0)
  ✓ transcript_fetcher  1289 tok
  ▶ financials_analyst  (level 1)
  ▶ sentiment_analyst   (level 1)
  ▶ competitor_scanner  (level 1)
  ✓ sentiment_analyst   601 tok
  ✓ financials_analyst  644 tok
  ✓ competitor_scanner  589 tok
  ▶ risk_synthesizer    (level 2)
  ✓ risk_synthesizer    512 tok
  ▶ brief_writer        (level 3)
  ✓ brief_writer        418 tok

  wall time: 11.4s  (agent time summed: 27.9s, parallelism won 16.5s back)
  total cost: $0.001210

━━━ Run 2: warm (response cache) ━━━
  ✓ ... [cache hit] ×6

  wall time: 0.1s
  total cost: $0.000000        ← cache hits bill $0

Architecture

Independent agents at the same DAG level execute concurrently. Dependent agents wait for their prerequisite level to complete before starting.

graph TD
    T[Task Input] --> L0["Level 0: Parallel"]
    L0 --> A1[researcher]
    L0 --> A2[fact_checker]
    A1 --> L1[Level 1]
    A2 --> L1
    L1 --> A3[writer]
    A3 --> R[PipelineResult]

Quick Start

import asyncio
from agentflow import Agent, Pipeline, LLM

llm = LLM(
    model="llama-3.3-70b-versatile",
    base_url="https://api.groq.com/openai/v1",
    api_key="your-groq-key",  # Free at console.groq.com
)

@Agent(name="researcher", role="Research Analyst")
async def researcher(task: str, context: dict) -> str:
    return f"Research this topic thoroughly: {task}"

@Agent(name="fact_checker", role="Fact Checker")
async def fact_checker(task: str, context: dict) -> str:
    return f"Find key facts and statistics about: {task}"

@Agent(name="writer", role="Content Writer")
async def writer(task: str, context: dict) -> str:
    research = context["researcher"]
    facts = context["fact_checker"]
    return f"Write an article using:\nResearch: {research}\nFacts: {facts}"

# researcher and fact_checker run in parallel (Level 0)
# writer runs after both complete (Level 1)
pipe = Pipeline(llm=llm)
pipe.add(researcher)
pipe.add(fact_checker)
pipe.add(writer, depends_on=["researcher", "fact_checker"])

async def main():
    result = await pipe.run("AI in Healthcare")
    print(result.output)
    print(f"Run ID: {result.run_id} | Tokens: {result.total_tokens} | Cost: ${result.total_cost:.6f}")

asyncio.run(main())

Features

Tool / Function Calling

Give an agent tools and it becomes a ReAct agent: the model decides which functions to call, agentflow runs them, feeds results back, and repeats until a final answer. Schemas are generated from your type hints, so you never write JSON.

from agentflow import Agent, Pipeline, LLM, tool

@tool
def get_stock_price(ticker: str) -> dict:
    """Look up the latest price for a stock ticker."""
    return {"ticker": ticker, "price": 229.87}

@tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

@Agent(name="analyst", role="Financial Analyst", tools=[get_stock_price, multiply])
async def analyst(task: str, context: dict) -> str:
    return task

pipe = Pipeline(llm=llm)
pipe.add(analyst)
result = await pipe.run("What do 10 shares of AAPL cost?")

# Inspect the tool calls the model made:
for call in result.get("analyst").metadata["tool_calls"]:
    print(call["tool"], call["arguments"], "->", call["result"])

Sync and async tools both work (sync tools run in a thread). The loop is bounded by max_tool_iterations (default 6), and tool errors are fed back to the model to recover rather than crashing the run.

Parallel Execution

Agents with no declared dependencies on each other run concurrently at the same DAG level:

pipe.add(agent_a)               # Level 0
pipe.add(agent_b)               # Level 0 (runs in parallel with agent_a)
pipe.add(agent_c, depends_on=["agent_a", "agent_b"])  # Level 1

Benchmark: 3 parallel agents (0.5s each) → total time ~0.5s vs 1.5s sequential.

LLM Response Caching

Cache identical LLM calls to save tokens and speed up repeated runs:

from agentflow import LLM, InMemoryCache

cache = InMemoryCache(default_ttl=3600)  # 1-hour TTL
llm = LLM(model="gpt-4o-mini", api_key="...", cache=cache)

Redis backend (requires pip install "agentflowkit[redis]"):

from agentflow import LLM, RedisCache

llm = LLM(model="gpt-4o", cache=RedisCache(url="redis://localhost:6379/0"))

Cache hits appear in results: result.agents_with_cache_hits, agent_result.cached.

Cost Tracking

Every result carries an estimated USD cost from built-in per-model pricing:

result = await pipe.run("Summarize the news")
print(f"Agent cost:    ${result.get('summarizer').cost:.6f}")
print(f"Pipeline cost: ${result.total_cost:.6f}")

Prices use longest-prefix matching, so gpt-4o-2024-08-06 resolves to gpt-4o. Cache hits bill $0.00.

A model with no price entry costs $0.00 (a placeholder, not a measurement). That is never silent: the model is logged once, and the run tells you which models it could not price.

if result.unpriced_models:
    print(f"total_cost is an undercount: no prices for {result.unpriced_models}")

# Register prices for custom / self-hosted / newly-released models.
# This is the authoritative override; the bundled table is indicative and drifts.
from agentflow import register_price
register_price("my-finetuned-model", prompt_per_1m=0.50, completion_per_1m=1.50)

Cost Budgets

Put a hard ceiling on a run. The budget is checked after each DAG level, and the error hands back the work you already paid for instead of discarding it:

from agentflow import BudgetExceededError

pipe = Pipeline(llm=llm, budget_usd=0.25)
try:
    result = await pipe.run("Analyze the filing")
except BudgetExceededError as exc:
    print(f"stopped at ${exc.spent_usd} of ${exc.budget_usd}")
    result = exc.partial_result          # the levels that did complete
    print(result.results.keys())

When an agent fails, the rest of its level is cancelled rather than left to finish producing output that would be thrown away.

Inspecting the DAG

explain() renders the resolved graph without running anything or calling an LLM, and fails on a cycle or an unknown dependency exactly as run() would:

print(pipe.explain())

Pipeline: 6 agents, 4 levels, max 3 concurrent
Level 0 (1 agent):
  transcript_fetcher  role=Fetcher
Level 1 (3 agents, run in parallel):
  financials_analyst  role=Financials  after=[transcript_fetcher]
  sentiment_analyst   role=Sentiment  after=[transcript_fetcher]
  competitor_scanner  role=Competitors  after=[transcript_fetcher]  timeout=30s
Level 2 (1 agent):
  risk_synthesizer  role=Risk  after=[competitor_scanner, financials_analyst, sentiment_analyst]
Level 3 (1 agent):
  brief_writer  role=Writer  after=[risk_synthesizer]  conditional (may be skipped at run time)

Limiting Concurrency

Without a cap, a level of 40 agents opens 40 concurrent LLM calls:

pipe = Pipeline(llm=llm, max_concurrency=5)   # at most 5 agents in flight per run

Token Streaming

Stream a completion token-by-token for interactive UIs:

messages = [{"role": "user", "content": "Explain async pipelines in one line."}]
async for token in llm.astream(messages):
    print(token, end="", flush=True)

Observability

Pipeline.run() is silent by default. Pass Hooks to observe the full lifecycle and bridge to logging, metrics, OpenTelemetry, or Langfuse:

from agentflow import Pipeline, LoggingHooks

pipe = Pipeline(llm=llm, hooks=LoggingHooks("research-pipeline"))
result = await pipe.run("AI in Healthcare")
# → {"event": "agent_complete", "agent": "writer", "tokens": 812, "cached": false, ...}

Subclass Hooks and override on_agent_start / on_agent_end / … to emit spans to your own backend. A hook that raises is caught and warned, never crashing the run.

Per-Agent Timeouts

Protect against slow or hung LLM calls:

pipe.add(slow_agent, timeout=10.0)   # raises AgentTimeoutError after 10s

Conditional Branching

Dynamically route execution based on upstream agent outputs:

pipe.add(classifier)

pipe.add(
    urgent_handler,
    depends_on=["classifier"],
    condition=lambda ctx: "urgent" in ctx["classifier"].lower(),
)
pipe.add(
    standard_handler,
    depends_on=["classifier"],
    condition=lambda ctx: "urgent" not in ctx["classifier"].lower(),
)

Skipped agents emit agent_skipped events in streaming mode.

Pipeline Retry

Automatically retry transient agent failures with exponential backoff:

pipe = Pipeline(llm=llm, retry_failed_agents=2)  # up to 2 retries: 1s, 2s

Structured Output Validation

Declare a Pydantic schema and agentflow handles the rest: the schema is sent to the model, the reply is validated, and a malformed reply is repaired rather than fatal:

from pydantic import BaseModel

class Report(BaseModel):
    title: str
    summary: str
    confidence: float

@Agent(name="analyst", role="Data Analyst", output_schema=Report)
async def analyst(task: str, context: dict) -> str:
    return f"Analyze this: {task}"      # no need to describe the schema yourself

# The validated output flows downstream: agents depending on "analyst"
# receive the validated dict in context["analyst"], and it's also on
# result.get("analyst").data

If the model answers with something that does not validate, agentflow shows it the validation errors and asks for a correction (output_retries=1 by default, 0 to disable). Repairs are real LLM calls, so they are billed to the agent and count against the budget. Responses wrapped in a ```json fence are unwrapped locally, for free.

This works identically on every OpenAI-compatible endpoint because the schema travels in the prompt. If you want a provider's native JSON mode instead, pass it yourself; LLM.generate() forwards any extra keyword to the provider:

await llm.generate(messages, response_format={"type": "json_object"}, seed=42)

Rate Limiting

Throttle API calls for rate-limited providers:

from agentflow import LLM, RateLimiter

limiter = RateLimiter(requests_per_minute=60, max_concurrent=5)
llm = LLM(model="gpt-4o-mini", api_key="...", rate_limiter=limiter)

Event Streaming

Real-time pipeline monitoring:

async for event in pipe.stream("AI in Healthcare"):
    match event.type:
        case "agent_start":
            print(f"▶ {event.agent} (level {event.data['level']})")
        case "agent_complete":
            print(f"✓ {event.agent}: {event.data['tokens']} tokens, cached={event.data['cached']}")
        case "agent_skipped":
            print(f"⏭ {event.agent} skipped")
        case "pipeline_complete":
            print(f"Done: {event.data['total_tokens']} tokens across {event.data['levels_executed']} levels")

Structured Logging

Production-ready JSON logging with run IDs:

from agentflow import PipelineLogger

log = PipelineLogger("research-pipeline", run_id=result.run_id)
log.log_pipeline_complete(result.run_id, result.total_tokens, result.total_duration)
# → {"timestamp": "...", "level": "INFO", "event": "pipeline_complete", "run_id": "a1b2c3d4", ...}

When to use agentflow (and when not to)

agentflow is a deliberately narrow library, not a framework. It covers one problem well: running typed, tool-using agents as a parallel DAG on any OpenAI-compatible API, with the operational basics (retries, timeouts, caching, cost tracking, streaming, hooks) built in rather than bolted on.

What that buys you:

• Two runtime dependencies (openai, pydantic). Optional extras pull in Redis, Docker, or MQTT only if you use those features.
• Auditability. The core is small enough to read in a sitting before you put it in production, and it ships py.typed with mypy --strict clean.
• Async-native design. Everything is async from the ground up; parallelism is asyncio.gather() on DAG levels, not threads or callbacks.
• A short learning curve. Two decorators (@Agent, @tool) and a Pipeline are the whole public surface for most programs.
• The boilerplate you were going to write anyway. Against the honest baseline of hand-rolling asyncio.gather(), agentflow is the ~2,000 lines of retries with Retry-After, cost tables, budgets, caching, timeouts, and event plumbing you'd otherwise write under deadline, already typed and covered by ~250 tests.

What agentflow deliberately does not do (reach for LangChain, CrewAI, or similar frameworks if you need these):

• No prompt-template library, document loaders, or vector-store integrations.
• No agent marketplace or prebuilt personas; you write the agents.
• No graph persistence / resumable long-running workflows across processes.
• No provider abstraction beyond OpenAI-compatible endpoints (OpenAI, Groq, OpenRouter, Ollama, vLLM, etc. all work; Bedrock-style native SDKs don't).

If your project already lives inside a larger framework's ecosystem, use that ecosystem. agentflow is for engineers who want a foundation they can read, type-check, and own.

Class-Based Agents

For agents with custom logic beyond prompt construction:

from agentflow import BaseAgent, AgentResult

class DatabaseAgent(BaseAgent):
    def __init__(self, db_connection):
        super().__init__(name="db_agent", role="Database Analyst")
        self.db = db_connection

    async def execute(self, task: str, context: dict, llm) -> AgentResult:
        # Fetch real data, then ask LLM to analyze it
        data = await self.db.query(task)
        response = await llm.generate([
            {"role": "system", "content": f"You are a {self.role}."},
            {"role": "user", "content": f"Analyze this data: {data}\nTask: {task}"},
        ])
        return AgentResult(
            agent=self.name,
            output=response.content,
            tokens_used=response.tokens,
            duration=response.duration,
        )

Supported Providers

# OpenAI
llm = LLM(model="gpt-4o-mini", api_key="sk-...")

# Groq (free tier available)
llm = LLM(model="llama-3.3-70b-versatile",
           base_url="https://api.groq.com/openai/v1",
           api_key="gsk_...")

# Ollama (local, no API key)
llm = LLM(model="llama3.2", base_url="http://localhost:11434/v1", api_key="ollama")

# Together AI
llm = LLM(model="meta-llama/Llama-3-70b-chat-hf",
           base_url="https://api.together.xyz/v1",
           api_key="...")

Examples

• examples/earnings_triage.py: the showcase, 6-agent diamond DAG with tools, parallel analysts, typed output, budget, and cache
• examples/tool_agent.py: ReAct agent that calls tools (calculator + stock lookup)
• examples/streaming_and_cost.py: token streaming + USD cost tracking
• examples/research_crew.py: 3-agent sequential research pipeline
• examples/code_reviewer.py: 2-agent code review pipeline
• examples/market_analysis_crew.py: 5-agent parallel market analysis (diamond DAG)
• examples/memory_chat_agents.py: two agents sharing context across separate runs via memory
• examples/research_react_agent.py: single ReAct agent researching with tools
• examples/cpp_build_pipeline.py: write / compile / test loop driving a real toolchain
• examples/robotics_mqtt_agent.py: Pipeline.serve() daemon fed by an MQTT trigger (needs the mqtt extra)
• examples/drone_telemetry_agent.py: MQTTDaemon with a Pydantic-validated trigger policy (needs the mqtt extra)
• benchmarks/parallel_speedup.py: measured parallel vs. sequential speedup (~2×)

Every example is import-tested in CI, so they cannot drift away from the API.

Contributing

See CONTRIBUTING.md for development setup, coding style, and PR requirements.

Changelog

See CHANGELOG.md.

License

MIT