zeroagent

Python-pure AI Agent + Workflow SDK. Zero external dependencies.

---

What it is

zeroagent is a minimalist SDK for building AI agents and workflows in Python, with zero external dependencies. Everything runs on the Python standard library.

Why?

• pip install langchain pulls 80+ transitive dependencies. For most use cases, that's unnecessary.
• Understanding what actually happens inside an agent loop is impossible when everything is abstracted behind opaque classes.
• Constrained environments (Kaggle, Alpine containers, Raspberry Pi) can't afford that weight.

zeroagent exposes the same primitives — ReAct agent, tool registry, memory, native async, workflow DAG — in code you can read in an afternoon.

Working today:

• One-shot CLI, interactive chat, token-by-token streaming
• Structured tool calling + full ReAct loop
• Conversational memory (Buffer, Window, Summary)
• Native async support (arun(), AsyncLLMClient, parallel tool calls)
• Workflow DAG (Node, State, Graph, conditional edges, parallel execution)
• Stable public SDK API

---

Quick start

No dependencies to install for zeroagent itself:

# 1. Clone
git clone https://github.com/Tahsine/zeroagent.git
cd zeroagent

# 2. Run directly (zero pip install)
python -m zeroagent "Calculate 123 * 456"

# 3. Interactive chat mode
python -m zeroagent --chat

# 4. Token-by-token streaming
python -m zeroagent --chat --stream

# 5. Watch the ReAct loop (thought → action → observation)
python -m zeroagent "Calculate (15 + 27) * 3" --verbose

Configuration

Create a .env file at the root:

LLM_BASE_URL=https://ollama.com/v1
LLM_MODEL=minimax-m3:cloud
LLM_API_KEY=your_api_key

zeroagent works with any OpenAI-compatible provider: OpenAI, Ollama (local or cloud), Kaggle LLM Server, Anthropic, and more.

---

Agent — SDK API

from zeroagent import Agent, tool, LLMClient

@tool(description="Search for information")
def search(query: str) -> str:
    return f"Results for: {query}"

llm = LLMClient(base_url="https://api.openai.com/v1", api_key="sk-...")
agent = Agent(llm=llm, tools=[search])

# Sync
result = agent.run("What is the capital of Benin?")
print(result)

# Async
import asyncio
result = asyncio.run(agent.arun("What is the capital of Benin?"))
print(result)

run() / run_full() / arun() / arun_full()

# run() — final answer (str), sync
answer = agent.run("question")

# run_full() — full RunResult, sync
result = agent.run_full("question")
print(result.answer)           # str
print(result.stop_reason)      # StopReason.FINAL_ANSWER | MAX_ITERATIONS | ERROR
print(result.iterations)       # int
print(result.tool_calls_made)  # int
print(result.thoughts)         # list[str]

# arun() — final answer (str), async
answer = await agent.arun("question")

# arun_full() — full RunResult, async
result = await agent.arun_full("question")

Conversational memory

from zeroagent import BufferMemory, WindowMemory, SummaryMemory

# Full buffer — keeps the entire conversation
agent = Agent(llm=llm, memory=BufferMemory())

# Sliding window — keeps the last N messages
agent = Agent(llm=llm, memory=WindowMemory(k=10))

# Auto-summary — compresses old messages via LLM
agent = Agent(llm=llm, memory=SummaryMemory(llm=llm, max_messages=20, keep_recent=6))

Hooks

agent = Agent(
    llm=llm,
    tools=[search],
    on_thought=lambda thought, i: print(f"[{i}] Thought: {thought}"),
    on_action=lambda name, args, i: print(f"[{i}] Action: {name}({args})"),
    on_observation=lambda name, obs, ok, i: print(f"[{i}] Obs: {obs}"),
    on_final=lambda answer, reason: print(f"Final ({reason}): {answer}"),
)

Hooks can be sync functions or async coroutines — arun() detects and awaits them automatically.

Available imports

# Main entry points
from zeroagent import Agent, tool, LLMClient

# Async
from zeroagent import AsyncLLMClient, AsyncAgentLoop, AsyncExecutor

