为什么你的 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.# Its a tool-augmented chatbot pretending to be one.import openaiimport jsondef bad_travel_agent(prompt: str): A travel agent thats really just a single LLM call with a tool. print(fUSER: {prompt}) response openai.chat.completions.create( modelgpt-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 fFound flights from {args[origin]} to {args[destination]} print(fRESULT: {result}) else: print(fASSISTANT: {msg.content})# --- Why this is broken ---# 1. No state: It cant 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_keyos.getenv(GEMINI_API_KEY))model GoogleModel(gemini-2.5-pro, providerprovider)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 agents 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(descriptione.g., 9:00 AM or Afternoon) name: str Field(descriptionName of the activity or place) description: str Field(descriptionOne-sentence why its worth it) estimated_cost_usd: float Field(ge0)class DayPlan(BaseModel): day_number: int Field(ge1) date: str theme: str Field(descriptione.g., Old City Street Food) activities: list[Activity] Field(min_length2, max_length6) meals: list[Activity] Field(min_length1, max_length3)class TripItinerary(BaseModel): title: str Field(descriptionA catchy name for the trip) summary: str Field(description2-3 sentence overview of the trip) days: list[DayPlan] Field(min_length1) total_estimated_cost_usd: float Field(ge0) packing_tips: list[str] Field(min_length1, max_length5)# 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_typeTripContext, output_typeTripItinerary, # -- 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 travelers budget. Respect their preferences. Be specific - use real place names, not generic suggestions. ), model_settings{ temperature: 0.3, max_tokens: 4096, }, retries2, # 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(fWeather 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.texttravel_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( fOver 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 fRoute: {t.origin} → {t.destination}\n fDates: {t.start_date} to {t.end_date} ({nights} nights)\n fBudget: ${t.budget_usd:.0f} total\n fTraveler notes: {t.traveler_notes}\n )# 7. RUN IT: With proper lifecycle managementasync def plan_trip(): async with httpx.AsyncClient(timeout30.0) as client: trip_ctx TripContext( originDelhi, destinationKyoto, start_datedate(2026, 4, 1), end_datedate(2026, 4, 5), budget_usd2000.0, traveler_notesI love temples and street food. Vegetarian., http_clientclient, ) result await travel_agent.run( Plan my trip!, depstrip_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 的输出如果输出有误框架会自动告诉模型修正错误。预算与 Guardrailsretries2对 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 MisuseLLM 为你的工具幻觉出参数。它可能在期望整数的地方发送字符串或者干脆编造参数。修复方式是严格的验证。使用 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%免费】
为什么你的 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.# Its a tool-augmented chatbot pretending to be one.import openaiimport jsondef bad_travel_agent(prompt: str): A travel agent thats really just a single LLM call with a tool. print(fUSER: {prompt}) response openai.chat.completions.create( modelgpt-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 fFound flights from {args[origin]} to {args[destination]} print(fRESULT: {result}) else: print(fASSISTANT: {msg.content})# --- Why this is broken ---# 1. No state: It cant 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_keyos.getenv(GEMINI_API_KEY))model GoogleModel(gemini-2.5-pro, providerprovider)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 agents 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(descriptione.g., 9:00 AM or Afternoon) name: str Field(descriptionName of the activity or place) description: str Field(descriptionOne-sentence why its worth it) estimated_cost_usd: float Field(ge0)class DayPlan(BaseModel): day_number: int Field(ge1) date: str theme: str Field(descriptione.g., Old City Street Food) activities: list[Activity] Field(min_length2, max_length6) meals: list[Activity] Field(min_length1, max_length3)class TripItinerary(BaseModel): title: str Field(descriptionA catchy name for the trip) summary: str Field(description2-3 sentence overview of the trip) days: list[DayPlan] Field(min_length1) total_estimated_cost_usd: float Field(ge0) packing_tips: list[str] Field(min_length1, max_length5)# 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_typeTripContext, output_typeTripItinerary, # -- 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 travelers budget. Respect their preferences. Be specific - use real place names, not generic suggestions. ), model_settings{ temperature: 0.3, max_tokens: 4096, }, retries2, # 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(fWeather 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.texttravel_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( fOver 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 fRoute: {t.origin} → {t.destination}\n fDates: {t.start_date} to {t.end_date} ({nights} nights)\n fBudget: ${t.budget_usd:.0f} total\n fTraveler notes: {t.traveler_notes}\n )# 7. RUN IT: With proper lifecycle managementasync def plan_trip(): async with httpx.AsyncClient(timeout30.0) as client: trip_ctx TripContext( originDelhi, destinationKyoto, start_datedate(2026, 4, 1), end_datedate(2026, 4, 5), budget_usd2000.0, traveler_notesI love temples and street food. Vegetarian., http_clientclient, ) result await travel_agent.run( Plan my trip!, depstrip_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 的输出如果输出有误框架会自动告诉模型修正错误。预算与 Guardrailsretries2对 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 MisuseLLM 为你的工具幻觉出参数。它可能在期望整数的地方发送字符串或者干脆编造参数。修复方式是严格的验证。使用 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%免费】
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
