Agents, Tools, and Workflow Graphs

An LLM agent is a system in which a language model decides, at runtime, which tools to call and in what order to accomplish a goal. The LLM acts as a reasoning engine; tools act as its effectors into the world. The combination enables behaviors that a single LLM call cannot produce: multi-step reasoning, web lookup, code execution, database queries, API calls - all orchestrated dynamically based on the task.

Agents introduce a new failure surface compared to single-call LLM systems. This module covers the patterns that make agents reliable enough to ship.

Agent = Model + Harness

Before the loop, a framing that changes how you build everything below it. The model is not the agent. The agent is the model plus the harness: the deterministic code around the model that decides which tools exist, validates every call the model proposes, executes it (or refuses), records what happened, and feeds the result back. Every property a team actually cares about in production lives in the harness, not the model: which actions are allowed, how much the run may cost, what gets logged, whether a human is asked before something irreversible happens, and how the system recovers when a step fails.

This matters because the most common way an agent project dies is not "the model was not smart enough". It is that the team shipped a prototype where the model did everything and the harness did nothing, and then could not answer basic questions from security, finance, or the on-call engineer: what can this thing do, what did it do last Tuesday, how much did that cost, and can it do it again by accident.

The practical discipline is a loop. When the agent makes a mistake, you do not "prompt harder" first. You ask which harness layer should have made that mistake impossible, and you change that layer:

Layer	What it owns	Example fix after a failure
Tool orchestration	Which tools exist, their schemas, which ones each task may see	Remove delete_record from the read-only research task's tool list
Verification loops	Checking outputs before the next step	Run the generated SQL against a sandbox and compare row counts before returning it
Context and memory	What the model sees each turn, what persists across runs	Compact tool output to a summary instead of dumping 40 KB of JSON into context
Guardrails	Permissions, budgets, schema validation, human confirmation	Require confirmation for any tool with a side effect on a customer record
Observability	Traces, costs, replayable runs	Log every proposed call, including the ones the harness refused

Most of the rest of this module is about those layers. The agent loop below is the model's contribution; the harness is yours. The dedicated modules that follow this one (context-engineering-designing-what-the-model-sees, harness-engineering-the-runtime-around-the-model, and mcp-and-agent-protocols-building-tool-servers) go deep on each layer.

The Agent Loop

The core agent pattern is a reasoning-action-observation loop (ReAct):

Thought: What do I need to do?
Action: call(tool, args)
Observation: tool_result
Thought: What does this tell me? What's next?
...
Answer: final_response

pythonCopy
import json
from openai import OpenAI

client = OpenAI()

# Tool definitions
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_documents",
            "description": "Search internal knowledge base for relevant information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "max_results": {"type": "integer", "default": 5},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "execute_sql",
            "description": "Run a read-only SQL query against the analytics database",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SELECT query only"},
                },
                "required": ["query"],
            },
        },
    },
]

def run_agent(user_message: str, tool_implementations: dict, max_turns: int = 10) -> str:
    messages = [{"role": "user", "content": user_message}]

    for turn in range(max_turns):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )
        msg = response.choices[0].message
        messages.append(msg)

        if msg.tool_calls is None:
            return msg.content  # Final answer

        # Execute each tool call
        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            if fn_name not in tool_implementations:
                result = f"Error: Tool '{fn_name}' not found"
            else:
                try:
                    result = tool_implementations[fn_name](**fn_args)
                except Exception as e:
                    result = f"Error executing {fn_name}: {str(e)}"

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result),
            })

    return "Agent reached maximum turns without completing the task."

Tool Design Principles

The LLM's tool-calling behavior is heavily influenced by your tool descriptions. Write them as if onboarding a new engineer:

pythonCopy
# Bad: vague and unhelpful
{"name": "db_query", "description": "Query the database"}

# Good: specific about what it can and cannot do
{
    "name": "execute_sql",
    "description": (
        "Run a read-only SELECT query against the analytics PostgreSQL database. "
        "Tables available: users, orders, events, product_catalog. "
        "Cannot write, update, or delete data. Returns results as a list of dicts. "
        "Limit queries to 1000 rows to avoid timeout."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "A valid read-only SELECT SQL query"
            }
        },
        "required": ["query"]
    }
}

