🤖 AgentFlow

A production-ready, zero-dependency multi-agent orchestration framework for Agentic AI.

---

Why AgentFlow?

Modern agentic AI systems need more than a single LLM call. They require planning, tool use, memory, multi-agent coordination, and self-criticism — all wired together reliably.

AgentFlow provides exactly that: a lightweight, extensible Python framework you can drop into any project to orchestrate networks of AI agents.

Goal ──► PlannerAgent ──► ExecutorAgent ──► CriticAgent ──► Result
                  ↑              │                │
                  └──────────────┴── retry? ──────┘

---

Features

Feature	Description
BaseAgent	Abstract foundation with retry logic, messaging, and state
PlannerAgent	Decomposes goals into ordered subtasks (heuristic or LLM-backed)
ExecutorAgent	Runs subtasks via tools, LLM, or custom handlers
CriticAgent	Scores outputs and triggers retries when quality is low
AgentPipeline	Sequential, parallel, and plan-execute orchestration modes
AgentRouter	Dynamically routes tasks to the best-matching agent
ShortTermMemory	In-process sliding-window memory
LongTermMemory	File-backed persistent JSON Lines store
ToolRegistry	Register and discover tools by name or keyword
Built-in Tools	Calculator, File I/O, Web Search stub, Timer
Zero core deps	Pure Python stdlib — add LLM/vector-DB extras as needed
Full test suite	40+ pytest tests across agents, memory, tools, and orchestrator
GitHub Actions CI	Tests on Python 3.9–3.12 with coverage reporting

---

Quick Start

Install

pip install agentflow
# or from source:
git clone https://github.com/your-org/agentflow
cd agentflow && pip install -e .

1 — Sequential pipeline (Planner → Executor → Critic)

from agentflow.agents.planner import PlannerAgent
from agentflow.agents.executor import ExecutorAgent
from agentflow.agents.critic import CriticAgent
from agentflow.orchestrator.pipeline import AgentPipeline

pipeline = AgentPipeline(
    name="MyPipeline",
    agents=[
        PlannerAgent(name="Planner"),
        ExecutorAgent(name="Executor"),
        CriticAgent(name="Critic"),
    ],
    mode="plan_execute",
)

run = pipeline.run("Build a REST API for a todo app")
print(run.final_output)
print(f"Success: {run.success} | Time: {run.total_time:.2f}s")

2 — Dynamic agent routing

from agentflow.orchestrator.router import AgentRouter

router = (
    AgentRouter(strategy="keyword")
    .register(planner,  keywords=["plan", "design", "decompose"])
    .register(executor, keywords=["build", "implement", "run"])
    .register(critic,   keywords=["review", "evaluate", "check"])
)

result = router.dispatch("Please review and evaluate the output quality")

3 — Parallel agents

pipeline = AgentPipeline(
    agents=[summary_agent, sentiment_agent, keyword_agent],
    mode="parallel",
    max_workers=3,
)
run = pipeline.run(document_text)
print(run.final_output)   # dict of {agent_name: output}

4 — Tools

from agentflow.tools.builtins import CalculatorTool, FileTool
from agentflow.tools.registry import ToolRegistry

registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(FileTool(base_dir="/tmp"))

# Or use the decorator
@registry.tool("web_search", keywords=["search", "find"])
def search(query: str, **_) -> str:
    ...  # plug in any search provider

5 — Memory

from agentflow.memory.short_term import ShortTermMemory
from agentflow.memory.long_term import LongTermMemory

# In-process
stm = ShortTermMemory(capacity=100)
stm.store({"result": "..."})
hits = stm.search("result")

# Persistent across runs
ltm = LongTermMemory(path="./agent_memory.jsonl")
ltm.store({"task": "translate", "output": "..."}, tags=["translation"])

6 — Bring your own LLM

class MyLLMClient:
    def complete(self, prompt: str) -> str:
        # call OpenAI, Anthropic, Ollama, etc.
        ...

executor = ExecutorAgent(
    name="LLMExecutor",
    mode="llm",
    llm_client=MyLLMClient(),
)

planner = PlannerAgent(
    name="LLMPlanner",
    strategy="llm",
    llm_client=MyLLMClient(),
)

---

Architecture

agentflow/
├── agents/
│   ├── base.py          ← BaseAgent, AgentMessage, AgentResult
│   ├── planner.py       ← PlannerAgent (sequential / dependency / llm)
│   ├── executor.py      ← ExecutorAgent (tool / llm / hybrid)
│   └── critic.py        ← CriticAgent  (heuristic / llm / custom)
├── orchestrator/
│   ├── pipeline.py      ← AgentPipeline (sequential / parallel / plan_execute)
│   └── router.py        ← AgentRouter   (keyword / round_robin / custom)
├── memory/
│   ├── short_term.py    ← ShortTermMemory (in-process deque)
│   └── long_term.py     ← LongTermMemory  (JSONL file)
└── tools/
    ├── registry.py      ← ToolRegistry, BaseTool, FunctionTool
    └── builtins.py      ← CalculatorTool, FileTool, WebSearchTool, TimerTool

---

Running Tests

pip install -e ".[dev]"
pytest tests/ -v --cov=agentflow

Running Examples

python examples/01_basic_pipeline.py
python examples/02_agent_router.py
python examples/03_tools_and_memory.py
python examples/04_parallel_pipeline.py

---

Roadmap

• Async / asyncio support for pipeline and agents
• Vector-store memory backends (FAISS, ChromaDB, Pinecone)
• LangChain & LlamaIndex tool adapters
• OpenAI & Anthropic LLM client integrations (built-in)
• Agent-to-agent delegation (hierarchical multi-agent)
• REST API server (uvicorn + FastAPI)
• Web dashboard for pipeline monitoring
• Streaming output support

---

Contributing

Contributions are welcome! Please open an issue or pull request.

1. Fork the repo and create a feature branch
2. Write tests for new functionality
3. Ensure pytest and ruff pass
4. Submit a PR with a clear description

---

License

MIT — see LICENSE.