LangChain Agent实战：从create_agent到结构化输出与性能优化

1. 从“工具调用”到“自主决策”：Agent 的本质是什么？

如果你用过 LangChain 的LLMChain或者RunnableSequence，可能会觉得它们像是一个“听话的流水线”：你给一个输入，它按部就班地调用工具或模型，然后给你一个输出。整个过程是线性的、确定的。但当你开始接触Agent时，感觉就完全不一样了。它不再是那个被动的执行者，而更像是一个拥有“思考回路”的自主决策者。它会根据你的指令，自己决定下一步该做什么、用什么工具、什么时候该停下来。这种从“执行”到“决策”的转变，正是 Agent 技术的核心魅力，也是当前 AI 应用从“玩具”走向“生产力工具”的关键一步。

简单来说，一个 LangChain Agent 就是一个由大语言模型驱动的自主实体。它被赋予了一个目标（比如“帮我查一下今天北京的天气，然后根据天气推荐穿什么衣服”），以及一套可以使用的工具（比如“网络搜索”、“计算器”、“代码执行器”）。Agent 的核心工作流程是一个经典的“感知-思考-行动”循环：它接收用户的输入（感知），用大语言模型分析当前状态和目标（思考），决定调用哪个工具或直接给出答案（行动），然后根据工具返回的结果再次“思考”，循环往复，直到任务完成或认为无法继续。

这听起来很酷，但真正用起来，新手往往会遇到几个典型问题：Agent 动不动就陷入死循环，在一个步骤里打转；输出的结果格式乱七八糟，难以被下游程序处理；或者，你明明想监控它的思考过程，却只能看到一个最终结果，中间发生了什么一概不知。这些痛点，恰恰是 LangChain 1.x 中围绕create_agent、中间件、结构化与流式输出这些特性所要解决的核心问题。接下来，我们就抛开概念，直接进入实战，看看如何用这些特性构建一个既强大又好用的智能体。

2. 构建智能体的基石：深入理解create_agent与执行器

在 LangChain 的早期版本中，构建一个 Agent 可能需要你手动组装AgentExecutor、定义复杂的Agent类，过程比较繁琐。create_agent函数是 LangChain 1.x 提供的一个更高级、更声明式的 API，它旨在简化智能体的创建过程，让你能更专注于定义“做什么”，而不是“怎么做”。

2.1create_agent的核心参数与工作流

create_agent函数的核心是三个部分：大语言模型（LLM）、工具集（Tools）和提示词（Prompt）。我们来看一个最基础的创建示例：

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain import hub # 1. 准备大模型 llm = ChatOpenAI(model="gpt-4o", temperature=0) # 2. 定义工具 def search_wikipedia(query: str) -> str: # 这里简化，实际应调用 Wikipedia API return f"根据查询 '{query}'，模拟返回的百科摘要信息..." search_tool = Tool( name="WikipediaSearch", func=search_wikipedia, description="用于搜索维基百科获取事实信息。输入应为明确的搜索查询词。" ) def calculator(expression: str) -> str: try: # 安全警告：生产环境务必使用更安全的评估方式，如 ast.literal_eval 或专用库 result = eval(expression) return str(result) except: return "计算错误：表达式无效。" calc_tool = Tool( name="Calculator", func=calculator, description="用于执行数学计算。输入应为有效的数学表达式，如 '3 + 5 * 2'。" ) tools = [search_tool, calc_tool] # 3. 获取预定义的提示词模板（从 LangChain Hub） prompt = hub.pull("hwchase17/react") # 4. 创建智能体 agent = create_react_agent(llm, tools, prompt) # 5. 创建执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # 6. 运行智能体 result = agent_executor.invoke({"input": "爱因斯坦的出生年份加上100等于多少？"}) print(result["output"])

