Building Your First AI Agent with Burr

Introduction

Burr is a Python framework for building stateful, production-ready AI agents. It uses a state machine abstraction that makes agent design clear and maintainable. In this tutorial, you'll learn how to build your first AI agent using Burr.

Prerequisites

• Python 3.10+
• Basic understanding of Python and async programming
• An OpenAI API key (or other LLM provider)

Installation

pip install burr
pip install openai  # or your preferred LLM provider

Core Concepts

Before writing code, let's understand Burr's key concepts:

State

State is the agent's memory — it holds all the information the agent needs to make decisions. In Burr, state is a dictionary that persists across actions.

Actions

Actions are units of work that transform state. Each action:

• Reads specific state fields
• Writes specific state fields
• Returns a result

Transitions

Transitions define how the agent moves from one action to the next based on conditions.

Step 1: Define Your State

First, define what state your agent needs:

from burr import state

# Define your state schema
class AgentState(state.State):
    messages: list[dict]  # Conversation history
    current_topic: str    # Current discussion topic
    confidence: float     # Agent's confidence in responses

Step 2: Create Actions

Actions are the building blocks of your agent. Here's a simple example:

from burr import action

@action(
    reads=["messages", "current_topic"],
    writes=["messages", "confidence"]
)
def respond(state: AgentState) -> tuple[dict, AgentState]:
    """Generate a response based on conversation history."""
    messages = state["messages"]
    topic = state["current_topic"]
    
    # Call your LLM (simplified example)
    response = call_llm(messages)
    
    # Update state
    new_messages = messages + [{"role": "assistant", "content": response}]
    confidence = calculate_confidence(response)
    
    return {
        "response": response
    }, state.update(
        messages=new_messages,
        confidence=confidence
    )

Step 3: Define Transitions

Transitions control the flow of your agent:

from burr import transition

# Always go to respond after initialization
initialize >> respond

# After responding, check if we should continue or end
respond >> transition.when(
    condition=lambda state: state["confidence"] > 0.5,
    then=respond  # Continue if confident
).otherwise(end)  # End if not confident

Step 4: Build the Application

Now assemble everything into a Burr application:

from burr import Application, System

# Define the application
app = Application(
    state=AgentState(messages=[], current_topic="general", confidence=0.0),
    actions=[initialize, respond, end],
    transitions=[
        initialize >> respond,
        respond >> transition.when(
            condition=lambda state: state["confidence"] > 0.5,
            then=respond
        ).otherwise(end)
    ],
    system_prompt="You are a helpful AI assistant."
)

Step 5: Run the Agent

# Start the agent
result = app.run(
    halt_after=["end"],
    inputs={"user_message": "Hello, can you help me with Python?"}
)

print(result["response"])

Advanced: Streaming Responses

Burr supports streaming for real-time responses:

for chunk in app.stream(
    halt_after=["respond"],
    inputs={"user_message": "Explain quantum computing"}
):
    print(chunk["response"], end="", flush=True)

Advanced: Parallel Actions

For complex tasks, run actions in parallel:

@action(reads=["query"], writes=["search_results", "code_examples"])
def parallel_research(state: AgentState) -> tuple[dict, AgentState]:
    """Run web search and code search in parallel."""
    from burr import parallel
    
    with parallel() as p:
        search = p.spawn(web_search, query=state["query"])
        code_search = p.spawn(code_search, query=state["query"])
    
    return {
        "search_results": search.result(),
        "code_examples": code_search.result()
    }, state

Advanced: A/B Testing

Burr has built-in A/B testing for comparing agent variants:

@action(reads=["messages"], writes=["response"])
def respond_v1(state: AgentState) -> tuple[dict, AgentState]:
    # Variant A: Direct responses
    response = call_llm_direct(state["messages"])
    return {"response": response}, state

@action(reads=["messages"], writes=["response"])
def respond_v2(state: AgentState) -> tuple[dict, AgentState]:
    # Variant B: Chain-of-thought
    response = call_llm_cot(state["messages"])
    return {"response": response}, state

# Configure A/B testing
app = Application(
    # ...
    ab_testing={
        "respond": {"v1": 0.5, "v2": 0.5}  # 50/50 split
    }
)

Complete Example: RAG Agent

Here's a complete RAG (Retrieval-Augmented Generation) agent:

from burr import Application, action, transition
from typing import TypedDict

class RAGState(TypedDict):
    query: str
    retrieved_docs: list[dict]
    answer: str
    confidence: float

@action(reads=[], writes=["query"])
def parse_input(state: RAGState, inputs: dict) -> tuple[dict, RAGState]:
    return {}, state.update(query=inputs["query"])

@action(reads=["query"], writes=["retrieved_docs"])
def retrieve(state: RAGState) -> tuple[dict, RAGState]:
    docs = vector_db.search(state["query"], k=5)
    return {}, state.update(retrieved_docs=docs)

@action(reads=["query", "retrieved_docs"], writes=["answer", "confidence"])
def generate(state: RAGState) -> tuple[dict, RAGState]:
    context = format_docs(state["retrieved_docs"])
    answer = llm.generate(f"Query: {state['query']}\nContext: {context}")
    confidence = evaluate_answer(answer, state["query"])
    return {}, state.update(answer=answer, confidence=confidence)

@action(reads=["answer"], writes=[])
def respond(state: RAGState) -> tuple[dict, RAGState]:
    return {"response": state["answer"]}, state

# Build the pipeline
app = Application(
    state=RAGState(query="", retrieved_docs=[], answer="", confidence=0.0),
    actions=[parse_input, retrieve, generate, respond],
    transitions=[
        parse_input >> retrieve,
        retrieve >> generate,
        generate >> respond
    ]
)

# Run
result = app.run(halt_after=["respond"], inputs={"query": "What is Burr?"})
print(result["response"])

Debugging and Observability

Burr provides built-in observability:

# Enable tracing
from burr import configure_logging
configure_logging(level="DEBUG")

# View traces
for trace in app.trace_history():
    print(f"Action: {trace.action}, State: {trace.state}")

Testing Your Agent

Burr includes a testing framework:

from burr.testing import test_action

def test_respond():
    state = AgentState(
        messages=[{"role": "user", "content": "Hello"}],
        current_topic="greeting",
        confidence=0.0
    )
    result, new_state = respond(state)
    assert "response" in result
    assert new_state["confidence"] > 0

test_action(respond, test_respond)

Next Steps

• Burr Documentation — Complete API reference
• Burr Examples — More example agents
• Burr GitHub — Source code and issues

Summary

In this tutorial, you learned:

1. How to define state for your agent
2. How to create actions that transform state
3. How to define transitions between actions
4. How to build and run a Burr application
5. Advanced features: streaming, parallel actions, A/B testing
6. How to build a complete RAG agent
7. How to debug and test your agent

Burr's state machine approach makes it easy to build clear, maintainable, and production-ready AI agents. Start with simple agents and gradually add complexity as you learn the framework.