AGNT5 Foundations
Four building blocks for reliable AI systems
Four building blocks for reliable AI workflows. Each solves a different problem.
The Four Primitives
| Function | Entity | Workflow | Agent | |
|---|---|---|---|---|
| What | Stateless operation | Stateful component | Multi-step process | AI-powered system |
| State | None | Persistent per key | Per workflow run | Conversation history |
| Recovery | Automatic retries | State survives restarts | Resume from checkpoint | Context preserved |
| Best For | API calls, embeddings | User sessions, memory | Pipelines, orchestration | Reasoning, decisions |
| Example | Generate embedding | Chat history | RAG pipeline | Support agent |
When to Use What
Need AI to make decisions? → Use Agent with tools
Need multiple coordinated steps? → Use Workflow to orchestrate them
Need to remember state across calls? → Use Entity with a unique key
Just execute and return? → Use Function
Common Patterns
Pattern 1: AI Chatbot
Entity (user session) + Agent (chat)Entity maintains conversation history, Agent handles responses.
Pattern 2: Content Generation Pipeline
Workflow orchestrates:
├─ Function: Retrieve documents
├─ Agent: Generate content
└─ Function: Publish resultWorkflow ensures pipeline completes even if steps fail.
Pattern 3: Autonomous Assistant
Workflow + Agent + ToolsWorkflow manages overall flow, Agent reasons and uses Tools to accomplish goals.
Quick Example
from agnt5 import Agent, workflow, tool, Context, WorkflowContext
# Tool for the agent
@tool
async def search_docs(ctx: Context, query: str) -> str:
"""Search documentation for answers."""
return f"Found results for: {query}"
# Agent with tools
agent = Agent(
name="assistant",
model="openai/gpt-4o-mini",
instructions="You are a helpful assistant.",
tools=[search_docs]
)
# Workflow orchestrating agent
@workflow()
async def process_question(ctx: WorkflowContext, question: str) -> dict:
"""Durable workflow with checkpointing."""
# Step 1: Get answer from agent (checkpointed)
result = await ctx.step("get_answer", agent.run(question))
# Step 2: Store result in workflow state
ctx.state.set("last_question", question)
ctx.state.set("last_answer", result.output)
return {"question": question, "answer": result.output}
# If crash happens after step 1, resumes from step 2