# This is NOT a good way to build an agent.
# It's a tool-augmented chatbot pretending to be one.
import openai
import json
def bad_travel_agent(prompt: str):
"""A 'travel agent' that's really just a single LLM call with a tool."""
print(f"USER: {prompt}")
response = openai.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
tools=[{
"type": "function",
"function": {
"name": "search_flights",
"description": "Search for flights between two cities",
"parameters": {
"type": "object",
"properties": {
"origin": {"type": "string"},
"destination": {"type": "string"},
"date": {"type": "string"},
},
"required": ["origin", "destination", "date"],
},
},
}],
)
msg = response.choices[0].message
if msg.tool_calls:
tc = msg.tool_calls[0]
args = json.loads(tc.function.arguments)
# No validation, no error handling, no retries
result = f"Found flights from {args['origin']} to {args['destination']}"
print(f"RESULT: {result}")
else:
print(f"ASSISTANT: {msg.content}")
# --- Why this is broken ---
# 1. No state: It can't plan a multi-step trip. It searches flights,
# then immediately forgets. It can never book a hotel next.
# 2. No structured output: The LLM returns free text. Good luck
# parsing "Found flights..." into something your UI can render.
# 3. No budget: If the LLM hallucinates a tool loop, you pay forever.
# 4. No validation: What if the LLM sends "tomorrow" instead of
# "2026-03-15" for the date? The API call will fail silently.
# 5. No recovery: Crash = all progress lost. No checkpoint, no resume.
# 6. No loop: It can only do ONE thing. A real trip needs flights,
# hotels, activities, restaurants - all composed together.
# A proper travel itinerary agent with structured state,
# typed outputs, validated tools, and built-in budgets.
#
# pip install pydantic-ai logfire
import asyncio
import httpx
from dataclasses import dataclass
from datetime import date
from dotenv import load_dotenv
import os
load_dotenv()
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext, ModelRetry
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider
provider = GoogleProvider(api_key=os.getenv("GEMINI_API_KEY"))
model = GoogleModel("gemini-2.5-pro", provider=provider)
import logfire
# 1. OBSERVABILITY: Set up tracing from the start
# Every decision, tool call, and token cost is tracked automatically.
logfire.configure()
logfire.instrument_pydantic_ai()
# 2. STRUCTURED STATE: Define what the agent knows
# This is the agent's "memory" - passed via dependency injection,
# not mixed into the prompt.
@dataclass
class TripContext:
origin: str
destination: str
start_date: date
end_date: date
budget_usd: float
traveler_notes: str
http_client: httpx.AsyncClient # shared client for all tool calls
# 3. STRUCTURED OUTPUT: Define exactly what the agent returns
# No more parsing free text. The LLM must return this schema.
# Pydantic AI validates it automatically; if the LLM gets it wrong,
# it retries with the validation error as feedback.
class Activity(BaseModel):
time: str = Field(descriptinotallow="e.g., '9:00 AM' or 'Afternoon'")
name: str = Field(descriptinotallow="Name of the activity or place")
description: str = Field(descriptinotallow="One-sentence why it's worth it")
estimated_cost_usd: float = Field(ge=0)
class DayPlan(BaseModel):
day_number: int = Field(ge=1)
date: str
theme: str = Field(descriptinotallow="e.g., 'Old City & Street Food'")
activities: list[Activity] = Field(min_length=2, max_length=6)
meals: list[Activity] = Field(min_length=1, max_length=3)
class TripItinerary(BaseModel):
title: str = Field(descriptinotallow="A catchy name for the trip")
summary: str = Field(descriptinotallow="2-3 sentence overview of the trip")
days: list[DayPlan] = Field(min_length=1)
total_estimated_cost_usd: float = Field(ge=0)
packing_tips: list[str] = Field(min_length=1, max_length=5)
# 4. THE AGENT: Configured with guardrails baked in
travel_agent = Agent(
# "anthropic:claude-sonnet-4-6",
# "google-gla:gemini-2.5-flash", (can pass like that or we can
# define it separately like we are doing here)
model,
deps_type=TripContext,
output_type=TripItinerary, # <-- Forces structured output
instructinotallow=(
"You are an expert travel planner. Use the available tools to research "
"the destination, then return a detailed day-by-day itinerary. "
"Stay within the traveler's budget. Respect their preferences. "
"Be specific - use real place names, not generic suggestions."
),
model_settings={
"temperature": 0.3,
"max_tokens": 4096,
},
retries=2, # Auto-retry on validation failure
)
# 5. TOOLS: Validated, typed, with access to shared state
# Tools use dependency injection (RunContext) to access TripContext.
# Pydantic AI validates all arguments BEFORE execution.
@travel_agent.tool
async def get_weather_forecast(
ctx: RunContext[TripContext],
city: str,
check_date: str,
) -> str:
"""Get the weather forecast for a city on a specific date."""
# The key point: ctx.deps gives typed access to shared state.
try:
r = await ctx.deps.http_client.get(
"https://api.weatherapi.com/v1/forecast.json",
params={"q": city, "dt": check_date, "key": os.getenv("WEATHERAPI_API_KEY")},
)
if r.status_code != 200:
# Raise ModelRetry so the agent knows the tool failed
# and can decide to try again or skip.
raise ModelRetry(f"Weather API returned {r.status_code}. Skipping weather forecast.")
data = r.json()
condition = data["forecast"]["forecastday"][0]["day"]["condition"]["text"]
high_c = data["forecast"]["forecastday"][0]["day"]["maxtemp_c"]
return f"{city} on {check_date}: {condition}, high of {high_c}°C"
except httpx.RequestError:
raise ModelRetry("Weather API unreachable. Skipping weather forecast.")
@travel_agent.tool
async def search_attractions(
ctx: RunContext[TripContext],
query: str,
category: str, # "sightseeing" | "food" | "nightlife" | "nature"
) -> str:
"""Search for attractions and activities at the destination."""
# In production, this calls Google Places, Yelp, or a similar API.
# For this repo snippet, we default to a deterministic mock so the demo runs offline.
base_url = os.getenv("ATTRACTIONS_API_URL")
if not base_url:
return (
f"(mock) Top {category} ideas near {ctx.deps.destination} for '{query}':\n"
"- Fushimi Inari Taisha (early morning)\n"
"- Kiyomizu-dera + Sannenzaka stroll\n"
"- Nishiki Market (vegetarian-friendly options)\n"
"- Arashiyama Bamboo Grove + Tenryu-ji\n"
)
try:
r = await ctx.deps.http_client.get(
f"{base_url.rstrip('/')}/attractions",
params={
"q": query,
"near": ctx.deps.destination,
"category": category,
},
)
except httpx.RequestError:
return (
f"(mock) Attractions API unreachable; using fallback results for {ctx.deps.destination}.\n"
"- Fushimi Inari Taisha\n"
"- Kiyomizu-dera\n"
"- Nishiki Market\n"
)
if r.status_code != 200:
raise ModelRetry("Attractions API failed. Try a broader query.")
return r.text
@travel_agent.tool_plain
def check_budget_remaining(
total_budget_usd: float,
spent_so_far_usd: float,
) -> str:
"""Check how much budget is left for planning remaining days."""
remaining = total_budget_usd - spent_so_far_usd
if remaining < 0:
raise ModelRetry(
f"Over budget by ${abs(remaining):.0f}! "
"Re-plan with cheaper alternatives."
)
return f"${remaining:.0f} remaining out of ${total_budget_usd:.0f}"
# 6. DYNAMIC INSTRUCTIONS: Inject user context at runtime
# This runs before each agent call and adds personalized context
# to the system prompt, keeping the base instructions clean.
@travel_agent.instructions
def personalize(ctx: RunContext[TripContext]) -> str:
t = ctx.deps
nights = (t.end_date - t.start_date).days
return (
f"\n--- Trip Details ---\n"
f"Route: {t.origin} → {t.destination}\n"
f"Dates: {t.start_date} to {t.end_date} ({nights} nights)\n"
f"Budget: ${t.budget_usd:.0f} total\n"
f"Traveler notes: {t.traveler_notes}\n"
)
# 7. RUN IT: With proper lifecycle management
async def plan_trip():
async with httpx.AsyncClient(timeout=30.0) as client:
trip_ctx = TripContext(
origin="Delhi",
destinatinotallow="Kyoto",
start_date=date(2026, 4, 1),
end_date=date(2026, 4, 5),
budget_usd=2000.0,
traveler_notes="I love temples and street food. Vegetarian.",
http_client=client,
)
result = await travel_agent.run(
"Plan my trip!",
deps=trip_ctx,
)
# result.output is a fully validated TripItinerary object.
# Not a string. Not JSON you have to parse. A typed Python object.
itinerary = result.output
print(f"\n{'='*50}")
print(f"🗾 {itinerary.title}")
print(f"{'='*50}")
print(f"\n{itinerary.summary}\n")
for day in itinerary.days:
print(f"\n📅 Day {day.day_number} - {day.theme}")
print(f" {day.date}")
for act in day.activities:
print(f" {act.time}: {act.name} (${act.estimated_cost_usd:.0f})")
print(f" {act.description}")
print(f" 🍽 ️ Meals:")
for meal in day.meals:
print(f" {meal.time}: {meal.name} (${meal.estimated_cost_usd:.0f})")
print(f"\n💰 Estimated Total: ${itinerary.total_estimated_cost_usd:.0f}")
print(f"🎒 Packing Tips: {', '.join(itinerary.packing_tips)}")
# Observability: token usage and cost are tracked automatically
# via Logfire. Check your dashboard at https://logfire.pydantic.dev
print(f"\n📊 Token usage: {result.usage()}")
if __name__ == "__main__":
asyncio.run(plan_trip())
图片
图片