这段代码清晰地展示了流程。但create_react_agent内部做了什么？它实际上帮你封装了ReAct框架的逻辑。ReAct（Reasoning + Acting）是一种让 LLM 将思考（Reasoning）和行动（Acting）结合起来的范式。在提示词中，模型会被要求以Thought:、Action:、Observation:的格式进行输出。create_react_agent生成的智能体就内置了解析这种格式、调用对应工具、并将工具返回结果作为新Observation喂给下一轮思考的能力。

注意：create_react_agent是create_agent的一种具体实现，针对 ReAct 框架。LangChain 也支持其他类型的智能体，如 OpenAI Functions Agent、Conversational Agent 等，它们有各自的创建函数或方式。

2.2AgentExecutor：掌控循环的“导演”

AgentExecutor是真正驱动智能体运行的核心引擎。你可以把它想象成电影导演，而agent对象是主角（LLM）。导演负责喊“卡”和“开始”，控制着整个拍摄（执行）流程。它的几个关键参数直接决定了智能体的行为和稳定性：

• max_iterations与max_execution_time：这是防止智能体“鬼打墙”的最重要保险丝。max_iterations限制最大循环次数（默认通常为15），max_execution_time限制最大执行时间。一旦超过，执行器会强制终止并返回当前结果或错误。对于复杂任务，你可能需要调高max_iterations；对于简单任务，调低它可以节省成本和时间。
• handle_parsing_errors：当 LLM 的输出无法被解析为有效的工具调用指令时（比如格式错误、调用了不存在的工具名），这个参数决定如何处理。设为True时，执行器会将错误信息作为Observation反馈给 LLM，让它有机会纠正自己。设为False则会直接抛出异常。在开发调试阶段，强烈建议设为True。
• early_stopping_method：决定何时提前停止。常用的是“force”（达到迭代上限时强制停止）和“generate”（让 LLM 自己判断是否该最终输出答案了）。后者更智能，但可能增加开销。
• verbose：设为True时，会在控制台打印出完整的Thought/Action/Observation链，这是调试和理解智能体思考过程的必备利器。

一个常见的误区是只关注create_agent而忽略了AgentExecutor的配置。实际上，一个智能体是否“好用”，很大程度上取决于执行器的参数调优。例如，一个需要多步检索和分析的任务，如果max_iterations设得太低，任务可能半途而废。

3. 为智能体注入“可观测性”：中间件（Middleware）实战

当你把verbose=True打开，在控制台看到刷屏的Thought和Action时，可能会想：这些信息如果能被程序捕获、分析、甚至持久化到数据库该多好？或者，我想在每次调用工具前都做个日志记录，每次LLM思考后都计算一下 token 消耗。这就是中间件（Middleware）的用武之地。

在 LangChain 1.x 的上下文中，中间件允许你在智能体执行的生命周期中的特定节点插入自定义逻辑。它就像是给执行流程加装的“监听器”或“过滤器”。

3.1 理解执行流程与钩子点

要使用中间件，首先要理解AgentExecutor的执行流程。一个简化的 ReAct 循环如下：

1. 接收输入：获得用户问题。
2. LLM 思考：模型基于当前上下文（历史+问题）生成Thought和Action。
3. 解析动作：执行器解析出要调用的工具名和输入参数。
4. 执行工具：调用对应的工具函数，并获得结果Observation。
5. 更新上下文：将Observation加入到历史上下文中。
6. 判断循环：判断是否满足停止条件（找到答案、达到上限等）。若不满足，回到第2步。
7. 生成最终输出：满足停止条件后，让 LLM 或直接返回最终结果。

中间件可以在这些步骤之间挂载。LangChain 提供了BaseCallbackHandler作为基础的中间件接口，但更灵活的方式是直接为AgentExecutor或底层组件添加钩子。不过，在 1.x 中，一个更直接的方式是利用Runnable的配置（configurable）和with_config方法，或者为工具调用添加装饰器。

这里，我分享一个更实用、更“接地气”的方法：包装工具函数本身。这虽然不是标准的“中间件”模式，但能达到同样的效果，且理解起来更直观。

3.2 实战：为工具调用添加日志与监控

