为什么你的 RAG 应用或 workflow 工具并不是你以为的那样，以及应该改而构建什么
agent 这个词现在无处不在。它出现在网站、Notion 文档、PRD，以及各种会议里；那些过去被称为“带工具的 chatbot”的东西，突然就变成了 agent。而这并不只是措辞问题。
把 agent 当作一个很酷的营销术语，而不是一份严谨的工程规格，是一个真实存在的问题。它掩盖了我们试图构建的东西所包含的复杂性。如果你无法清楚地框定什么是 agent，你就无法可靠地构建它；当它出问题时，你也无法调试它；更不用说把它交付给用户时还能不预期各种问题会发生。
所以，我们来试着修正这一点。这不是又一篇高屋建瓴的思辨文章。它是一份面向开发者的实用指南，写给那些已经厌倦 buzzword、想知道 AI agent 到底是什么、什么时候该构建一个，以及更重要的是，什么时候不该构建的人。
Agent vs. RAG vs. Workflow
在定义 agent 是什么 之前，我们需要先明确它 不是什么。今天大多数被称为 agent 的系统，实际上属于另外两个类别：RAG 应用和 workflow。
RAG (Retrieval-Augmented Generation) 本质上是智能搜索。你有一个问题，系统从知识库中检索相关文档，然后 LLM 使用这些文档生成答案。这是一个知识访问问题。整个过程是无状态的，通常在单轮交互中完成。
Workflow 是一系列确定性的步骤。例如：一个 Zapier 集成，或一个企业入职流程脚本。如果 X 发生，那么执行 Y，再执行 Z。它可能有分支和逻辑，但路径是可预测的。这是一个流程问题，主要挑战在于确保这些步骤是持久的，并且在失败时可以重试。
而 Agent 则试图解决一个控制问题。它有一个目标和一组工具，并且必须自己找出路径。它是一个循环，而不是一条直线。它会根据当前情况决定下一步做什么。这就是根本区别。
最小可行的 agent 循环大致如下：Event -> Policy(LLM) -> Action(tool) -> State(write) -> Guardrails -> Stop
它接收输入，LLM 做出决策，可能调用一个工具，更新自己对刚才发生了什么的内部记忆，然后决定是否已经完成，或者是否需要再次循环。
AI Agent 的实用定义
如果我们要构建这些东西，就需要一个真正可以用来检验代码的定义。对我来说，一个系统只有在满足四个具体标准时，才有资格被称为“AI agent”。如果你正在构建某个东西，却无法勾选这四项，你拥有的很可能是一个工具增强型聊天应用，而不是 agent。这完全没问题，但用正确的名称称呼它，有助于你用正确的方式构建它。
1. 它维护超出单个 prompt 的状态
Agent 需要一种不只是聊天历史的 memory。我指的是一个显式的、结构化的状态对象（一个 schema）。这个状态应该跟踪以下内容：
总体目标。
上一次被调用工具的输出。
用于 agent 内部思考和计划的 scratchpad。
它已经收集到的任何中间结果或数据。
仅仅把之前的消息传回给 LLM 是不够的。那是对话，而不是有状态的执行。状态让 agent 能够推理自己的进展，并在循环中做出有依据的决策。
2. 它基于该状态选择行动
Agent 中的 LLM 不只是生成文本。它充当的是 policy engine 或 router。基于输入和当前状态，它决定下一步做什么。
可能的行动通常包括：
使用特定参数调用某个具体工具。
向用户请求澄清。
更新其内部状态或计划。
判定任务已完成并停止。
这种决策过程就是“agentic”的部分。它把 agent 和那种下一步已经写死的简单 workflow 区分开来。
3. 它在预算约束下运行
一个完全放任自流的 agent 可能会永远运行下去。我见过一个工具循环因为没有 max_steps 而运行了数百步。这就是你一觉醒来收到意外账单的原因。没有预算的 agent 是生产环境中的风险点。
每个 agent 都需要明确定义的预算：
Max steps：循环可运行次数的硬性上限（例如 10 步）。这是你对抗无限循环的主要防线。
Timeouts：为整个运行过程预先定义的超时。
Cost: 金额上限。这个更难实现，但如果你使用昂贵的模型或 API，它至关重要。
Retries：某个具体工具失败时可重试次数的上限。
没有这些，你就无法保证 agent 最终会结束，也无法预测它会花费多少成本。
4. 它能够安全恢复或降级
如果你的服务器崩溃，或者 pod 在 agent 运行到一半时重启，会发生什么？如果答案是“一切都丢失了”，那它就不是一个生产就绪的 agent。
这意味着你需要两样东西：
Idempotency: 使用相同输入多次调用工具，应该产生相同结果。如果 agent 重试发送邮件，它不应该把同一封邮件发送两次。
Durable Checkpoints: 在每一步之后（或者至少在每次成功的工具调用之后），必须把 agent 的状态保存到一个持久化存储中，例如数据库（Postgres、Redis 等）。这允许一次运行从它中断的确切位置继续。
如果你的系统无法承受一次随机重启，它就是原型，而不是可靠的 agent。
从糟糕到更好：一个代码示例
让我们把这件事讲得更具体一些。下面是很多人一开始会构建的“bad agent”。它实际上只是一个伪装成更高级系统的工具增强型 chatbot。
“Bad agent”——不要这样做
# 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.
这段代码对于任何真实世界任务而言都是根本有问题的。它只能执行一个动作，然后就结束了。它无法规划多步骤旅行，因为它没有记忆。搜索航班之后，它会立刻忘记自己做过什么，因此无法预订酒店或寻找餐厅。输出只是一个字符串，这对 UI 毫无用处。而且它没有我们讨论过的任何安全特性：没有预算、没有对日期格式的验证，也没有在崩溃时恢复的方式。
一个最小可行 Agent——带 Guardrails 构建
现在，我们来看看如何正确地构建它。这个示例使用了一个名为 pydantic-ai 的库，它旨在强制执行我们一直在讨论的规则：结构化状态、经过验证的工具，以及内置预算。
# 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(description="e.g., '9:00 AM' or 'Afternoon'")
name: str = Field(description="Name of the activity or place")
description: str = Field(description="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(description="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(description="A catchy name for the trip")
summary: str = Field(description="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
instructions=(
"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",
destination="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())
这与我们构建的第一个版本有很大不同。我们来拆解一下为什么这是一个真正的 agent。
结构化状态：
TripContext是一个显式的、带类型的对象，充当 agent 的 memory。它通过 dependency injection 被干净地传递，而不是被揉进聊天历史里。结构化输出： agent 被强制返回一个
TripItinerary对象。不再需要解析字符串。Pydantic 会验证 LLM 的输出；如果输出有误，框架会自动告诉模型修正错误。预算与 Guardrails：
retries=2对 agent 可重试失败步骤的次数设置了硬性上限。在工具内部抛出ModelRetry是一种让工具向 agent 表示可恢复失败的方式，从而给 agent 一个自我纠正的机会。工具验证： 这个库会自动根据类型提示验证传给工具的所有参数。LLM 不能在不被捕获的情况下发送格式错误的日期或缺失参数。
可观测性： 通过使用 Logfire 这类工具进行 instrumentation，每一次 LLM 调用、工具调用和 token 成本都会被自动 trace，这对调试来说非常关键。
恢复与 dependency injection： 这类框架通常支持 durable execution，也就是说你可以 checkpoint 状态，并在崩溃后恢复。此外，
RunContext为工具访问 HTTP client 这类共享资源提供了一种干净方式，而无需使用全局变量。
Agent 可能出问题的诸多方式
构建一个健壮的 agent，本质上主要是在预判失败。下面是我经常见到的常见 failure modes，以及如何防护。
无限循环： agent 卡住了。也许它认为自己需要天气信息，调用了工具，但随后又决定自己仍然需要天气信息。最简单的修复方式是设置硬性的
max_steps预算。如果它在 10 步内还没解决问题，大概率也解决不了。Tool Misuse：LLM 为你的工具幻觉出参数。它可能在期望整数的地方发送字符串，或者干脆编造参数。修复方式是严格的验证。使用 Pydantic 或 JSON Schema 之类的东西，在执行工具调用 之前 验证参数。如果验证失败，就向 agent 返回错误消息，让它可以自我纠正。
成本失控： agent 不断重试一个持续失败的工具。这会很快烧光你的 API 预算。修复方式是实现带 exponential backoff 的重试上限。连续失败 3 次后，停止并报告失败。对于关键系统，你可以使用 circuit breaker pattern：如果某个工具的错误率过高，就临时禁用它。
非持久化执行： 服务器崩溃，agent 的全部进度被清空。正如我们讨论过的，唯一真正的修复方式是 checkpointing。每完成一步之后，都要序列化 agent 的状态并保存到数据库。你的 agent runner 应该设计为能够从 DB 加载状态并恢复执行。
Prompt injection： 恶意用户构造 prompt，诱导你的 agent 使用工具造成危害。例如，如果你有一个
run_bash_command工具，他们可能试图让它运行rm -rf /。这里的修复方式是多层防护。首先，对可使用的工具设置严格的 allowlist。其次，实现 policy checks。在执行工具之前，检查参数或行动本身是否违反安全策略。永远不要在没有 human in the loop 的情况下，让 agent 访问强大且具有破坏性的工具。
你看不见的东西，就无法修复
当 agent 做出意外行为时，你需要能够弄清楚原因。黑盒是不可能调试的。良好的可观测性是不可协商的。
Logging：要足够详细。对于每一次运行，你都应该记录：
agent 在每一步的决策（它选择了哪个工具，以及如果可能的话，为什么选择）。
每次工具调用的确切输入。
每次工具调用的确切输出（确保对敏感数据进行脱敏）。
步数和时间戳。
Metrics：你的 dashboard 应该跟踪高层趋势：
每次运行的平均步数。这个数字上升可能意味着你的 agent 正在变得更低效。
工具错误率。某个具体工具的错误突然飙升，说明那里有问题。
每次运行成本。要严格跟踪 token 使用量和 API 成本。
端到端延迟。用户需要等多久才能得到答案？
Tracing：对于调试单次运行，trace 非常宝贵。使用 OpenTelemetry 这样的工具，为整个 agent 运行创建一个 span。在其中，为每个“decide”步骤和每次工具调用创建子 span。这会让你准确可视化时间花在哪里，并定位瓶颈。
给下一个项目的 Checklist
把它放在手边。在你交付任何称为“agent”的功能之前，先过一遍这份 checklist。如果你无法勾选所有项，可能值得重新思考设计。
我们是否有一个明确的、独立于聊天历史的状态 schema？
我们是否有绝对停止条件（例如 max_steps、timeouts 或 cost budgets）？
所有工具输入和输出是否都根据 schema 进行了验证？
工具执行是否具备 idempotent 特性，以便重试不会造成重复操作？
agent 的运行是否可以在崩溃后从 checkpoint 恢复？
我们是否记录了每一次决策和工具调用，以便调试？
接下来
把定义弄清楚是第一步。它给了我们共同语言和一组原则，用于构建更可靠的系统。但真正的挑战在于实现细节。我们在示例中使用的简单状态对象，很快就会变得混乱。
在下一篇文章中，我会讨论如何使用计算机科学中最古老也最可靠的工具之一来管理这种复杂性：状态机。事实证明，把 agent 看作一个可以遍历的 graph，是让它们真正可调试并准备好进入生产环境的最佳方式之一。