2026 年 AI Agent 实用指南:别再把 RAG 和 Workflow 叫成 Agent
2026/7/23 1:06:29 网站建设 项目流程

为什么你的 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 openaiimport jsondef 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 logfireimport asyncioimport httpxfrom dataclasses import dataclassfrom datetime import datefrom dotenv import load_dotenvimport osload_dotenv()from pydantic import BaseModel, Fieldfrom pydantic_ai import Agent, RunContext, ModelRetryfrom pydantic_ai.models.google import GoogleModelfrom pydantic_ai.providers.google import GoogleProviderprovider = 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.@dataclassclass 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 intravel_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.toolasync 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.toolasync 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_plaindef 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.instructionsdef 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 managementasync 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 恢复?
  • 我们是否记录了每一次决策和工具调用,以便调试?

接下来

把定义弄清楚是第一步。它给了我们共同语言和一组原则,用于构建更可靠的系统。但真正的挑战在于实现细节。我们在示例中使用的简单状态对象,很快就会变得混乱。

学AI大模型的正确顺序,千万不要搞错了

🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!

有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!

就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!

我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询