假设我们想记录每个工具被调用的时间、参数、结果以及耗时。

import time import logging from functools import wraps from typing import Any, Callable # 设置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def tool_logging_middleware(func: Callable) -> Callable: """一个简单的工具日志中间件装饰器""" @wraps(func) def wrapper(*args, **kwargs): tool_name = func.__name__ start_time = time.time() logger.info(f"[Tool Middleware] 开始调用工具 '{tool_name}'，参数: args={args}, kwargs={kwargs}") try: result = func(*args, **kwargs) end_time = time.time() duration = end_time - start_time # 对过长结果进行截断 result_preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) logger.info(f"[Tool Middleware] 工具 '{tool_name}' 调用成功，耗时: {duration:.2f}秒，结果预览: {result_preview}") return result except Exception as e: end_time = time.time() duration = end_time - start_time logger.error(f"[Tool Middleware] 工具 '{tool_name}' 调用失败，耗时: {duration:.2f}秒，错误: {e}", exc_info=True) raise e return wrapper # 使用装饰器包装我们之前的工具 @tool_logging_middleware def logged_search_wikipedia(query: str) -> str: # 模拟网络延迟 time.sleep(0.5) return f"根据查询 '{query}'，模拟返回的百科摘要信息：阿尔伯特·爱因斯坦出生于1879年3月14日。" @tool_logging_middleware def logged_calculator(expression: str) -> str: try: result = eval(expression) return str(result) except Exception as e: return f"计算错误：{e}" # 重新定义工具 tools_logged = [ Tool(name="WikipediaSearch", func=logged_search_wikipedia, description="搜索维基百科。"), Tool(name="Calculator", func=logged_calculator, description="执行数学计算。"), ] # 创建并使用带有“中间件”工具的智能体 agent_logged = create_react_agent(llm, tools_logged, prompt) executor_logged = AgentExecutor(agent=agent_logged, tools=tools_logged, verbose=False) # 关闭verbose，用我们的日志 print("=== 执行带日志的智能体 ===") result = executor_logged.invoke({"input": "爱因斯坦的出生年份加上100等于多少？"}) print(f"最终答案: {result['output']}")

运行这段代码，你不仅能在控制台看到智能体的最终输出，还能在日志中清晰看到：

INFO:__main__:[Tool Middleware] 开始调用工具 'logged_search_wikipedia'，参数: args=('爱因斯坦的出生年份',), kwargs={} INFO:__main__:[Tool Middleware] 工具 'logged_search_wikipedia' 调用成功，耗时: 0.50秒，结果预览: 根据查询 '爱因斯坦的出生年份'，模拟返回的百科摘要信息：阿尔伯特·爱因斯坦出生于1879年3月14日。... INFO:__main__:[Tool Middleware] 开始调用工具 'logged_calculator'，参数: args=('1879 + 100',), kwargs={} INFO:__main__:[Tool Middleware] 工具 'logged_calculator' 调用成功，耗时: 0.00秒，结果预览: 1979

这样一来，我们就实现了对工具层的监控。你可以将这个装饰器逻辑扩展，集成到监控系统（如 Prometheus）、数据库（记录每次调用）或进行权限校验、输入清洗等，这就是中间件思想的落地。

4. 从混乱文本到规整数据：驾驭结构化输出（Structured Output）

智能体默认的输出是文本字符串。这对于直接给人看没问题，但如果你想将智能体的回答作为输入，自动传递给下一个系统（比如存入数据库、触发另一个API、生成图表），文本格式就非常麻烦了。你需要写复杂的正则表达式或解析逻辑去提取信息，既脆弱又容易出错。

结构化输出（Structured Output）就是为了解决这个问题。它让 LLM 按照你预先定义好的格式（如 Pydantic 模型、JSON Schema）来输出数据。在 LangChain 中，这通常通过with_structured_output方法来实现。

4.1 为智能体定义输出模型

假设我们想让智能体在回答关于人物的问题时，不仅给出文本答案，还能结构化地返回人物的姓名、出生年份和一项主要成就。