Tool design guidelines:

• One tool per action type - don't bundle unrelated capabilities
• Name tools as verbs: search_, get_, calculate_, create_
• Include what the tool cannot do (prevents the model from trying impossible things)
• Return structured data (JSON/dict), not free text, so the model can extract reliably

Guardrails and Safety

Agents that execute code, write to databases, or call external APIs need explicit guardrails:

pythonCopy
import re

def safe_sql_execute(query: str, conn) -> list[dict]:
    # Only allow SELECT - no writes, no schema changes
    normalized = query.strip().upper()
    if not normalized.startswith("SELECT"):
        raise ValueError(f"Only SELECT queries are permitted. Got: {query[:50]}")

    # Block dangerous patterns
    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "TRUNCATE", "ALTER", "CREATE"]
    for keyword in dangerous:
        if re.search(r'\b' + keyword + r'\b', normalized):
            raise ValueError(f"Query contains prohibited keyword: {keyword}")

    cursor = conn.cursor()
    cursor.execute(query)
    columns = [desc[0] for desc in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchmany(1000)]

def safe_code_executor(code: str) -> str:
    # Use a sandboxed execution environment - never exec() user/LLM code directly
    # Options: RestrictedPython, Docker subprocess with resource limits, E2B API
    raise NotImplementedError("Use a sandboxed execution environment")

Never exec() LLM-generated code in your application process. Use a sandboxed environment (Docker, Firecracker, E2B).

Workflow Graphs: Going Beyond Single Agents

For complex tasks, a single agent with many tools becomes unreliable. Graph-based workflows decompose tasks into defined steps with explicit state:

pythonCopy
from typing import TypedDict, Annotated
import operator

# Using LangGraph pattern (conceptual - adapt to your framework)
class WorkflowState(TypedDict):
    query: str
    retrieved_docs: list[dict]
    analysis: str
    answer: str
    sources: list[str]

def retrieve_node(state: WorkflowState) -> WorkflowState:
    docs = hybrid_search(state["query"])
    return {**state, "retrieved_docs": docs}

def analyze_node(state: WorkflowState) -> WorkflowState:
    context = "\n".join([d["content"] for d in state["retrieved_docs"][:5]])
    analysis = call_llm(f"Summarize key facts:\n{context}")
    return {**state, "analysis": analysis}

def answer_node(state: WorkflowState) -> WorkflowState:
    answer = call_llm(
        f"Query: {state['query']}\nFacts: {state['analysis']}\nAnswer concisely."
    )
    sources = list({d["source"] for d in state["retrieved_docs"][:5]})
    return {**state, "answer": answer, "sources": sources}

# Define the graph: retrieve → analyze → answer
pipeline = [retrieve_node, analyze_node, answer_node]

def run_workflow(query: str) -> dict:
    state = WorkflowState(query=query, retrieved_docs=[], analysis="", answer="", sources=[])
    for node in pipeline:
        state = node(state)
    return {"answer": state["answer"], "sources": state["sources"]}

Long-Running Agents: Memory, Durable Execution, and Human-in-the-Loop

A single-turn tool call is easy. The hard version is an agent that runs for twenty minutes, calls forty tools, waits on a human approval halfway through, and must survive the process being restarted. Three pieces of infrastructure make that possible.

Memory is more than one thing. Working memory is the current context window. Episodic memory is a record of what happened in previous runs, usually retrieved by similarity or recency. Long-term memory is curated facts (user preferences, project conventions) that you deliberately write and can inspect. Treat these as three separate stores with three separate write policies. The common failure is to let the model write freely into long-term memory, which is how a single bad run permanently poisons every future one.

Compaction is lossy. When context fills, something has to give. Summarizing earlier turns is the usual answer, and it silently deletes the detail the model may need later ("the user said the deadline was Thursday, not Friday"). Two mitigations: keep tool outputs retrievable by ID so the summary can point back to the raw artifact, and prefer reconstructing context from durable state over relying on the summary as the only source of truth.

