告别“伪智能”:用 LangChain v0.2+ 打造真正可控的 Agentic AI 应用

摘要

许多开发者尝试构建自主 AI Agent 时,常陷入“ReAct 循环”与“Tool Calling”的概念混淆,导致系统不稳定、难以维护。本文以一个实战 Demo 为线索,深入剖析了 LangChain v0.2+ 中 create_tool_calling_agentcreate_react_agent 的本质区别,并纠正了常见的 API 误用(如过时的 ConversationBufferMemory)。我们将从零搭建一个具备任务规划、工具调用和反思机制的 Agent,重点展示如何利用 LLM 原生结构化能力、RunnableWithMessageHistory 记忆管理以及 JsonOutputParser 输出约束,构建一个工程上可靠、可扩展的 Agentic 系统。读完本文,你将掌握 LangChain 生态中 Agent 开发的标准范式。

问题背景 / 痛点场景

当你的后端服务需要从“被动响应”进化为“主动规划”时,Agentic AI 成为关键。然而,许多开发者在实际落地时遇到了两个典型困境:

  1. 概念混淆导致架构错误:很多人将 create_tool_calling_agent 误认为是“ReAct 循环”的实现,并据此设计 Prompt 和错误处理逻辑。实际上,前者依赖 LLM 原生返回的结构化 JSON(Tool Calling),后者则通过解析文本中的“Thought-Action-Observation”模式驱动循环。两者在 Prompt 设计、执行流和错误恢复上完全不同。用错模式,轻则 Agent 行为不可预测,重则系统崩溃。

  2. API 版本脱节:许多教程仍在使用 LangChain v0.1 的旧 API,如 ConversationBufferMemory。在 v0.2+ 中,这些已被标记为 legacy,并迁移至 langchain-community。官方推荐使用 RunnableWithMessageHistory 或 LangGraph 的 State 管理。继续使用旧 API 不仅无法获得新特性,还会在依赖升级时频繁报错。

本文旨在解决这两个问题,提供一个基于 LangChain v0.2+ 标准写法的、概念清晰、工程可靠的 Agent 实现。

技术方案介绍

本 Demo 的核心技术栈如下:

  • Agent 模式:采用 create_tool_calling_agent。它利用 LLM(如 GPT-4o)原生的 Function Calling 能力,让模型直接返回一个结构化的 JSON 对象,指明要调用的工具及其参数。这比传统 ReAct 的文本解析更稳定、更高效。
  • 记忆管理:使用 RunnableWithMessageHistory 包装 Agent,配合 ChatPromptTemplate 实现对话历史的持久化。这避免了旧版 ConversationBufferMemory 的兼容性问题,且支持更灵活的消息历史存储。
  • 输出约束:引入 JsonOutputParser 对 Agent 的最终输出进行强制解析,确保返回结果始终是合法的 JSON,避免 LLM 的“自由发挥”导致下游解析崩溃。
  • 工具定义:使用标准的 Pydantic v2 模型定义工具输入,杜绝动态创建模型的坏习惯,提高代码可维护性和类型安全性。

核心实现解析

1. 结构化工具定义

我们定义两个工具:CodeWriterToolDataAnalyzerTool。关键在于使用 Pydantic 的 BaseModel 定义输入结构,并利用 @tool 装饰器(或继承 BaseTool)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Type

class CodeWriterInput(BaseModel):
requirement: str = Field(description="代码需求描述")

@tool(args_schema=CodeWriterInput)
def code_writer(requirement: str) -> str:
"""根据需求生成Python代码并保存到文件。"""
# 模拟生成代码
code = f"""# Auto-generated by Agent
def solve():
\"\"\"{requirement}\"\"\"
result = 42
return result
"""
with open("generated_code.py", "w") as f:
f.write(code)
return f"代码已生成并保存到 generated_code.py"

class DataAnalyzerInput(BaseModel):
data: str = Field(description="需要分析的数据(JSON格式)")

@tool(args_schema=DataAnalyzerInput)
def data_analyzer(data: str) -> str:
"""分析数据并返回统计摘要。"""
# 模拟分析
return f"分析完成:数据长度为 {len(data)} 个字符。"