from pydantic import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 1. 用 Pydantic 定义我们期望的输出结构 class PersonInfo(BaseModel): """关于一个人的结构化信息""" name: str = Field(description="人物的全名") birth_year: int = Field(description="人物的出生年份") key_achievement: str = Field(description="一项主要成就或贡献") summary: str = Field(description="基于查询的简要文本总结") # 2. 创建一个支持结构化输出的 LLM Chain（注意：并非所有模型都完美支持，GPT-4系列通常较好） structured_llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(PersonInfo) # 3. 创建提示词 structured_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个信息提取助手。请根据用户的问题，从提供的上下文中提取信息，并严格按照要求的结构输出。"), ("human", "用户问题：{question}\n\n相关上下文：{context}") ]) # 4. 创建链 extraction_chain = structured_prompt | structured_llm # 5. 运行 context = "阿尔伯特·爱因斯坦（Albert Einstein），出生于1879年3月14日，是德裔理论物理学家。他最为人所知的是提出了相对论（包括狭义相对论和广义相对论），并因此获得了1921年的诺贝尔物理学奖。他的质能方程 E=mc² 被誉为世界上最著名的方程之一。" question = "告诉我关于爱因斯坦的基本信息。" result = extraction_chain.invoke({"question": question, "context": context}) print(type(result)) # 输出：<class '__main__.PersonInfo'> print(result) # 输出类似： # name='阿尔伯特·爱因斯坦' birth_year=1879 key_achievement='提出了相对论（包括狭义相对论和广义相对论）' summary='阿尔伯特·爱因斯坦出生于1879年，是德裔理论物理学家，他最著名的成就是提出了相对论。'

现在，result是一个PersonInfo对象，你可以直接通过result.name、result.birth_year来访问属性，完美适配后续的程序化处理。

4.2 将结构化输出与智能体结合

那么，如何让一个使用了工具的智能体也输出结构化内容呢？一个常见的模式是“两阶段法”：

1. 阶段一：智能体执行。让智能体像往常一样运行，使用工具收集信息和推理，最终生成一个丰富的文本摘要。
2. 阶段二：结构化提取。将第一阶段生成的文本摘要，作为上下文，喂给另一个支持结构化输出的 LLM Chain（就像上面的extraction_chain），提取出最终的结构化数据。

# 假设我们有一个已经能完成复杂任务的智能体 executor def get_agent_summary(question: str) -> str: """使用智能体获取问题的文本摘要答案""" result = executor_logged.invoke({"input": question}) return result["output"] # 用户复杂问题 complex_question = "请比较爱因斯坦和牛顿的主要贡献，并总结他们的出生年份。" # 阶段一：智能体获取文本摘要 text_summary = get_agent_summary(complex_question) print("=== 智能体生成的文本摘要 ===") print(text_summary) # 阶段二：定义新的输出模型用于提取比较信息 class ScientistComparison(BaseModel): einstein_birth_year: int = Field(description="爱因斯坦的出生年份") newton_birth_year: int = Field(description="牛顿的出生年份") einstein_contribution: str = Field(description="爱因斯坦的主要贡献") newton_contribution: str = Field(description="牛顿的主要贡献") comparative_summary: str = Field(description="两者的比较总结") comparison_llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(ScientistComparison) comparison_prompt = ChatPromptTemplate.from_template( "请从以下文本中提取关于爱因斯坦和牛顿的结构化信息：\n\n{summary}" ) comparison_chain = comparison_prompt | comparison_llm # 从摘要中提取结构化信息 structured_comparison = comparison_chain.invoke({"summary": text_summary}) print("\n=== 提取的结构化比较信息 ===") print(f"爱因斯坦出生年: {structured_comparison.einstein_birth_year}") print(f"牛顿出生年: {structured_comparison.newton_birth_year}") print(f"爱因斯坦贡献: {structured_comparison.einstein_contribution}") print(f"牛顿贡献: {structured_comparison.newton_contribution}") print(f"比较总结: {structured_comparison.comparative_summary}")