# LLM types
from zeroagent import Message, LLMResponse, ToolCall

# Registry
from zeroagent import ToolRegistry, ToolSchema

# Memory
from zeroagent import BufferMemory, WindowMemory, SummaryMemory

# Loop control and results
from zeroagent import RunResult, StopReason, RunConfig

# Advanced (hooks, introspection)
from zeroagent import ParsedAction, ActionType, ExecutionResult

# Workflow
from zeroagent import Graph, Node, State, GraphResult

---

Workflow DAG

from zeroagent import Agent, LLMClient, tool
from zeroagent.workflow import Graph, Node, State

@tool(description="Add two numbers")
def add(a: int, b: int) -> int:
    return a + b

llm = LLMClient(base_url="...", api_key="...")
agent = Agent(llm=llm, tools=[add])

# Pure functions as nodes
def prepare(state: State) -> State:
    state["input"] = f"Calculate {state['a']} + {state['b']}"
    return state

def format_output(state: State) -> State:
    state["final"] = f"Result: {state['output']}"
    return state

# Build the graph
graph = Graph(name="pipeline")
graph.add_node(Node("prepare", fn=prepare))
graph.add_node(Node("agent", fn=agent))     # zeroagent Agent as a node
graph.add_node(Node("format", fn=format_output))
graph.add_edge("prepare", "agent")
graph.add_edge("agent", "format")
graph.set_entry("prepare")
graph.set_finish("format")

# ASCII visualization
print(graph.visualize())
# Graph 'pipeline'
#
# ┌─────────┐
# │ prepare │ (entry)
# └─────────┘
#   │
#   ▼
# ┌───────┐
# │ agent │
# └───────┘
#   │
#   ▼
# ┌────────┐
# │ format │ (finish)
# └────────┘

# Sync execution
result = graph.run(State({"a": 42, "b": 58}))
print(result.state["final"])   # "Result: 42 + 58 = 100"
print(result.executed)         # ['prepare', 'agent', 'format']

# Async execution
result = await graph.arun(State({"a": 42, "b": 58}))

Conditional edges

def router(state: State) -> str:
    return "success" if state.get("ok") else "fallback"

graph.add_conditional_edge("validate", router, {
    "success": "process",
    "fallback": "error_handler",
})

Parallel execution

Nodes without dependencies between them run automatically in parallel via asyncio.gather() during arun():

# branch_a and branch_b run in parallel
graph.add_edge("start", "branch_a")
graph.add_edge("start", "branch_b")
graph.add_edge("branch_a", "merge")
graph.add_edge("branch_b", "merge")

State

from zeroagent.workflow import State

# Basic
state = State({"query": "test", "result": ""})
state["result"] = "done"
state.merge({"step": 2, "ok": True})

# Subclassed with schema and default values
class PipelineState(State):
    query: str
    result: str = ""
    retries: int = 0

state = PipelineState({"query": "test"})
# result and retries are injected automatically
errors = state.validate()   # checks required fields

Node

from zeroagent.workflow import Node, State

# Pure function
node = Node("step", fn=my_function)

# With retries and timeout
node = Node("step", fn=my_function, retries=3, timeout=30.0)

# Native async
async def async_step(state: State) -> State:
    await asyncio.sleep(0.1)
    state["done"] = True
    return state

node = Node("async_step", fn=async_step)

# zeroagent Agent (duck-typed: reads state["input"], writes state["output"])
node = Node("research", fn=my_agent)

---

Native async

AsyncLLMClient implements pure async HTTP/1.1 via asyncio — zero external dependencies, same API as LLMClient:

from zeroagent import AsyncLLMClient
from zeroagent.core.llm import Message

async def main():
    client = AsyncLLMClient(
        base_url="https://api.openai.com/v1",
        model="gpt-4o",
        api_key="sk-...",
    )

    # Completion
    response = await client.acomplete([
        Message(role="user", content="Hello!")
    ])
    print(response.content)

    # Streaming
    async for chunk in client.astream(messages):
        print(chunk, end="", flush=True)

    # From an existing LLMClient
    async_client = AsyncLLMClient.from_sync(sync_client)

