35–45 min read · Hands-on coding
LangGraph, PydanticAI, and CrewAI — the orchestration frameworks you will explore in the next lessons — are built on top of the same raw API calls you have been writing since Lesson 2. Before you meet those frameworks, it pays to build what they build, by hand, once. That way you will know exactly what heavy lifting they are doing for you, and when a framework behaves unexpectedly, you will be able to reason about what is happening underneath rather than guessing.
This lesson walks through a complete, production-shaped agent implementation in plain Python with nothing but the OpenAI SDK and Pydantic — the exact foundation every agent framework builds on.
When you write a LangGraph workflow or a PydanticAI agent, the framework does exactly four things for you:
None of these require a framework. They are straightforward Python. A hand-rolled agent is often shorter than a framework-based one for simple use cases, and always more debuggable.
The following is a production-shaped agent class. It wraps everything you built across Lessons 12–14 into a single, reusable component:
import json
import os
import time
from dataclasses import dataclass, field
from typing import Callable, Any
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# ── Tool definition helpers ────────────────────────────────────
@dataclass
class Tool:
"""Wraps a Python function as an agent tool."""
name: str
description: str
parameters: dict # JSON Schema for the arguments
fn: Callable[..., Any]
def to_openai_schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
}
def execute(self, arguments_json: str) -> str:
args = json.loads(arguments_json)
try:
result = self.fn(**args)
return str(result)
except Exception as e:
return f"Tool error: {e}"
# ── Agent state ────────────────────────────────────────────────
@dataclass
class AgentState:
"""Mutable state for one agent run."""
messages: list = field(default_factory=list)
iterations: int = 0
total_tokens: int = 0
elapsed_s: float = 0.0
finished: bool = False
error: str | None = None
# ── Agent ──────────────────────────────────────────────────────
class Agent:
"""
A ReAct agent over the OpenAI API.
Instantiate once; call run() for each new task.
"""
def __init__(
self,
model: str = "gpt-4o-mini",
system_prompt: str = "You are a helpful agent. Use tools to accomplish the task, then provide a final answer.",
tools: list[Tool] = None,
max_iterations: int = 10,
max_tokens_per_run: int = 50_000,
timeout_s: float = 120.0,
):
self.model = model
self.system_prompt = system_prompt
self.tools = tools or []
self.max_iterations = max_iterations
self.max_tokens = max_tokens_per_run
self.timeout_s = timeout_s
self._tool_map = {t.name: t for t in self.tools}
self._openai_schemas = [t.to_openai_schema() for t in self.tools]
# ── Run ───────────────────────────────────────────────────
def run(self, user_message: str, verbose: bool = False) -> str:
state = AgentState()
state.messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_message},
]
start = time.monotonic()
while True:
# ── Stopping conditions ────────────────────────────
if state.iterations >= self.max_iterations:
return self._abort(state, "max iterations reached")
if state.total_tokens >= self.max_tokens:
return self._abort(state, "token budget exceeded")
if time.monotonic() - start > self.timeout_s:
return self._abort(state, "timeout")
# ── LLM call ──────────────────────────────────────
response = client.chat.completions.create(
model=self.model,
messages=state.messages,
tools=self._openai_schemas or None,
)
state.iterations += 1
state.total_tokens += response.usage.total_tokens
msg = response.choices[0].message
# ── No tool calls → final answer ──────────────────
if not msg.tool_calls:
state.finished = True
if verbose:
print(f"[Agent] Done in {state.iterations} iterations, "
f"{state.total_tokens} tokens")
return msg.content
# ── Execute tool calls ────────────────────────────
state.messages.append(msg) # assistant message with tool_calls
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name
tool = self._tool_map.get(tool_name)
if tool is None:
result = f"Unknown tool: {tool_name}"
else:
result = tool.execute(tool_call.function.arguments)
if verbose:
print(f"[Tool] {tool_name}({tool_call.function.arguments[:60]}) → {result[:80]}")
state.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
def _abort(self, state: AgentState, reason: str) -> str:
state.error = reason
try:
r = client.chat.completions.create(
model=self.model,
messages=state.messages + [{
"role": "user",
"content": f"Stopping early: {reason}. Summarise what you found so far.",
}],
)
return f"[{reason}] {r.choices[0].message.content}"
except Exception:
return f"[{reason}] No partial result available."
Define tools as Tool objects, instantiate the agent once, and call run() for each task:
import datetime
# ── Define tools ────────────────────────────────────────────────
tools = [
Tool(
name="get_current_date",
description="Return today's date in ISO format (YYYY-MM-DD).",
parameters={"type": "object", "properties": {}, "required": []},
fn=lambda: datetime.date.today().isoformat(),
),
Tool(
name="calculate",
description="Evaluate a safe arithmetic expression. Example: '2 + 2 * 10'.",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "The arithmetic expression"}
},
"required": ["expression"],
},
fn=lambda expression: str(eval(expression, {"__builtins__": {}})),
),
]
# ── Create and run ───────────────────────────────────────────────
agent = Agent(
model="gpt-4o-mini",
system_prompt=(
"You are a research assistant. Use your tools to gather information. "
"Always cite sources and acknowledge uncertainty."
),
tools=tools,
max_iterations=8,
timeout_s=60.0,
)
answer = agent.run(
"What is today's date? Also, if a $1000 investment grows 7% per year, "
"how much is it worth after 10 years?",
verbose=True,
)
print(answer)
The Agent above is stateless between run() calls — each call starts fresh. For conversational agents, you want to carry history across runs. The minimal approach:
class ConversationalAgent(Agent):
"""Extends Agent with cross-run memory."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._history: list = []
def chat(self, user_message: str, verbose: bool = False) -> str:
answer = self.run(user_message, verbose=verbose)
# Update history
self._history.append({"role": "user", "content": user_message})
self._history.append({"role": "assistant", "content": answer})
# Trim to last 20 turns to stay within context window
if len(self._history) > 40:
self._history = self._history[-40:]
return answer
This is the same pattern that every chat SDK and framework uses — they just wrap it in more abstraction. Seeing the raw version makes it trivial to debug context window issues, trim strategies, and memory bugs.
| Scenario | Recommendation |
|---|---|
| Simple agent, 2–5 tools, single step | Raw API — fewer moving parts |
| Agent inside a larger application | Raw API — cleaner integration |
| Strict type guarantees on result | PydanticAI — typed result enforcement |
| Complex branching, persistence, human-in-the-loop | LangGraph — built for this |
| Debugging a framework agent | Reproduce the failing case with raw API first |
| Teaching or onboarding a team | Raw API first — then introduce a framework once the team understands the fundamentals |
Agent class with an evaluator-optimizer tool.
Agent and Tool classes from this lessonself_critique tool that the agent can call when it wants to evaluate its own draft answer: it sends the draft to a second LLM call and returns a quality score (1–10) and improvement suggestionsAnswer each question from memory, then click to reveal.
It checks msg.tool_calls on the response message. If tool_calls is empty or None, the model has no more tools to call — the loop exits and returns msg.content as the final answer.
It maintains a self._history list of past user and assistant messages. Before each run it prepends the system prompt + history to the messages, and after each run it appends the new user message and answer to the history. It trims to the last 40 messages (20 turns) to stay within the context window.