这种方法结合了智能体的强大推理、工具调用能力和结构化输出的规整性，是构建生产级 AI 应用的常见模式。

5. 告别“黑盒”等待：实现流式输出（Streaming）

当你向一个复杂的智能体提出一个需要多步工具调用的问题时，如果等待十几秒后屏幕上才突然蹦出所有结果，体验会很差。用户不知道程序是卡住了还是在工作。流式输出（Streaming）就是为了改善这种体验，它允许你将智能体的思考过程（Thought）、行动（Action）、观察（Observation）以及最终答案，像打字一样实时地、一块一块地返回给前端。

在 LangChain 中，流式输出主要依赖于Runnable接口的.stream()或.astream()（异步）方法。对于AgentExecutor，我们需要确保其底层组件支持流式。

5.1 让智能体的“思考过程”流式输出

AgentExecutor本身可能不直接暴露最细粒度的流（如每个Thought的 token），但它可以流式返回每个步骤的完整输出块。更常见且实用的做法是，我们流式获取智能体的最终文本答案。同时，结合前面提到的中间件或回调，我们可以将中间步骤也实时推送出去。

首先，我们创建一个支持流式输出的执行器。关键是要使用支持流式的 LLM（如ChatOpenAI默认支持），并且以流式方式调用。

from langchain.agents import AgentExecutor, create_react_agent from langchain_core.runnables import RunnableConfig # 使用支持流式的LLM streaming_llm = ChatOpenAI(model="gpt-4o", temperature=0, streaming=True) # 创建智能体和执行器（工具沿用之前的） agent_stream = create_react_agent(streaming_llm, tools_logged, prompt) executor_stream = AgentExecutor(agent=agent_stream, tools=tools_logged, verbose=False, handle_parsing_errors=True) # 定义异步流式调用函数 import asyncio async def stream_agent_response(question: str): """异步流式获取智能体的最终输出""" print(f"提问: {question}") print("回答（流式）: ", end="", flush=True) full_answer = "" async for chunk in executor_stream.astream({"input": question}): # 注意：AgentExecutor的流式输出可能包含多个键，我们通常关心 "output" if "output" in chunk: content = chunk["output"] print(content, end="", flush=True) full_answer += content # 你也可以处理中间步骤，例如 chunk 中可能包含 "intermediate_steps" # elif "intermediate_steps" in chunk: # print(f"\n[中间步骤更新]: {chunk['intermediate_steps']}") print() # 换行 return full_answer # 运行异步函数 async def main(): answer = await stream_agent_response("珠穆朗玛峰的高度是多少米？") print(f"\n完整答案: {answer}") # 在 Jupyter 或异步环境中直接 await，在脚本中： # asyncio.run(main())

上面的astream会逐步返回包含output等字段的字典。但请注意，对于复杂的多步 Agent，output字段可能只在最后一步才出现。如果你希望看到更细粒度的Thought/Action/Observation流，需要配置自定义的回调处理器（AsyncCallbackHandler），这是一个更进阶的话题。不过，对于大多数需要“最终答案”流式输出的场景，上述方法已经足够。

5.2 实战：构建一个简单的实时问答终端

结合流式输出和中间件日志，我们可以构建一个更有趣的演示：

