1. 项目概述:为什么说“框架是双刃剑”——LangChain 入门的真实代价
“Day 6:LangChain 入门——框架是双刃剑”,这个标题不是修辞,而是我踩了整整七天坑之后,在凌晨三点改完第17版agent调试日志时,盯着终端里一行tool_call_id报错信息写下的真实体会。它不煽情,不夸张,就是一句干巴巴的、带着咖啡渍和黑眼圈的实话。
LangChain 是什么?它不是魔法,也不是银弹。它是一套为 LLM 应用而生的胶水层——把模型(LLM)、工具(Tool)、记忆(Memory)、提示(Prompt)和执行流(Chain/Agent)强行拧在一起的工程化封装。它的核心价值在于:让你不用从零手写system prompt的 JSON 结构、不用手动解析{"tool_calls": [...]}的嵌套字典、不用自己维护消息历史的HumanMessage/AIMessage/ToolMessage三元组状态机。它用 Python 类、装饰器(如@tool)和链式调用(.bind_tools()→.invoke())把原本需要 200 行胶水代码才能跑通的“调用计算器再回答”流程,压缩成 30 行可读性尚可的代码。
但代价呢?代价就藏在标题那个“双刃剑”的“刃”字里。一面是效率之刃:它能让你在 15 分钟内搭出一个能联网查天气、能算加减乘除、能读取本地文件的 demo;另一面是抽象之刃:它用一层又一层的封装,把底层模型行为、网络协议细节、错误传播路径、异步流控逻辑全部藏了起来。你调用create_agent,它返回一个AgentExecutor对象;你传入query,它内部可能触发 5 次模型调用、3 次工具执行、2 次消息合并、1 次重试;而当最终结果是None或InvalidToolCall时,你根本不知道是模型没理解指令、是args字典键名拼错了、是tool_call_id在消息链中丢失了、还是PydanticToolsParser解析时类型转换失败。这种“黑盒感”,正是所有框架初学者在 Day 3 到 Day 7 之间集体崩溃的根源。
这跟ruoyi框架、springboot框架、pytest框架本质一样——它们都通过约定优于配置(Convention over Configuration)来换取开发速度,但同时也用抽象契约锁死了你对底层的掌控力。区别只在于,LangChain 的契约对象是“大语言模型”这个本身就不确定、不可控、会幻觉的黑箱。所以它的双刃性被放大了十倍:你越依赖它省事,它就越容易在最关键的地方给你一个无法 debug 的“优雅失败”。
适合谁看这篇?如果你是刚学完 Python 基础、正准备用 LLM 做点实际东西的开发者,或者你是业务方想快速验证一个 AI 功能点的技术负责人,又或者你是被老板催着“三天上线一个智能客服”的工程师——这篇就是为你写的。它不讲高深理论,不堆砌 API 文档,只讲我在真实项目里怎么把@tool装饰器从报错写到稳定、怎么让tool_call_id在消息流里不丢、怎么一眼看出create_agent返回的AgentExecutor到底卡在哪一步。它是一份带血丝的入门笔记,不是一份光鲜的宣传册。
2. 核心设计思路拆解:为什么 LangChain 必须用“工具调用”作为锚点
LangChain 的整个大厦,是围绕“工具调用(Tool Calling)”这个核心原语搭建起来的。这不是一个可选项,而是它的地基。要理解“框架是双刃剑”,必须先看清这个地基是怎么打的,以及它为什么既坚固又危险。
2.1 工具调用:LLM 与现实世界的唯一握手协议
大语言模型(LLM)本质上是一个概率生成器,它只能输出 token,不能执行任何外部操作。你想让它查数据库、发邮件、调用 API、甚至只是算个 119×8,它都做不到——除非你给它一个“协议”,告诉它:“当你想做这件事时,请按这个固定格式输出一段结构化文本,我会帮你解析、执行,并把结果塞回给你。” 这个协议,就是工具调用。
LangChain 实现了这个协议的标准化接口。它定义了Tool类(或@tool装饰器),要求你提供:
- 名称(name):一个字符串,模型输出时必须精确匹配;
- 描述(description):一段自然语言,告诉模型这个工具是干什么的(这是模型决策的关键依据);
- 参数模式(args_schema):一个 Pydantic 模型,定义参数名、类型、是否必填、默认值——这决定了模型输出的
argumentsJSON 字符串能否被安全反序列化。
这个设计背后有极强的工程逻辑:它把“模型能力”和“执行能力”彻底解耦。模型只负责“思考该调用哪个工具、传什么参数”,执行层(你的 Python 函数)只负责“拿到参数后干实事”。这种分离,让模型可以专注推理,让开发者可以专注业务逻辑,是框架的第一重价值。
2.2 双刃性的第一面:@tool装饰器——便利与陷阱并存
@tool是 LangChain 最诱人的糖衣炮弹。看这段代码:
from langchain_core.tools import tool @tool def add(a: int, b: int) -> int: """Adds a and b.""" return a + b短短 4 行,你就定义了一个工具。它自动帮你:
- 从函数签名推导
args_schema(a: int, b: int→{"a": 1, "b": 2}); - 从 docstring 提取
description("Adds a and b."); - 将函数包装成符合 LangChain
Tool接口的对象。
便利性毋庸置疑。但陷阱就藏在description和args_schema的自动生成里。
陷阱一:描述歧义导致模型乱调用"""Adds a and b."""这句话太单薄。模型看到What is 119 times 8 minus 20?,它可能认为add工具也能处理“minus”,于是生成{"name": "add", "args": {"a": 952, "b": -20}}。但你的add函数没有处理负数的逻辑,或者更糟——它根本没被设计用来处理减法。解决方案?把描述写得像产品需求文档一样具体:
@tool def add(a: int, b: int) -> int: """Perform addition of two integers. Only use this for pure addition (e.g., '5 + 3'). Do NOT use for subtraction, multiplication, or division."""陷阱二:参数类型推导失真@tool会把a: int当作硬性约束。但如果模型输出{"a": "119", "b": "8"}(字符串),PydanticToolsParser会尝试强制转换,失败则抛ValidationError。而这个错误不会直接暴露给用户,它会静默变成invalid_tool_calls,然后AgentExecutor可能选择重试或放弃。更稳妥的做法是显式定义args_schema,允许字符串输入并在函数体内做转换:
from pydantic import BaseModel, Field class AddInput(BaseModel): a: str = Field(description="First number as string, will be converted to int") b: str = Field(description="Second number as string, will be converted to int") @tool(args_schema=AddInput) def add(input: AddInput) -> str: try: result = int(input.a) + int(input.b) return str(result) except ValueError: return "Error: invalid number format"你看,@tool看似省事,实则把“描述清晰度”和“类型鲁棒性”这两个关键责任,悄悄转嫁给了开发者。框架给了你一把快刀,但没告诉你刀刃有多薄,稍一用力就会割伤自己。
2.3 双刃性的第二面:tool_call_id——消息流的命脉与断点
tool_call_id是 LangChain 消息系统里最精妙也最脆弱的设计。它不是一个随意生成的 UUID,而是一个消息链路的唯一标识符,其作用是将三个离散事件精准缝合:
- 模型输出的调用请求(
AIMessage.tool_calls[0].id); - 工具执行后的返回结果(
ToolMessage(tool_call_id=...)); - 模型收到结果后的最终回复(下一次
llm.invoke(messages))。
这个 ID 就像一根看不见的线,把“提问→思考→调用→执行→反馈→回答”这个闭环串起来。一旦这根线断了,整个 Agent 就会失联。
最常见的断点有三个:
- ID 生成不一致:
llm.bind_tools(tools)生成的tool_call_id是随机字符串,而你手动创建ToolMessage时如果用了不同的 ID(比如漏写了tool_call_id=参数),模型就找不到对应的结果。 - 消息顺序错乱:
messages.append(ToolMessage(...))必须紧接在AIMessage之后,且不能插入其他HumanMessage。否则模型会把ToolMessage当作新的人类输入,而不是对上一次调用的响应。 - ID 在流式处理中丢失:
astream()返回的是AIMessageChunk,它的tool_call_chunks是分片的。如果你只取第一个 chunk 的id,而后续 chunk 的args才完整,那么你用这个不完整的 ID 去创建ToolMessage,模型解析时就会失败。
这就是为什么tool_call_id是双刃剑的锋刃——它让复杂的异步交互变得可能,但也让调试变得极其反直觉。你不能只看最终结果,必须像侦探一样,逐帧检查每一条消息的id、content、tool_calls、tool_call_chunks,确保它们严丝合缝。框架把“如何管理状态”这个难题交给了你,而它只提供了一套极易出错的 ID 机制。
3. 核心实操要点解析:从create_agent到稳定运行的 7 个生死关
create_agent是 LangChain 官方推荐的“开箱即用”方案,但它绝不是“一键部署”。它是一个高度封装的工厂函数,内部组合了ChatPromptTemplate、LLM、Tools、AgentExecutor等多个组件。要让它真正跑起来,你必须亲手拆解、校准每一个齿轮。以下是我在真实项目中总结出的 7 个决定成败的核心关卡,每个都附有血泪教训。
3.1 关卡一:模型选型——别迷信gpt-4o-mini,先看tool_call支持度
很多新手一上来就pip install langchain-openai,然后ChatOpenAI(model="gpt-4o-mini")。这很危险。gpt-4o-mini确实便宜、快,但它对tool_call的支持是渐进式的。早期版本的gpt-4o-mini在处理复杂参数(如嵌套字典、列表)时,arguments字符串经常格式错误,导致PydanticToolsParser解析失败。
实操验证法:不要直接进AgentExecutor,先做原子测试。
# Step 1: 创建带工具的模型 llm = ChatOpenAI(model="gpt-4o-mini") llm_with_tools = llm.bind_tools([add, multiply]) # Step 2: 直接调用,观察原始输出 response = llm_with_tools.invoke("What is 3 * 12?") print("Raw response:", response) print("tool_calls:", response.tool_calls) print("invalid_tool_calls:", response.invalid_tool_calls)如果response.tool_calls是空列表,或者invalid_tool_calls里有内容,说明模型根本没理解你要它调用工具。这时,立刻降级到gpt-3.5-turbo-0125(官方明确支持tool_call的稳定版)或claude-3-haiku。等你的工具链完全跑通、逻辑验证无误后,再回来优化模型成本。记住:稳定性永远优先于性能和价格。
3.2 关卡二:工具注册——tools列表不是“扔进去就行”,必须满足命名一致性
create_agent的tools参数接收一个列表,比如[add, multiply]。但这里有个致命细节:add和multiply这两个函数名,必须和模型tool_calls输出中的"name"字段完全一致(大小写、下划线、连字符)。模型不会做任何模糊匹配。
踩坑现场:我曾定义了一个函数叫get_weather_by_city,但在@tool装饰时,因为手快写成了@tool(name="weather")。结果模型输出{"name": "weather", ...},而AgentExecutor在tools列表里找weather这个名字,却只找到get_weather_by_city,于是报错Tool not found: weather。
正确姿势:
- 方案 A(推荐):不指定
name,让@tool自动用函数名。@tool会把def get_weather_by_city(...)的函数名get_weather_by_city作为name。 - 方案 B:如果必须自定义
name,请确保tools列表里的对象name属性和模型输出的name严格一致,并且在create_agent之前打印出来确认:
for t in tools: print(f"Tool name: '{t.name}', description: '{t.description}'")3.3 关卡三:消息构造——ToolMessage的tool_call_id必须来自AIMessage.tool_calls
这是tool_call_id断点问题的终极解决方案。AgentExecutor内部的消息循环是这样的:
llm.invoke(messages)→ 返回AIMessage,其中tool_calls是一个列表;- 遍历
tool_calls,对每个call,执行tool.invoke(call["args"]); - 将结果
tool_output和call["id"]一起,创建ToolMessage(content=tool_output, tool_call_id=call["id"]); - 将
ToolMessage追加到messages后,再次llm.invoke(messages)。
关键点:ToolMessage的tool_call_id必须等于AIMessage.tool_calls[i]["id"],不能是AIMessage.tool_calls[i]["name"],也不能是str(uuid4()),更不能是call["id"]的副本(比如call["id"] + "_result")。它是唯一的、不可变的、由模型生成的“令牌”。
实操代码(务必抄作业):
messages = [HumanMessage(content="What is 3 * 12?")] ai_msg = llm_with_tools.invoke(messages) # 正确:直接使用 model 输出的 id for tool_call in ai_msg.tool_calls: # 注意:tool_call 是一个 dict,key 是 'name', 'args', 'id' selected_tool = {"add": add, "multiply": multiply}[tool_call["name"]] tool_output = selected_tool.invoke(tool_call["args"]) # ✅ 绝对正确:tool_call_id = tool_call["id"] messages.append(ToolMessage(content=str(tool_output), tool_call_id=tool_call["id"])) # 错误示例(会导致断点) # messages.append(ToolMessage(content=str(tool_output), tool_call_id=tool_call["name"])) # ❌ name 不是 id # messages.append(ToolMessage(content=str(tool_output), tool_call_id=str(uuid4()))) # ❌ 随机 id 无效3.4 关卡四:错误处理——invalid_tool_calls不是日志,是必须拦截的信号
AIMessage对象有两个关键属性:tool_calls和invalid_tool_calls。前者是模型“认为有效”的调用,后者是模型“认为无效”的调用(比如arguments不是合法 JSON、name不在已知工具列表中)。
很多教程忽略invalid_tool_calls,认为它只是调试信息。大错特错。在生产环境中,invalid_tool_calls是模型失控的第一个红灯。如果你不处理它,AgentExecutor会把它当作普通消息继续流转,最终可能导致无限循环或返回垃圾答案。
标准处理流程:
if ai_msg.invalid_tool_calls: # 记录日志,用于分析模型弱点 for invalid in ai_msg.invalid_tool_calls: logger.warning(f"Invalid tool call: {invalid.name}, args: {invalid.args}, error: {invalid.error}") # 构造一个友好的、引导性的回复,而不是让 Agent 自己瞎猜 fallback_response = "I'm sorry, I couldn't understand how to perform that action. Could you please rephrase your request?" messages.append(AIMessage(content=fallback_response)) else: # 正常流程:执行 tool_calls for tool_call in ai_msg.tool_calls: # ... 执行工具,追加 ToolMessage ...3.5 关卡五:流式响应——astream()不是简单的for chunk in ...,必须累积tool_call_chunks
astream()是实现“边思考边输出”的利器,但它的tool_call_chunks是分片的。一个完整的{"name": "add", "args": {"a": 3, "b": 12}}可能被切成 5 个 chunk:
{"name": "add", "args": ""}{"name": null, "args": "{"}{"name": null, "args": '"a": 3'}{"name": null, "args": ', "b": 1'}{"name": null, "args": "2}"}
如果你对每个 chunk 都单独处理,你会得到 5 个不完整的ToolMessage,全部失败。
正确累积法(官方推荐):
async def stream_with_tool_handling(query: str): messages = [HumanMessage(content=query)] llm_with_tools = llm.bind_tools(tools) # 初始化一个空的 AIMessageChunk 来累积 gathered = None async for chunk in llm_with_tools.astream(messages): if gathered is None: gathered = chunk else: gathered = gathered + chunk # 只有当 gathered.tool_call_chunks 不为空时才处理 if gathered.tool_call_chunks: # 这里 gathered.tool_call_chunks 是一个 list,每个元素是 ToolCallChunk # 我们需要等待它“完成”,即所有 index 都出现,且 args 是完整字符串 # 简化版:检查最后一个 chunk 的 args 是否以 '}' 结尾 last_chunk = gathered.tool_call_chunks[-1] if last_chunk.get("args", "").strip().endswith("}"): # ✅ 认为这是一个完整的 tool call,可以提取了 full_args_str = last_chunk["args"] # 解析 JSON,执行工具... break3.6 关卡六:少样本提示(Few-shot Prompting)——不是锦上添花,是救命稻草
当你的工具逻辑复杂(比如涉及运算顺序、多步骤调用),仅靠description很难让模型学会。这时,few-shot prompting就是你的救命稻草。它通过给模型展示 2-3 个“人类提问 → 模型调用工具 → 工具返回 → 模型最终回答”的完整范例,来教会模型“正确的调用模式”。
为什么它能破局?因为模型的tool_call行为,本质上是一种模式识别。description是文字规则,而 few-shot 是视觉示例。人教小孩算术,是先讲“加法是把两个数合起来”,还是先摆两堆积木说“1块加1块是2块”?答案不言而喻。
实操模板(直接复用):
from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # 构建一个高质量的 few-shot 示例 examples = [ # 示例1:简单加法 HumanMessage("What is 5 plus 3?"), AIMessage("", tool_calls=[{"name": "add", "args": {"a": 5, "b": 3}, "id": "1"}]), ToolMessage("8", tool_call_id="1"), AIMessage("5 plus 3 is 8."), # 示例2:复合运算(强调顺序) HumanMessage("What is 10 times 2 plus 5? Remember: do multiplication first, then addition."), AIMessage("", tool_calls=[{"name": "multiply", "args": {"a": 10, "b": 2}, "id": "2"}]), ToolMessage("20", tool_call_id="2"), AIMessage("", tool_calls=[{"name": "add", "args": {"a": 20, "b": 5}, "id": "3"}]), ToolMessage("25", tool_call_id="3"), AIMessage("10 times 2 is 20, and 20 plus 5 is 25.") ] system_prompt = """You are an expert calculator assistant. You MUST use the provided tools for all arithmetic operations. Never do math in your head. Always follow the order of operations (PEMDAS). Use the examples above as your exact template for how to structure your tool calls and final answers.""" few_shot_prompt = ChatPromptTemplate.from_messages( [("system", system_prompt)] + examples + [("human", "{query}")] ) # 组装链 chain = few_shot_prompt | llm_with_tools | PydanticToolsParser(tools=tools)3.7 关卡七:create_agent的替代方案——何时该亲手组装AgentExecutor
create_agent是一个“傻瓜模式”,它用默认的ChatPromptTemplate和AgentExecutor配置,帮你省去大量样板代码。但它最大的问题是:当它出错时,你不知道错在哪一层。
我的建议是:Day 1-3 用create_agent快速验证想法;Day 4 开始,必须亲手组装AgentExecutor。这样你才能真正掌控:
prompt:你可以自定义 system message,加入领域知识约束;llm:你可以精细控制temperature、max_tokens;tools:你可以动态增删工具;agent_executor:你可以注入自己的handle_parsing_errors回调,捕获OutputParserException。
亲手组装的最小可行代码:
from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 1. 自定义 Prompt(关键!) prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful AI assistant. Use the tools provided to answer questions. Be concise and accurate."), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 这是 AgentExecutor 自动注入消息历史的地方 ]) # 2. 创建 agent(注意:create_openai_tools_agent 是针对 OpenAI 的专用工厂) agent = create_openai_tools_agent(llm, tools, prompt) # 3. 创建 executor(这才是核心!) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 强烈建议开启,看清楚每一步在干什么 handle_parsing_errors="Check your output and make sure it's a valid JSON object with 'action' and 'action_input' keys.", # 自定义错误提示 ) # 4. 调用 result = agent_executor.invoke({"input": "What is 3 * 12?"}) print(result["output"])亲手组装,意味着你放弃了“一键”的便利,但换来了对整个执行流的绝对主权。这才是专业开发者的起点。
4. 完整实操流程:一个稳定、可调试的add/multiplyAgent 的诞生
现在,让我们把前面所有的知识点,整合成一个可直接运行、可调试、可扩展的完整项目。这个项目的目标很朴素:构建一个能准确回答What is 3 * 12? Also, what is 11 + 49?这类混合运算问题的 Agent。它不追求炫技,只追求稳定、透明、易维护。
4.1 环境准备与依赖安装
我们采用最精简、最可控的依赖组合。避免langchain大包带来的版本冲突。
# 创建虚拟环境(强烈推荐) python -m venv langchain_env source langchain_env/bin/activate # Linux/Mac # langchain_env\Scripts\activate # Windows # 安装核心依赖(只装必需的) pip install --upgrade pip pip install langchain-core langchain-openai python-dotenv pydantic # 安装 OpenAI SDK(注意:不是 langchain-openai,那是 LangChain 的封装) pip install openai # 验证安装 python -c "import langchain_core; print('langchain-core OK')" python -c "import langchain_openai; print('langchain-openai OK')"提示:
langchain-openai是 LangChain 官方的 OpenAI 适配器,它封装了openaiSDK 并提供了ChatOpenAI类。我们同时安装两者,是为了在需要时能直接调用底层openai.ChatCompletion.create进行深度调试。
4.2 工具定义:从@tool到Pydantic的演进
我们定义两个工具:add和multiply。但这次,我们不走捷径,而是用Pydantic显式建模,为未来扩展(比如增加subtract、divide)打下基础。
# tools.py from pydantic import BaseModel, Field, validator from typing import Optional from langchain_core.tools import BaseTool class MathOperationInput(BaseModel): """Base input schema for math operations.""" a: str = Field(description="First operand as string, e.g., '123'") b: str = Field(description="Second operand as string, e.g., '456'") @validator('a', 'b') def validate_number(cls, v): try: float(v) # 允许浮点数 except ValueError: raise ValueError(f"'{v}' is not a valid number") return v class AddInput(MathOperationInput): """Input for addition.""" pass class MultiplyInput(MathOperationInput): """Input for multiplication.""" pass class AddTool(BaseTool): name: str = "add" description: str = "Perform addition of two numbers. Input must be two strings representing numbers." args_schema: type = AddInput def _run(self, a: str, b: str) -> str: try: result = float(a) + float(b) # 如果结果是整数,返回整数字符串,避免 .0 if result.is_integer(): return str(int(result)) return str(result) except Exception as e: return f"Error in add: {str(e)}" class MultiplyTool(BaseTool): name: str = "multiply" description: str = "Perform multiplication of two numbers. Input must be two strings representing numbers." args_schema: type = MultiplyInput def _run(self, a: str, b: str) -> str: try: result = float(a) * float(b) if result.is_integer(): return str(int(result)) return str(result) except Exception as e: return f"Error in multiply: {str(e)}" # 导出工具列表,供主程序使用 tools = [AddTool(), MultiplyTool()]为什么用BaseTool而不是@tool?
BaseTool给你完全的控制权,_run方法可以包含任意异常处理逻辑;args_schema是一个真正的 Pydantic 模型,类型校验更严格;name和description是类属性,一目了然,不会因函数名变更而意外失效。
4.3 模型与执行器初始化:可插拔、可监控
# agent_setup.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.output_parsers.openai_tools import PydanticToolsParser from tools import tools # 加载环境变量 load_dotenv() # 1. 初始化模型(带详细日志) llm = ChatOpenAI( model="gpt-3.5-turbo-0125", # 稳定首选 temperature=0.0, # 确定性输出,便于调试 max_tokens=512, # 防止过长响应 verbose=True, # 开启详细日志 # logging_enabled=True # 如果需要更底层日志,可开启 ) # 2. 构建 Prompt(包含 Few-shot) examples = [ HumanMessage("What is 5 plus 3?"), AIMessage("", tool_calls=[{"name": "add", "args": {"a": "5", "b": "3"}, "id": "1"}]), ToolMessage("8", tool_call_id="1"), AIMessage("5 plus 3 is 8."), HumanMessage("What is 10 times 2 plus 5? Remember: multiplication first, then addition."), AIMessage("", tool_calls=[{"name": "multiply", "args": {"a": "10", "b": "2"}, "id": "2"}]), ToolMessage("20", tool_call_id="2"), AIMessage("", tool_calls=[{"name": "add", "args": {"a": "20", "b": "5"}, "id": "3"}]), ToolMessage("25", tool_call_id="3"), AIMessage("10 times 2 is 20, and 20 plus 5 is 25.") ] system_prompt = """You are a precise calculator assistant. Your job is to answer arithmetic questions by using the provided tools. Rules: 1. ALWAYS use the tools for calculation. Never compute in your head. 2. For multi-step problems, break them down into single operations (e.g., '10*2+5' becomes multiply(10,2) then add(20,5)). 3. Use the examples above as your exact template for tool calling and final answer format. 4. If you encounter an error, state it clearly in your final answer.""" prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), *examples, ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) # 3. 创建 Agent 和 Executor agent = create_openai_tools_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, handle_parsing_errors=lambda e: f"Failed to parse tool call. Error: {str(e)}. Please try again with a simpler question.", # 添加一个回调,用于记录每次调用的耗时 callbacks=[], ) # 4. 导出可调用对象 def run_agent(query: str) -> dict: """Run the agent and return structured result.""" try: result = agent_executor.invoke({"input": query}) return { "success": True, "output": result["output"], "intermediate_steps": result.get("intermediate_steps", []) } except Exception as e: return { "success": False, "error": str(e), "query": query } # 5. 一个用于深度调试的“裸调用”函数(绕过 AgentExecutor) def debug_llm_call(query: str, messages: list = None) -> dict: """Debug the raw LLM call with tools bound.""" if messages is None: messages = [HumanMessage(content=query)] llm_with_tools = llm.bind_tools(tools) response = llm_with_tools.invoke(messages) return { "raw_response": response, "tool_calls": response.tool_calls, "invalid_tool_calls": response.invalid_tool_calls, "content": response.content }4.4 主程序与调试入口:让一切变得可视化
# main.py from agent_setup import run_agent, debug_llm_call from tools import tools def main(): print("=== LangChain Agent Debug Console ===") print("Type 'quit' to exit, 'debug' to enter raw LLM mode, 'list' to show tools.") while True: try: query = input("\n> ").strip() if not query: continue if query.lower() == "quit": break if query.lower() == "list": print("Available tools:") for t in tools: print(f" - {t.name}: {t.description}") continue if query.lower() == "debug": print("Entering raw LLM debug mode. Enter your query:") debug_query = input("Debug > ").strip() if debug_query: debug_result = debug_llm_call(debug_query) print(f"Raw LLM Response: {debug_result['content']}") print(f"Tool Calls: {debug_result['tool_calls']}") print(f"Invalid Tool Calls: {debug_result['invalid_tool_calls']}") continue # 正常运行 Agent print("Running agent...") result = run_agent(query) if result["success"]: print(f"✅ Agent Output: {result['output']}") # 打印中间步骤,了解 Agent 思考过程 if result.get("intermediate_steps"): print("🔍 Intermediate Steps:") for i, step in enumerate(result["intermediate_steps"]): print(f" Step {i+1}: {step}") else: print(f"❌ Agent Failed: {result['error']}") except KeyboardInterrupt: print("\nGoodbye!") break except Exception as e