Multiple tool calls in the same ReAct iteration run in parallel via asyncio.gather() inside AsyncAgentLoop.

---

CLI flags

Flag	Default	Description
question	—	One-shot question (positional)
--base-url	$LLM_BASE_URL	LLM provider URL
--model	$LLM_MODEL	Model name
--api-key	$LLM_API_KEY	API key
--system-prompt	default	Custom system prompt
--max-tokens	1024	Max tokens to generate
--temperature	0.3	Sampling temperature
--max-iterations	6	Max ReAct iterations
--no-tools	—	Disable default tools
--verbose	—	Show Thought/Action/Observation
--quiet	—	Minimal output (answer only)
--fail-on-error	—	Exit code 1 on failure
--chat	—	Interactive chat mode (REPL)
--stream	—	Token-by-token streaming (chat only)
--env	.env	Custom path to .env file

---

Architecture

src/zeroagent/
├── __init__.py         # Package root — re-exports the full public API
├── __main__.py         # CLI — argparse, colors, one-shot/chat/stream modes
├── sdk.py              # Public entry point — all stable imports
│
├── core/               # Foundations
│   ├── llm.py          # LLMClient — pure HTTP, OpenAI-compatible + Anthropic native
│   ├── async_llm.py    # AsyncLLMClient — pure async HTTP/1.1 (asyncio)
│   ├── tools.py        # @tool decorator + ToolRegistry (auto JSON Schema)
│   └── memory.py       # BufferMemory + WindowMemory
│
├── harness/            # Full ReAct loop
│   ├── agent.py        # Agent — public facade (run, arun, run_full, arun_full)
│   ├── loop.py         # AgentLoop sync — Thought → Action → Observation
│   ├── async_loop.py   # AsyncAgentLoop + AsyncExecutor (parallel tool calls)
│   ├── parser.py       # Parses LLM decisions (tool call, final answer, thought)
│   ├── executor.py     # Executes actions, captures observations
│   └── memory.py       # SummaryMemory (auto-summary via LLM)
│
├── workflow/           # Multi-node DAG
│   ├── state.py        # State — wrapped dict, merge, deep copy, validation
│   ├── node.py         # Node — fn(State)->State, sync/async, retries, Agent support
│   └── graph.py        # Graph — DAG, edges, conditional, parallel, ASCII viz
│
└── tests/              # 235 unit tests (stdlib unittest)
    ├── test_llm.py
    ├── test_tools.py
    ├── test_harness.py
    ├── test_memory.py
    ├── test_async.py
    └── test_workflow.py

---

zeroagent vs LangChain

A CLI script strictly identical to zeroagent was written with LangChain to enable a side-by-side comparison:

pip install langchain langchain-openai
python langchain-sdk-test.py "Calculate 123 * 456"
python langchain-sdk-test.py --chat --verbose

Same flags, same colors, same calculator tool, same .env. Key differences:

• zeroagent: zero dependencies, readable and auditable code
• LangChain: 40+ packages, black-box internals, but native structured tool calling

---

What it's not

• A production framework with checkpointing, observability, and distributed execution → use LangGraph
• A wrapper around a specific API → zeroagent is provider-agnostic
• A LangChain reimplementation → same primitives, different design choices

---

Roadmap

• Pure HTTP LLMClient (OpenAI-compatible + native Anthropic)
• @tool decorator + ToolRegistry (auto JSON Schema)
• Full ReAct loop (Thought → Action → Observation)
• Conversational memory (Buffer, Window, Summary)
• CLI one-shot, chat, streaming
• Stable public SDK API (from zeroagent import Agent, tool, LLMClient)
• Native async (arun(), AsyncLLMClient, AsyncAgentLoop, parallel tool calls)
• Workflow DAG (Node, State, Graph, conditional edges, parallel execution)
• PyPI publication

---

Status

Active development. Public updates on LinkedIn and YouTube.

Architecture decisions are documented in DESIGN.md.

---

License

MIT