import sys import asyncio from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler class ThoughtStreamingHandler(StreamingStdOutCallbackHandler): """一个简单的回调处理器，尝试捕获并流式打印Thought等内容""" def on_llm_new_token(self, token: str, **kwargs) -> None: # 这个回调是针对LLM生成每个token的，但无法直接区分是Thought还是最终输出。 # 更精细的控制需要解析LLM的原始输出，这里做简单演示。 sys.stdout.write(token) sys.stdout.flush() # 创建一个带有流式回调的LLM llm_with_stream = ChatOpenAI( model="gpt-4o", temperature=0, streaming=True, callbacks=[ThoughtStreamingHandler()] # 添加回调 ) # 注意：将回调添加到LLM，而不是AgentExecutor。这样只能流式LLM的思考文本。 agent_with_thought_stream = create_react_agent(llm_with_stream, tools_logged, prompt) executor_with_thought_stream = AgentExecutor(agent=agent_with_thought_stream, tools=tools_logged, verbose=False) print("=== 智能体思考过程流式演示（LLM Token级）===") # 注意：由于AgentExecutor的工作方式，我们这里用invoke，但LLM内部的生成会通过回调流式出来。 # 这并不能完美流式化整个Agent步骤，但展示了可能性。 result = executor_with_thought_stream.invoke({"input": "计算圆周率π的前5位小数。"}) print(f"\n最终输出: {result['output']}")

重要提示：让AgentExecutor完美地流式输出每一步的Thought、Action、Observation是一个复杂任务，因为执行器需要协调LLM生成、工具调用、解析等多个环节。社区中常见的做法是使用LangGraph（另一个库，用于构建有状态的、多环节的AI工作流）来更精细地控制流程和流式。对于 LangChain 1.x 的AgentExecutor，获取最终答案的流式是稳定的，但获取中间步骤的流式通常需要自定义CallbackHandler并深入理解其内部事件。

6. 避坑指南与性能优化实战

在实际项目中集成 Agent，你会遇到各种预料之外的问题。下面是我从多个项目中总结出的常见“坑”及其解决方案。

6.1 坑一：智能体陷入死循环或无效动作

现象：智能体反复调用同一个工具，或者不断生成无意义的Thought，无法到达最终答案。

根因分析：

1. 工具描述不清晰：LLM 不理解工具的具体功能或输入格式。
2. 提示词（Prompt）不佳：没有明确指示停止条件，或者思考框架（如 ReAct）的指令不够强。
3. max_iterations设置不当：设得过高，放任了循环；设得过低，在复杂任务中提前终止。
4. 工具返回结果质量差：如果工具总是返回“未找到”或错误信息，LLM 可能不知道如何继续。

解决方案：

• 优化工具描述：描述要具体、包含示例。例如，不要写“搜索网络”，而是写“使用此工具搜索互联网获取最新信息。输入应为明确的关键词，如‘2024年奥运会举办城市’。”
• 强化提示词：在系统提示中明确加入“如果你已经获得了足够的信息来回答问题，请直接给出最终答案，不要再使用工具。”或“在最多进行X步推理后，你必须给出最终答案。”
• 配置执行器参数：合理设置max_iterations（例如 5-10 步用于简单QA，15-20 步用于复杂任务）。同时，可以启用early_stopping_method="generate"，让 LLM 自己判断是否应该停止。
• 完善工具功能：确保工具能处理边缘情况，并返回对 LLM 友好的信息。例如，搜索工具在无结果时应返回“未找到相关信息，请尝试其他关键词。”而不是空字符串或错误堆栈。

6.2 坑二：解析错误（Parsing Error）

现象：控制台出现OutputParserException，提示无法将 LLM 的输出解析为有效的动作。

根因分析：LLM 没有严格按照Action:后接JSON或指定格式输出。可能是模型能力问题，也可能是提示词中格式指令不够突出。

解决方案：