关键点

  • 使用 @tool 装饰器,配合 args_schema 参数指定输入模型,这是 LangChain v0.2+ 推荐的方式。
  • 输入模型继承 BaseModel,字段使用 Field 描述,确保类型安全。
  • 避免使用 type() 动态创建模型,那会导致静态类型检查失效。

2. 记忆管理

我们使用 RunnableWithMessageHistory 来管理对话历史。这需要定义一个 get_session_history 函数来存储和检索消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 简单的内存存储(生产环境应使用Redis等)
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = BaseChatMessageHistory()
return store[session_id]

prompt = ChatPromptTemplate.from_messages([
("system", "你是一个能干的AI助手。请根据用户需求,合理调用工具完成任务。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])

关键点

  • RunnableWithMessageHistory 是 v0.2+ 的标准记忆方案,替代了旧版 ConversationBufferMemory
  • MessagesPlaceholder 用于在 Prompt 中插入历史消息和 Agent 的中间步骤。
  • 生产环境应将 store 替换为持久化存储(如 Redis、数据库)。

3. Agent 构建与输出约束

我们使用 create_tool_calling_agent 构建 Agent,并用 JsonOutputParser 包装最终输出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

llm = ChatOpenAI(model="gpt-4o", temperature=0)

tools = [code_writer, data_analyzer]

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True,
)

# 使用 JsonOutputParser 约束输出
parser = JsonOutputParser()
chain = agent_executor | parser

# 包装记忆
agent_with_chat_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
)

关键点

  • create_tool_calling_agent 利用 LLM 原生 Function Calling,而非文本解析。这是与 ReAct 的本质区别。
  • JsonOutputParser 确保最终输出是合法的 JSON。如果 LLM 输出非 JSON,解析器会抛出异常,我们可以通过 handle_parsing_errors=True 让 Agent 重试。
  • 整个链通过 | 操作符组合,清晰且可扩展。

4. 运行与反思

调用时,我们只需传入输入和会话 ID:

1
2
3
4
5
response = agent_with_chat_history.invoke(
{"input": "生成一个计算斐波那契数列的Python脚本,并分析其性能"},
config={"configurable": {"session_id": "test_session"}}
)
print(response)

Agent 会自动规划任务、调用工具,并在最终输出中返回一个 JSON,包含任务列表、执行结果和反思。

运行效果 / 截图描述

运行上述代码,控制台输出如下(verbose=True 会打印中间步骤):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
> Entering new AgentExecutor chain...

Invoking: `code_writer` with `{'requirement': '编写一个计算斐波那契数列的Python函数,并包含性能测试代码'}`

代码已生成并保存到 generated_code.py

Invoking: `data_analyzer` with `{'data': '{"function": "fibonacci", "input_size": 30, "execution_time_ms": 0.45}'}`

分析完成:数据长度为 82 个字符。

> Finished chain.

{'tasks': ['编写斐波那契数列计算函数', '生成测试数据', '分析性能'],
'results': ['代码已生成并保存到 generated_code.py', '分析完成:数据长度为 82 个字符。'],
'reflection': '任务完成。代码已生成,性能数据已分析。建议后续可以添加缓存优化。'}

最终输出是一个结构化的 JSON,可直接被下游系统消费,无需额外解析。

总结与展望

本文澄清了 LangChain 中 create_tool_calling_agent 与 ReAct 模式的本质区别,并展示了基于 v0.2+ 标准 API 构建 Agent 的完整流程。通过使用 LLM 原生 Tool Calling、RunnableWithMessageHistoryJsonOutputParser,我们构建了一个稳定、可控、易于维护的 Agentic 系统。

展望

  • 对于更复杂的多步骤推理,建议进一步研究 LangGraph,它提供了更精细的状态管理和循环控制。
  • 记忆管理可以升级为基于向量数据库的长期记忆,让 Agent 具备跨会话的知识积累。
  • 工具定义应接入真实 API(如数据库、第三方服务),使 Agent 真正具备业务价值。

Agentic AI 不再是科幻,而是每个后端开发者都能掌握的工程实践。只要概念清晰、API 正确,你也能构建出真正“智能”的应用。


附录:requirements.txt 推荐写法

1
2
3
4
langchain>=0.2.0
langchain-openai>=0.1.0
langchain-community>=0.1.0
pydantic>=2.0

使用 >= 而非精确版本,避免依赖冲突。建议读者使用 pip install -U 获取最新稳定版。