Durable execution means resuming, not restarting. If a five-step workflow crashes at step four, the correct behavior is to resume at step four with the results of steps one through three intact. Two patterns do this:

pythonCopy
# Pattern 1: checkpoint the graph state after every node (LangGraph-style)
graph = build_graph()
app = graph.compile(checkpointer=SqliteSaver.from_conn_string("runs.db"))
config = {"configurable": {"thread_id": run_id}}
app.invoke(initial_state, config)          # crashes after node 3
app.invoke(None, config)                   # resumes from the last checkpoint

# Pattern 2: wrap each step as a durable activity (Temporal-style)
@workflow.defn
class ResearchWorkflow:
    @workflow.run
    async def run(self, task):
        plan = await workflow.execute_activity(plan_step, task, ...)
        docs = await workflow.execute_activity(retrieve_step, plan, ...)
        # a human approval is just a signal the workflow waits on
        await workflow.wait_condition(lambda: self.approved)
        return await workflow.execute_activity(write_step, docs, ...)

Either way, the thing you get is a replayable history of every step. That history is also your best debugging tool: you can inspect the exact state the model saw before the step that went wrong.

Human-in-the-loop is a first-class state, not an exception. Any tool that is irreversible (sending, paying, deleting, deploying) should pause the run and wait for approval. Design the pause as a normal node in the graph so the run can be resumed hours later by a different process. If your "approval" is a print() and input() in the same Python process, you have a demo, not a system.

Observability for Agents

Agent traces are harder to debug than single LLM calls. Log every step:

pythonCopy
import time

class AgentTracer:
    def __init__(self, trace_id: str):
        self.trace_id = trace_id
        self.steps = []
        self.start_time = time.time()

    def record_step(self, step_type: str, input_data: dict, output_data: dict):
        self.steps.append({
            "trace_id": self.trace_id,
            "step": len(self.steps),
            "type": step_type,
            "input": input_data,
            "output": output_data,
            "elapsed_ms": int((time.time() - self.start_time) * 1000),
        })

    def to_dict(self):
        return {
            "trace_id": self.trace_id,
            "total_steps": len(self.steps),
            "total_ms": int((time.time() - self.start_time) * 1000),
            "steps": self.steps,
        }

Emit traces to your observability platform (Datadog, Honeycomb, LangSmith). Alert on: max_turns reached (agent stuck), tool error rate > threshold, and total latency > SLA.

Common Mistakes and Bad Instincts

No max_turns limit. An agent without a step budget can loop indefinitely. Always set a maximum number of turns and return a graceful failure message when exceeded. In production, set a timeout as well.

Trusting LLM-generated arguments to tools without validation. The LLM might hallucinate a field name, generate a malformed SQL query, or pass a value outside allowed bounds. Validate every tool argument before execution - treat LLM output as untrusted input at system boundaries.

Building complex multi-agent systems before making single agents reliable. Multi-agent architectures (orchestrator + sub-agents) compound failure modes. Start with single agents, get them to high reliability, then add parallelism and specialization.

Not distinguishing between conversational memory and task state. Agents need working memory (the current task's intermediate results) and optionally long-term memory (user preferences, past interactions). Mixing these in a single conversation history makes agents slow, expensive, and confused.

Where to Go Next

• context-engineering-designing-what-the-model-sees: control the context budget, tool surface, and compaction policy for your agent

• harness-engineering-the-runtime-around-the-model: build the permission, budget, verification, and tracing layers around any model

• mcp-and-agent-protocols-building-tool-servers: expose your tools through MCP so any client can use them

• agent-evals-trajectories-tool-calls-and-regression-suites: grade the trajectory, not just the final answer

• fine-tuning-adaptation-and-when-not-to-fine-tune: fine-tune a model on tool-calling examples to improve reliability

• observability-drift-feedback-loops-and-llm-evals: monitor agent behavior in production at scale

• ai-system-design-quality-cost-latency-and-safety-tradeoffs: design the architecture when choosing between agents, RAG, and fine-tuning