• 设置handle_parsing_errors=True：这是第一道防线，让执行器将错误反馈给 LLM 自我纠正。
• 使用更强大的模型：GPT-4 系列在遵循格式指令上通常比 GPT-3.5 好很多。
• 在提示词中强化格式：使用 ````json` 代码块包裹示例，或使用非常醒目的标记。
• 尝试不同的 Agent 类型：OpenAIFunctionsAgent或StructuredChatAgent可能比ReAct对格式错误更鲁棒，因为它们依赖于函数调用（Function Calling）这种更结构化的协议。

6.3 坑三：Token 消耗与响应速度慢

现象：任务简单但耗时很长，API 调用费用高昂。

根因分析：

1. 上下文（Context）膨胀：每次循环，所有历史Thought、Action、Observation都会作为上下文再次发送给 LLM，导致 token 数快速增长。
2. 工具响应慢：如果工具依赖外部 API（如网络请求、数据库查询），其延迟会直接加到每次循环中。
3. 不必要的复杂推理：LLM 可能进行了过于详细的推理。

优化策略：

• 上下文压缩与摘要：这是最有效的优化手段。不要将完整的原始历史全部传递。可以使用ConversationSummaryBufferMemory或自定义中间件，在每次循环后，将冗长的历史对话摘要成一段简洁的文字，再放入下一轮的上下文。LangChain 提供了ConversationSummaryBufferMemory，但将其无缝集成到 Agent 循环中需要一些技巧，通常需要自定义AgentExecutor或使用LangGraph。
• 工具优化：
  • 缓存：为工具添加缓存层（如functools.lru_cache），对相同参数的调用直接返回缓存结果。
  • 超时与重试：为网络工具设置合理的超时，并实现重试逻辑，避免单次失败阻塞整个流程。
  • 异步工具：如果工具是 I/O 密集型（如网络请求），将其改造成异步函数，并使用AsyncTool包装。然后在异步环境中运行智能体，可以显著提升并发性能。
• 模型选择：对于不需要顶级推理能力的步骤，可以考虑使用更小、更快的模型（如gpt-3.5-turbo）来承担部分工作，或者使用大模型生成计划，用小模型执行简单步骤。

6.4 一个综合优化示例：为智能体添加简易记忆摘要

下面演示一个简化版的思路，在工具调用后对历史进行摘要，以减少上下文长度：

from langchain.memory import ConversationSummaryBufferMemory from langchain_openai import ChatOpenAI summary_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 可以用小模型做摘要 memory = ConversationSummaryBufferMemory( llm=summary_llm, max_token_limit=200, # 摘要后的最大token数 memory_key="chat_history", return_messages=True ) # 注意：将memory集成到AgentExecutor中需要自定义Agent或使用特定的Agent类型（如ConversationalAgent）。 # 以下是一个概念性代码，展示思路： from langchain.agents import ConversationalChatAgent, AgentExecutor from langchain.tools import Tool # 假设我们有工具 tools = [search_tool, calc_tool] # 创建支持对话记忆的Agent提示词 from langchain.agents import load_tools from langchain.agents.conversational_chat.base import ConversationalChatAgent # 这种方法需要更复杂的设置，通常建议查阅最新LangChain文档中关于“Agent with Memory”的示例。 # 一个更直接但“笨”的办法是：在每一轮Agent执行后，手动将输入输出保存到memory，然后在下一轮调用前，从memory加载摘要作为系统提示的一部分。 def run_agent_with_memory(question: str): # 1. 从memory加载历史摘要 history = memory.load_memory_variables({})["chat_history"] history_summary = "" if history: # 这里简化处理，实际应使用memory的摘要功能 history_summary = "之前的对话摘要：" + "\n".join([msg.content for msg in history[-2:]]) # 取最后两条 # 2. 将历史摘要融入本次问题 enhanced_question = f"{history_summary}\n\n当前问题：{question}" # 3. 执行智能体（使用一个没有内置记忆的普通执行器） result = executor_logged.invoke({"input": enhanced_question}) # 4. 将本轮问答保存到memory memory.save_context({"input": question}, {"output": result["output"]}) return result["output"] # 测试 print("第一轮:") ans1 = run_agent_with_memory("爱因斯坦是谁？") print(ans1) print("\n第二轮（依赖历史）:") ans2 = run_agent_with_memory("他什么时候出生的？") # 理想情况下，智能体应该能利用历史中的“爱因斯坦”信息 print(ans2)

这个示例展示了思路，但生产环境需要更严谨的设计。对于复杂的记忆和状态管理，LangGraph是比基础AgentExecutor更强大和灵活的选择，它天然支持有状态的工作流和更精细的控制。