Metadata-Version: 2.4
Name: agentflowkit
Version: 0.7.0
Summary: Lightweight multi-agent AI pipeline framework with parallel DAG execution, tool calling, and cost tracking
Project-URL: Homepage, https://github.com/KaramQ6/agentflow
Project-URL: Repository, https://github.com/KaramQ6/agentflow
Project-URL: Changelog, https://github.com/KaramQ6/agentflow/blob/main/CHANGELOG.md
Project-URL: Bug Tracker, https://github.com/KaramQ6/agentflow/issues
Author: KaramQ6
License-Expression: MIT
License-File: LICENSE
Keywords: agent-framework,agents,ai,async,dag,function-calling,llm,multi-agent,parallel,pipeline,react,tool-calling
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: openai>=1.0.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: mypy>=1.9; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: docker
Requires-Dist: docker>=7.0; extra == 'docker'
Provides-Extra: docs
Requires-Dist: mkdocs-material>=9.5; extra == 'docs'
Requires-Dist: mkdocstrings[python]>=0.24; extra == 'docs'
Provides-Extra: mqtt
Requires-Dist: aiomqtt>=2.0; extra == 'mqtt'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == 'otel'
Requires-Dist: opentelemetry-sdk>=1.20; extra == 'otel'
Provides-Extra: redis
Requires-Dist: chromadb>=0.4; extra == 'redis'
Requires-Dist: redis>=5.0; extra == 'redis'
Description-Content-Type: text/markdown

# agentflow

[![PyPI version](https://badge.fury.io/py/agentflowkit.svg)](https://pypi.org/project/agentflowkit/)
[![CI](https://github.com/KaramQ6/agentflow/actions/workflows/ci.yml/badge.svg)](https://github.com/KaramQ6/agentflow/actions/workflows/ci.yml)
[![codecov](https://codecov.io/gh/KaramQ6/agentflow/branch/main/graph/badge.svg)](https://codecov.io/gh/KaramQ6/agentflow)
[![PyPI Downloads](https://img.shields.io/pypi/dm/agentflowkit)](https://pypi.org/project/agentflowkit/)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT)

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 →](docs/index.md)** · [Public API & stability contract](PUBLIC_API.md) · [Design decisions](docs/adr/)

## Install

```bash
pip install agentflowkit

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

## The showcase: earnings-call triage

One run of [`examples/earnings_triage.py`](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).

```python
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`):

```text
━━━ 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.

```mermaid
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

```python
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.

```python
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:

```python
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:

```python
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]"`):

```python
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:

```python
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.

```python
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:

```python
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:

```python
print(pipe.explain())
```

```text
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:

```python
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:

```python
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:

```python
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:

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

### Conditional Branching

Dynamically route execution based on upstream agent outputs:

```python
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:

```python
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:

```python
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:

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

### Rate Limiting

Throttle API calls for rate-limited providers:

```python
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:

```python
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:

```python
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:

```python
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

```python
# 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`](examples/earnings_triage.py): **the showcase**, 6-agent diamond DAG with tools, parallel analysts, typed output, budget, and cache
- [`examples/tool_agent.py`](examples/tool_agent.py): ReAct agent that calls tools (calculator + stock lookup)
- [`examples/streaming_and_cost.py`](examples/streaming_and_cost.py): token streaming + USD cost tracking
- [`examples/research_crew.py`](examples/research_crew.py): 3-agent sequential research pipeline
- [`examples/code_reviewer.py`](examples/code_reviewer.py): 2-agent code review pipeline
- [`examples/market_analysis_crew.py`](examples/market_analysis_crew.py): 5-agent parallel market analysis (diamond DAG)
- [`examples/memory_chat_agents.py`](examples/memory_chat_agents.py): two agents sharing context across separate runs via memory
- [`examples/research_react_agent.py`](examples/research_react_agent.py): single ReAct agent researching with tools
- [`examples/cpp_build_pipeline.py`](examples/cpp_build_pipeline.py): write / compile / test loop driving a real toolchain
- [`examples/robotics_mqtt_agent.py`](examples/robotics_mqtt_agent.py): `Pipeline.serve()` daemon fed by an MQTT trigger *(needs the `mqtt` extra)*
- [`examples/drone_telemetry_agent.py`](examples/drone_telemetry_agent.py): `MQTTDaemon` with a Pydantic-validated trigger policy *(needs the `mqtt` extra)*
- [`benchmarks/parallel_speedup.py`](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](CONTRIBUTING.md) for development setup, coding style, and PR requirements.

## Changelog

See [CHANGELOG.md](CHANGELOG.md).

## License

MIT
