1. 项目概述为什么我们需要更“聪明”的Agent如果你已经用LangChain的Agent跑通了几个简单的任务比如查天气、算数学题可能会觉得它挺“听话”的。但当你试图让它处理更复杂的、需要多步骤协作或返回特定格式数据的任务时最初的兴奋感可能很快就会消失。你会发现默认的Agent有时像个“愣头青”你让它“分析一下这份财报并给出JSON格式的总结”它可能先给你来一大段散文式的分析最后才勉强塞给你一个格式乱七八糟的JSON对象或者干脆忘了这茬。又或者在需要长时间运行的任务中你只能干等着屏幕一片空白完全不知道它“思考”到哪一步了。这就是我们今天要聊的“高级玩法”要解决的问题。它不仅仅是调用几个API而是让Agent变得更可控、更专业、更像一个能融入生产流程的可靠组件。命名Custom Agent Name、结构化输出Structured Output和流式模式Streaming这三板斧正是提升Agent工业级可用性的关键。命名让你能清晰管理和调用不同的智能体结构化输出让LLM的“自由发挥”变得规整便于下游系统解析而流式模式则极大地改善了用户体验让等待过程变得可知、可控。这不仅仅是“炫技”而是当你真正想把AI Agent投入实际应用时绕不开的工程化课题。接下来我会结合大量实战代码和踩坑经验带你逐一拆解。2. 核心玩法一为你的Agent赋予专属身份与记忆2.1 为什么需要给Agent命名不止是“起个花名”刚开始接触时你可能会觉得给Agent起名字只是个可有可无的趣味功能。但在实际项目中尤其是当你有多个Agent协同工作比如一个负责检索一个负责分析一个负责格式化时命名就变得至关重要。首先清晰的命名是系统可维护性的基石。在日志中看到[DataAnalysisAgent] 开始处理用户请求...远比看到[Agent-0x7f9b1c] 开始处理...要直观得多。当出现错误时你能快速定位是哪个环节的Agent出了问题。其次命名与记忆Memory系统紧密关联。LangChain的ConversationBufferMemory或ConversationSummaryMemory等组件在存储对话历史时会以Agent的名字作为会话的标识符之一。这意味着你可以为不同的Agent创建独立的、隔离的对话记忆。例如一个名为“ResearchBot”的Agent可以专注于技术调研保留相关的上下文而另一个名为“CustomerServiceBot”的Agent则处理客服对话两者记忆互不干扰。from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain_community.tools import DuckDuckGoSearchRun from langchain_openai import ChatOpenAI # 初始化LLM和工具 llm ChatOpenAI(modelgpt-4, temperature0) search_tool DuckDuckGoSearchRun() # 为不同的Agent创建独立的内存 research_memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) research_memory.chat_memory.add_ai_message(我是ResearchBot专注于技术资料搜索与整理。) support_memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) support_memory.chat_memory.add_ai_message(我是CustomerServiceBot乐于为您解决产品使用问题。) # 创建两个具有不同名称和记忆的Agent这里使用简易示例实际需定义Prompt # 注意create_react_agent需要prompt此处为展示概念简化处理 research_agent AgentExecutor.from_agent_and_tools( agentcreate_react_agent(llm, [search_tool], promptNone), # 实际应用需传入完整prompt tools[search_tool], memoryresearch_memory, verboseTrue, handle_parsing_errorsTrue, agent_kwargs{name: ResearchBot} # 自定义Agent名称 ) support_agent AgentExecutor.from_agent_and_tools( agentcreate_react_agent(llm, [search_tool], promptNone), tools[search_tool], memorysupport_memory, verboseTrue, handle_parsing_errorsTrue, agent_kwargs{name: CustomerServiceBot} ) # 使用时对话历史会根据Agent名称隔离 # research_agent.invoke(帮我找找LangChain最新版本的特性和文档) # support_agent.invoke(我的账号登录不上去了)注意上面的代码示例中agent_kwargs{name: ...}是一种示意。在LangChain的标准AgentExecutor中并没有直接的name参数。更常见的做法是将Agent名称融入System Prompt中或者使用CustomAgent类并在其内部状态中定义名称。对于记忆隔离关键是使用不同的memory对象实例。实操心得在定义Agent的System Prompt时第一句话就明确其身份和职责范围例如“你是一个名为FinanceAnalyst的AI助手专门处理金融数据查询与格式化输出。” 这比通过代码参数设置更有效因为LLM会严格遵守Prompt中的指令。记忆隔离则通过实例化不同的Memory对象来实现。2.2 实现命名的两种实战路径路径一通过System Prompt深度定制。这是最有效、最主流的方式。你可以在构建Agent的Prompt模板时将名称、角色、职责、输出格式要求等都写进去。from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import create_tool_calling_agent, AgentExecutor system_prompt 你是一个名为DataFormatter的专用AI助手。 你的核心职责是接收用户提供的杂乱数据或文本严格按照要求的格式如JSON、CSV、Markdown表格进行清洗、整理和输出。 你自身不具备数据分析能力若用户请求分析你应明确告知职责边界并建议其联系DataAnalyst。 你的所有输出都必须遵循response_format指令。 当前对话历史{chat_history} prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 假设tools是已定义的工具列表 agent create_tool_calling_agent(llmllm, promptprompt, toolstools) agent_executor AgentExecutor(agentagent, toolstools, memorymemory, verboseTrue)路径二创建自定义Agent类。当你需要更复杂的控制逻辑比如根据名称路由任务、记录特定Agent的调用指标时就需要走这条路。from typing import Any, List, Optional, Tuple from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain.schema import AgentAction, AgentFinish from langchain.callbacks.manager import CallbackManagerForChainRun class NamedToolCallingAgent(BaseSingleActionAgent): 一个带有名称属性的自定义Agent。 name: str base_agent: BaseSingleActionAgent # 包装一个已有的Agent如工具调用Agent property def input_keys(self): return self.base_agent.input_keys def plan( self, intermediate_steps: List[Tuple[AgentAction, str]], callbacks: CallbackManagerForChainRun None, **kwargs: Any, ) - AgentAction | AgentFinish: # 在调用底层Agent前可以注入名称信息或记录日志 print(f[{self.name}] 正在规划下一步行动...) # 也可以修改kwargs比如在用户输入前加上Agent名称提示 # kwargs[input] f[{self.name}] 处理请求: {kwargs[input]} return self.base_agent.plan(intermediate_steps, callbackscallbacks, **kwargs) async def aplan( self, intermediate_steps: List[Tuple[AgentAction, str]], callbacks: CallbackManagerForChainRun None, **kwargs: Any, ) - AgentAction | AgentFinish: # 异步版本 print(f[{self.name}] (异步)正在规划下一步行动...) return await self.base_agent.aplan(intermediate_steps, callbackscallbacks, **kwargs) # 使用方式 base_agent create_tool_calling_agent(llm, prompt, tools) named_agent NamedToolCallingAgent(nameResearchBot, base_agentbase_agent) agent_executor AgentExecutor(agentnamed_agent, toolstools, memorymemory)常见问题自定义Agent类看起来复杂什么时候该用我的经验是如果你的需求只是简单的命名和日志用System Prompt就够了。只有当你要拦截并修改Agent的决策流程例如根据工具执行结果动态调整策略、或需要为不同Agent实现完全不同的plan逻辑时才需要继承BaseSingleActionAgent或BaseMultiActionAgent来自定义。否则过度设计会引入不必要的复杂度。3. 核心玩法二驯服LLM的输出——结构化输出实战3.1 从“散文家”到“程序员”结构化输出的必要性LLM天生是“散文家”它擅长生成连贯、自然的文本。但程序需要的是结构化的数据比如一个Python字典、一个JSON对象或者一个Pandas DataFrame。让LLM直接输出“{“name”: “Alice”, “age”: 30}”这样的字符串它可能会在中间加上解释或者漏掉引号导致json.loads()直接崩溃。LangChain通过response_format参数和Pydantic库的深度集成提供了优雅的解决方案。其核心思想是用代码里定义的数据结构Schema去约束和引导LLM的生成过程。这比在Prompt里写“请输出JSON”要可靠得多。3.2 基于Pydantic的强类型结构化输出这是目前最推荐的方式。你首先用Pydantic定义一个数据模型Schema然后让LLM根据这个模型来生成内容。from langchain.output_parsers import PydanticOutputParser from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field from typing import List # 1. 定义你期望的数据结构 class PersonInfo(BaseModel): name: str Field(description人物的全名) age: int Field(description人物的年龄必须是整数) hobbies: List[str] Field(description人物的爱好列表) address: Optional[str] Field(defaultNone, description人物的住址可选) # 2. 创建输出解析器 parser PydanticOutputParser(pydantic_objectPersonInfo) # 3. 构建Prompt将格式指令自动注入 prompt_template 请根据以下用户描述提取出结构化信息。 {format_instructions} 用户描述{query} prompt PromptTemplate( templateprompt_template, input_variables[query], partial_variables{format_instructions: parser.get_format_instructions()}, # 关键 ) # 4. 组装链并调用 chain prompt | llm | parser query 我叫张三今年25岁喜欢打篮球和编程。住在北京海淀区。 try: result: PersonInfo chain.invoke({query: query}) print(f解析成功{result}) print(f类型{type(result)}) # class __main__.PersonInfo print(f姓名{result.name}) # 张三 print(f爱好{result.hobbies}) # [打篮球, 编程] except Exception as e: print(f解析失败{e})运行上述代码parser.get_format_instructions()会自动生成一段详细的文本指令嵌入到Prompt中告诉LLM必须严格按照Pydantic模型的JSON Schema来输出。LLM如GPT-4会努力遵守输出一个可以被parser正确解析的字符串。关键技巧字段描述Field description至关重要清晰、无歧义的description能极大提高LLM填充字段的准确率。把它当作给LLM的微型Prompt来写。处理可选字段和默认值像上面address字段使用了Optional[str]和defaultNone这样当用户描述中没提地址时LLM会输出null解析后得到None而不会报错。错误处理一定要用try...except包裹解析过程。即使有response_formatLLM偶尔还是会“抽风”输出格式错误。好的程序应该能优雅降级比如记录日志并返回一个默认值或友好错误。3.3 与Agent集成让Agent的输出也结构化上面的例子是简单链。在Agent中我们通常希望最终的答案是结构化的。这需要将结构化输出解析器与Agent的output_parser结合起来。from langchain.agents import AgentExecutor, create_react_agent from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.prompts import StringPromptTemplate from langchain.tools import Tool # 假设我们有一个工具get_current_weather def get_weather(city: str) - str: # 模拟工具调用 return f{city}的天气是晴朗25摄氏度。 weather_tool Tool( nameGetWeather, funcget_weather, description根据城市名查询当前天气 ) # 定义最终输出的结构 class AgentFinalAnswer(BaseModel): city: str Field(description查询的城市) weather_description: str Field(description天气描述) temperature: int Field(description温度整数) suggested_activity: str Field(description根据天气建议的活动) # 创建针对最终答案的解析器 final_parser PydanticOutputParser(pydantic_objectAgentFinalAnswer) # 自定义一个Prompt模板将结构指令融入Agent的思考框架中 agent_prompt_template 你是一个天气助手Agent。请使用工具获取天气信息然后对结果进行总结并严格按照指定格式输出最终答案。 你可以使用的工具 {tools} 工具调用格式Action: 工具名 Action Input: 输入参数 Observation: 工具返回结果 ...这个循环可以重复多次 当你需要给出最终答案时必须使用以下格式 Final Answer: {format_instructions} 开始 Question: {input} Thought: 我需要先思考用户想问哪个城市的天气。{agent_scratchpad} agent_prompt StringPromptTemplate.from_template( agent_prompt_template, partial_variables{ format_instructions: final_parser.get_format_instructions(), tools: weather_tool.name : weather_tool.description } ) # 创建Agent这里使用ReAct范式示例 agent create_react_agent(llm, [weather_tool], agent_prompt) # 注意这里输出解析器用标准的ReAct解析器它负责解析Action/Action Input。 # 最终答案的格式约束已经在Prompt里由final_parser在链的最后处理。 agent_executor AgentExecutor(agentagent, tools[weather_tool], verboseTrue) # 为了得到结构化输出我们需要在调用后手动用final_parser解析executor的输出 user_query 上海天气怎么样适合去外滩散步吗 raw_output agent_executor.invoke({input: user_query}) print(Agent原始输出:, raw_output[output]) # 尝试从原始输出中提取Final Answer部分并进行结构化解析 # 这里需要根据你的Agent实际输出格式做适配可能需要进行字符串匹配 try: # 假设原始输出中包含 Final Answer: {...} 这样的字符串 import re import json match re.search(rFinal Answer:\s*(\{.*\}), raw_output[output], re.DOTALL) if match: final_answer_str match.group(1) # 可能需要简单清理一下字符串 structured_answer final_parser.parse(final_answer_str) print(结构化最终答案:, structured_answer) else: print(未找到格式化的最终答案) except Exception as e: print(f解析最终答案失败: {e})重要提示将结构化输出与复杂Agent特别是ReAct这类需要多步推理的Agent完美结合是LangChain应用中的一个高级话题。上面的示例是一种简化演示。在实际中更成熟的方案可能是使用OpenAI Function Calling或Tools Calling驱动的Agent它们天生对结构化输出支持更好。使用create_structured_output_agent如果LangChain版本支持或自定义Agent将输出解析器深度集成到Agent的停止逻辑中。采用LangGraph这类框架将“生成结构化输出”作为一个独立的节点Node放在工作流的最后。踩坑记录最大的坑在于LLM的不稳定性。即使提供了最清晰的format_instructionsLLM尤其是低版本或小参数模型仍有可能输出格式错误的JSON比如缺少逗号、引号不匹配。解决方案除了用try-except还可以后处理清洗在解析前用简单的正则或字符串替换修复一些常见错误。使用更强大的模型GPT-4在遵循格式指令方面远强于GPT-3.5-turbo。设置更低temperature在需要严格输出的环节将temperature设为0或接近0减少随机性。采用“两阶段”法先让LLM输出一个“思考草稿”再让另一个LLM或同一个LLM在后续Prompt中根据草稿和格式要求生成最终答案。4. 核心玩法三告别“转圈等待”——流式输出全解析4.1 流式模式的价值从“黑盒”到“白盒”想象一下你问Agent一个复杂问题它沉默了20秒然后一次性吐出所有答案。在这20秒里用户可能以为程序卡死了甚至直接关闭了页面。流式输出Streaming就是为了解决这个体验痛点。它允许你将LLM生成的Token词元实时地、逐个或逐批地返回给前端。这样做的好处显而易见提升用户体验用户立即看到文字开始出现知道系统正在工作减少焦虑感。展示“思考过程”对于Agent你不仅可以流式输出最终答案还可以流式输出它的“思考”Thought和“工具调用”Action让用户仿佛看到AI的大脑在运转这非常酷也便于调试。实现中断机制由于输出是流式的前端可以在任何时候中断生成过程。4.2 实现LLM本身的Token流式传输这是最基础的流式LangChain对OpenAI等主流模型提供了开箱即用的支持。from langchain_openai import ChatOpenAI from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler # 方法1使用内置的StreamingStdOutCallbackHandler输出到标准输出 streaming_llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, streamingTrue, # 开启流式 callbacks[StreamingStdOutCallbackHandler()] # 添加流式回调 ) # 当你调用invoke或stream方法时Token会实时打印出来 # 注意对于简单的链这样就能看到流式效果 prompt 用100字介绍人工智能。 for chunk in streaming_llm.stream(prompt): # chunk是一个AIMessageChunk对象 # 如果你不需要回调处理器而是想自己处理每个chunk可以这样 # print(chunk.content, end, flushTrue) pass在Web应用中的实践在FastAPI或Flask中你需要创建一个生成器Generator将Token通过Server-Sent Events (SSE) 或 WebSocket 推送到前端。# FastAPI 示例 (简化版) from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate import asyncio app FastAPI() llm ChatOpenAI(modelgpt-3.5-turbo, streamingTrue, temperature0) async def stream_generator(prompt_text: str): 异步生成器用于流式响应 prompt ChatPromptTemplate.from_messages([(human, prompt_text)]) chain prompt | llm async for chunk in chain.astream({input: prompt_text}): # chunk.content 是逐步增加的字符串 if chunk.content: # 通常以data: ...格式发送给前端SSE yield fdata: {chunk.content}\n\n yield data: [DONE]\n\n # 结束标记 app.post(/chat/stream) async def chat_stream(request: Request): data await request.json() user_input data.get(message, ) return StreamingResponse( stream_generator(user_input), media_typetext/event-stream )4.3 Agent的流式输出思考、行动与答案的实时展示让Agent流式输出比单纯LLM流式要复杂因为Agent的输出包含多个部分Thought思考、Action调用哪个工具、Action Input工具输入、Observation工具返回结果、Final Answer最终答案。我们需要将这些部分都拆解成流式事件。LangChain提供了StreamingStdOutCallbackHandler的增强版——AgentStreamingStdOutCallbackHandler但它主要针对标准输出。对于Web应用我们需要自定义Callback Handler。from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import AgentAction, AgentFinish, LLMResult from typing import Any, Dict, List class CustomAgentStreamingCallback(BaseCallbackHandler): 自定义回调处理器用于捕获Agent的流式事件并推送到前端。 def on_llm_new_token(self, token: str, **kwargs: Any) - None: 处理LLM生成的新Token。 # 这里可以将token通过WebSocket或SSE发送出去 # 例如websocket.send_json({type: token, content: token}) print(fToken: {token}, end, flushTrue) def on_agent_action(self, action: AgentAction, **kwargs: Any) - None: 当Agent决定采取一个行动调用工具时触发。 # 发送工具调用开始事件 # websocket.send_json({type: action_start, tool: action.tool, input: action.tool_input}) print(f\n[行动] 调用工具: {action.tool}) print(f 输入: {action.tool_input}) def on_tool_end(self, output: str, **kwargs: Any) - None: 当工具执行结束时触发。 # 发送工具执行结果事件 # websocket.send_json({type: tool_result, output: output}) print(f[工具结果] {output}) def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) - None: 当Agent完成所有工作输出最终答案时触发。 # 发送最终答案事件 # websocket.send_json({type: final_answer, output: finish.return_values[output]}) print(f\n[最终答案] {finish.return_values[output]}) # 在创建AgentExecutor时传入这个回调 agent_executor AgentExecutor( agentagent, toolstools, verboseFalse, # 关闭默认的verbose输出用我们的回调 callbacks[CustomAgentStreamingCallback()], handle_parsing_errorsTrue ) # 调用时回调方法会自动触发 # agent_executor.invoke({input: 查询北京和上海的天气对比一下。})实战部署要点事件类型区分前端需要根据不同的type如tokenaction_starttool_resultfinal_answer来渲染不同的UI组件如思考气泡、工具调用卡片、逐步输出的文本。错误处理在流式过程中工具调用或LLM生成都可能出错。需要在回调中增加on_error方法并将错误事件流式推送给前端而不是让整个请求崩溃。性能考量频繁地发送小Token如每个字一发会给网络和后端带来压力。常见的优化是缓冲Buffering例如累积5-10个Token或等待一个短时间如100毫秒再发送一次在实时性和性能间取得平衡。与前端配合前端需要使用EventSourceSSE或WebSocket来接收事件流并动态更新界面。对于思考过程可以用斜体或灰色文字显示对于工具调用可以折叠或高亮显示。5. 三大高级玩法的融合实战与避坑指南5.1 构建一个具备完整特性的高级Agent现在让我们把命名、结构化输出和流式模式组合起来构建一个“豪华版”的天气查询Agent。import asyncio from typing import Any, Dict from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.tools import Tool from langchain.schema.runnable import RunnableConfig from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser # ---------- 1. 定义结构化输出模型 ---------- class WeatherComparison(BaseModel): city_a: str Field(description第一个城市名) city_b: str Field(description第二个城市名) weather_a: str Field(description城市A的天气情况) weather_b: str Field(description城市B的天气情况) comparison_summary: str Field(description对比总结哪个更适宜出行) suggested_activity: str Field(description综合建议的活动) # ---------- 2. 模拟工具 ---------- def get_weather(city: str) - str: 模拟天气查询工具返回固定格式字符串便于解析。 # 模拟不同城市的不同天气 weather_map { 北京: 晴朗28摄氏度微风, 上海: 多云25摄氏度东南风3级, 广州: 阵雨30摄氏度湿度85%, 深圳: 雷阵雨29摄氏度湿度80% } return weather_map.get(city, f{city}的天气信息暂不可知。) weather_tool Tool( nameGetCurrentWeather, funcget_weather, description根据城市名称查询当前天气详情。输入应为单个城市名。 ) # ---------- 3. 自定义流式回调 (简化版) ---------- class WeatherAgentCallback(BaseCallbackHandler): def on_agent_action(self, action: AgentAction, **kwargs): print(f\n [Agent] 正在执行工具: {action.tool}) print(f 输入: {action.tool_input}) def on_tool_end(self, output: str, **kwargs): print(f ✅ 工具返回: {output[:50]}...) # 截断长输出 def on_llm_new_token(self, token: str, **kwargs): print(token, end, flushTrue) def on_agent_finish(self, finish: AgentFinish, **kwargs): print(f\n\n [Agent] 任务完成) # ---------- 4. 构建Agent ---------- llm ChatOpenAI(modelgpt-4, temperature0, streamingTrue) # System Prompt中定义Agent身份和输出格式要求 system_message f你是一个名为WeatherComparerPro的高级天气对比助手。 你的职责是同时查询两个城市的天气进行对比分析并严格按照指定格式输出结论。 输出格式要求如下 {parser.get_format_instructions()} 请先使用工具获取天气信息然后进行分析。 prompt ChatPromptTemplate.from_messages([ (system, system_message), (human, 请对比一下{city1}和{city2}的天气给出出行建议。), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 使用OpenAI Tools Agent (推荐对结构化输出和流式支持更好) agent create_openai_tools_agent(llm, [weather_tool], prompt) agent_executor AgentExecutor( agentagent, tools[weather_tool], verboseFalse, callbacks[WeatherAgentCallback()], handle_parsing_errorsTrue ) # ---------- 5. 执行并解析 ---------- async def run_agent(): query_input {city1: 北京, city2: 上海} print(开始流式执行Agent...) # 注意为了流式输出Thought/Action我们使用astream_log。 # astream_log会返回包括中间步骤在内的所有日志事件。 async for event in agent_executor.astream_events(query_input, versionv1): kind event.get(event) if kind on_chain_start and event[name] Agent: print(\n Agent开始思考...) # 这里可以根据event的类型做更精细的处理如过滤只显示特定内容 # 为了演示我们主要依靠CallbackHandler输出 # 获取最终结果非流式部分 final_result await agent_executor.ainvoke(query_input) print(\n--- 原始输出 ---) print(final_result[output]) # 尝试结构化解析最终答案 try: # 注意由于我们在System Prompt中嵌入了格式指令LLM的最终输出应该已经是符合格式的字符串。 # 我们需要从输出中提取出JSON部分。这里假设输出就是纯JSON。 import json # 一个简单的提取方法实际中可能需要更健壮的解析 output_str final_result[output].strip() if output_str.startswith(json): output_str output_str[7:-3].strip() # 去除 json ... parsed_data WeatherComparison.parse_raw(output_str) print(\n--- 结构化解析结果 ---) print(f城市A: {parsed_data.city_a}) print(f城市B: {parsed_data.city_b}) print(f对比总结: {parsed_data.comparison_summary}) print(f建议活动: {parsed_data.suggested_activity}) except Exception as e: print(f\n⚠️ 结构化解析失败: {e}) print(尝试手动提取信息...) # 运行 asyncio.run(run_agent())5.2 避坑指南与经验总结在融合这些高级特性时我踩过不少坑这里总结出最关键的五点流式与结构化的冲突流式输出是Token-by-Token的而结构化输出如JSON需要完整的文本才能解析。你不能一边流式传输一边解析JSON。解决方案是分通道流式一个通道流式传输“思考过程”和“工具调用”的纯文本另一个通道在Agent完全结束后再将结构化的最终答案一次性或作为一个整体消息发送。前端需要区分这两种事件流。工具描述的重要性当Agent需要输出结构化数据时它所调用工具的description和参数必须尽可能精确。模糊的工具描述会导致LLM错误地使用工具进而无法获得生成结构化答案所需的数据。花时间打磨每个工具的文档字符串。错误处理的复杂性在流式Agent中错误可能发生在LLM生成、工具调用、输出解析任何一个环节。必须为每个环节设计独立的错误处理回调并向前端发送友好的错误事件如{“type”: “error”, “stage”: “tool_execution”, “message”: “...”}而不是抛出异常导致流中断。Prompt工程的精度结构化输出的成功率90%取决于Prompt。除了format_instructions在System Prompt中反复强调格式要求甚至给出一个完美的示例Few-shot能显著提升效果。例如“你必须输出如下格式的JSON不要有任何其他文字{“city_a”: “...”, “comparison_summary”: “...”}”。性能监控与调试开启流式后传统的打印日志会变得混乱。建议使用像LangSmith这样的追踪平台它可以可视化Agent的整个执行过程包括每个步骤的耗时、输入输出对于调试复杂的流式Agent至关重要。在本地开发时可以将关键事件如工具调用开始/结束输出到单独的文件或使用不同颜色的日志级别。最后记住这些高级特性是为了提升生产力和用户体验。不要为了用而用。如果一个内部工具不需要流式界面那就用简单的同步调用。如果一个任务的结果不需要被其他程序解析那就让LLM自由发挥。始终根据实际需求来选择技术方案。把这些玩法摸透你构建的LangChain Agent将不再是一个玩具而是一个真正强大、可靠、用户友好的AI应用核心。