🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Claude 随心用,限时 5 折。 👉 点击领海量免费额度
1. 从“对话”到“行动”:为什么现在必须理解 AI Agent
别再只把大模型当成一个更聪明的聊天机器人了。如果你还在用“用户提问-模型回答”的单轮对话模式来思考 AI 应用,那你可能已经落后了。真正的变化在于,AI 正在从“回答问题”转向“完成任务”。这就是 AI Agent 的核心。
一个 Agent 不是一次性的问答机,而是一个能自主规划、调用工具、循环执行直到完成目标的数字员工。它处理的是“长时程任务”——比如,你让它“帮我研究一下 LangChain 1.0 的新特性并写一份报告”,它需要自己决定先去搜索、然后整理信息、最后生成结构化的内容。这个过程涉及多轮思考、决策和行动,这就是所谓的 ReAct(Reasoning + Acting)循环。
为什么现在必须关注这个?因为技术栈已经就位。大模型的理解和推理能力足够支撑这种循环,而像 LangChain 这样的框架,则把构建这种循环的“脏活累活”标准化了。你不用自己从头写状态管理、循环控制、工具调度的复杂逻辑。对于开发者来说,这意味着你可以把精力从“如何让 AI 跑起来”转移到“让 AI 做什么”上。
这篇文章适合两类人:一是对 AI 应用开发感兴趣,但觉得直接调用 API 只能做简单问答的开发者;二是已经听说过 Agent,但被各种概念和框架搞晕,不知道如何下手的技术人。我会带你拆解 LangChain 实现 Agent 的核心机制,并从一个可运行的实战例子开始,让你理解从“对话”到“行动”的转变到底是怎么发生的。
2. LangChain 如何成为 Agent 的“脚手架”:核心机制拆解
直接调用大模型 API 就像给你一块最先进的芯片,但 LangChain 帮你造好了主板、电源和散热系统。它的价值不在于让模型更聪明,而在于为模型的“非确定性”行为提供一个确定性的、可管理的运行环境。
2.1 裸调 API 的三大短板
很多人会问,我直接用requests库调用 OpenAI 的接口不行吗?对于构建 Agent 来说,确实不够。主要卡在三个地方:
- 无法优雅处理循环:Agent 的核心是 ReAct 循环。你需要让模型思考(Thought)、决定行动(Action)、执行工具(Tool)、观察结果(Observation),然后再思考。自己写这个循环,意味着要手动拼接上下文、管理对话历史、判断终止条件,代码会迅速变得臃肿且难以调试。
- 难以支撑长时任务:一个研究任务可能运行几分钟,调用十几次工具。你需要考虑任务的持久化(万一中间断了怎么办)、状态恢复、以及超时控制。这不是几行
try...except能解决的。 - 缺乏可观测性:这是最关键的。Agent 的决策路径是非确定性的,你无法像调试普通代码一样设断点。如果 Agent 跑偏了或者卡住了,你根本不知道它“心里”在想什么,哪一步出了问题。没有运行轨迹的追踪,上线就等于开盲盒。
LangChain,特别是其底层的 LangGraph 运行时,就是为了解决这些问题而生的。它把循环、状态、工具调用和观测能力,封装成了标准化的组件。
2.2 Agent 的四大核心组件
理解 LangChain 的 Agent,可以把它想象成一个团队:
- LLM(大脑):负责理解和推理。它接收当前状态(用户问题+历史+工具结果),然后输出“思考”和“决策”。
- Agent(调度员):这是 LangChain 框架的核心逻辑。它根据 LLM 的决策,决定下一步是调用工具还是结束任务,并管理整个循环流程。
- Tools(工具箱):任何 Agent 可以执行的具体操作。可以是一个搜索 API、一个数据库查询函数、一段代码执行器,甚至是另一个 Agent。Tools 是 Agent 与外部世界交互的手和脚。
- Memory(记忆体):负责记住对话历史、工具调用结果等状态。短期记忆服务于当前会话,长期记忆可以跨会话存储关键信息。
一个能工作的 Agent,就是这四部分协同的结果:大脑做决策,调度员管流程,工具箱干实事,记忆体保状态,循环跑到任务完成为止。
2.3 ReAct 循环:决策与执行的舞蹈
ReAct 是 Agent 的工作模式,LangChain 通过System Prompt来规范这个模式。一个典型的 ReAct 格式的思考过程如下:
Thought: 用户想了解 LangChain 1.0。我需要先获取最新信息。 Action: web_search Action Input: LangChain 1.0 new features Observation: LangChain 1.0 引入了新的 create_agent 函数,与 LangGraph 深度集成,强化了状态管理和可观测性... Thought: 我已经获得了关键信息,现在可以组织答案了。 Final Answer: LangChain 1.0 的主要新特性包括...框架的作用,就是自动解析 LLM 的输出,识别出Action和Action Input,然后去调用对应的 Tool,并把Observation结果塞回上下文,开启下一轮Thought。这个循环会一直持续,直到 LLM 输出Final Answer或达到最大循环次数。
3. 实战:从零构建你的第一个“研究助手” Agent
理论说再多不如跑一遍代码。我们来实现一个“智能研究助手”:你问它一个问题,它能自动去网上搜索信息,并整理成答案。
3.1 环境准备与依赖安装
首先,确保你的 Python 环境(建议 3.8+)已经就绪。我们将使用 LangChain 和 OpenAI 的模型。此外,为了让 Agent 能“动手”搜索,我们需要一个搜索工具。这里使用 Tavily,它是一个为 AI 优化的搜索引擎 API。
打开终端,安装必要的包:
pip install langchain langchain-openai tavily-python安装完成后,你需要准备两个 API Key:
- OpenAI API Key:用于调用 GPT 模型作为 Agent 的“大脑”。
- Tavily API Key:去 Tavily 官网注册一个免费账户即可获取。
将这两个 Key 设置为环境变量,这是更安全的做法:
# Linux/macOS export OPENAI_API_KEY='你的-openai-key' export TAVILY_API_KEY='你的-tavily-key' # Windows (PowerShell) $env:OPENAI_API_KEY='你的-openai-key' $env:TAVILY_API_KEY='你的-tavily-key'3.2 定义工具:给 Agent 装上“手”
Tools 是 Agent 能力的扩展。我们先定义一个网络搜索工具。
from langchain.tools import tool from tavily import TavilyClient # 初始化 Tavily 客户端 tavily_client = TavilyClient(api_key=“你的-tavily-key”) # 生产环境请从环境变量读取 @tool def web_search(query: str) -> str: """使用 Tavily 搜索网络信息。输入是一个搜索查询字符串,返回是搜索结果的摘要文本。""" try: # 执行搜索,限制结果数量以控制上下文长度 response = tavily_client.search(query=query, max_results=3) # 从结果中提取内容并拼接 contents = [result["content"] for result in response["results"]] return "\n\n".join(contents) except Exception as e: # 工具层处理异常,返回错误信息,避免异常直接抛给Agent导致崩溃 return f"搜索工具调用失败: {str(e)}"这里有几个关键点:
@tool装饰器:这是 LangChain 的标记,告诉框架这是一个可被 Agent 调用的工具。- 文档字符串(Docstring):非常重要!LLM 会根据这个描述来决定在什么情况下使用这个工具。描述要清晰准确。
- 异常处理:在工具内部捕获异常并返回友好信息。永远不要让工具调用抛出的异常直接中断 Agent 循环。
3.3 组装 Agent:连接大脑与工具
有了工具,接下来我们需要创建 LLM 实例,并用它将工具和提示词组装成一个可运行的 Agent。
from langchain_openai import ChatOpenAI from langchain.agents import create_agent # 1. 创建 LLM “大脑” llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 使用 gpt-4o-mini 性价比高,temperature=0 使输出更稳定 # 2. 创建 Agent agent = create_agent( model=llm, tools=[web_search], # 传入我们定义的工具列表 system_prompt="你是一个专业的研究助手。当用户提问时,你应该优先考虑使用搜索工具获取最新、最准确的信息,然后基于这些信息给出清晰、有条理的回答。", max_iterations=5, # 关键安全设置:防止无限循环 verbose=True # 设置为 True 可以在控制台看到详细的 ReAct 步骤,便于调试 )参数解读:
model: 这里使用ChatOpenAI,你也可以替换为其他 LangChain 支持的模型。tools: 一个列表,可以放入多个工具。Agent 会知道它“会”所有这些技能。system_prompt: 系统提示词,用于设定 Agent 的角色和行为准则。这是引导 Agent 行为的关键。max_iterations:必设参数。它限制了 ReAct 循环的最大轮数,是防止 Agent“陷入沉思”无法退出的安全阀。verbose: 调试神器。打开后,你能在控制台看到每一轮Thought、Action、Observation的详细信息。
3.4 运行与观察:看 Agent 如何工作
现在,让我们问它一个问题,并观察整个过程。
# 定义用户问题 user_query = "LangChain 1.0 版本相比之前有哪些主要改进?" # 运行 Agent result = agent.invoke({ "messages": [{"role": "user", "content": user_query}] }) # 打印最终答案 final_message = result["messages"][-1] print("="*50) print("用户问题:", user_query) print("="*50) print("Agent 最终答案:\n", final_message.content) print("="*50)当你运行这段代码(并设置verbose=True)时,在控制台会看到类似这样的输出:
> Entering new AgentExecutor chain... Thought: 用户想了解 LangChain 1.0 的改进。这是一个关于技术框架最新版本的问题,我需要获取准确和最新的信息。我应该使用搜索工具。 Action: web_search Action Input: LangChain 1.0 major improvements new features Observation: [这里会是 Tavily 返回的搜索摘要,包含 1.0 版本信息...] Thought: 根据搜索结果,LangChain 1.0 引入了新的 `create_agent` 高阶函数,与 LangGraph 运行时深度集成以支持复杂状态管理,并增强了可观测性工具 LangSmith。我现在有足够的信息来组织答案了。 Final Answer: LangChain 1.0 的主要改进集中在以下几个方面:1. **简化 Agent 创建**:提供了 `create_agent` 函数... 2. **强化状态管理**:与 LangGraph 深度集成... 3. **提升可观测性**:进一步完善了 LangSmith 的集成... > Finished chain. ================================================== 用户问题: LangChain 1.0 版本相比之前有哪些主要改进? ================================================== Agent 最终答案: LangChain 1.0 的主要改进集中在以下几个方面... ==================================================看到这个过程了吗?Agent 自动完成了“思考-搜索-再思考-回答”的完整流程。你没有写任何if语句来决定何时调用搜索,你只是告诉了 Agent 它有这个能力,并把决策权交给了模型。这就是“目标导向”编程与“指令式”编程的区别。
4. 从 Demo 到工程化:避开 Agent 开发的常见深坑
让一个 Agent 在 Jupyter Notebook 里跑通,和让它在一个生产系统中稳定可靠地运行,中间隔着一道巨大的鸿沟。以下是几个你必须提前考虑的工程问题。
4.1 控制循环:防止 Agent“鬼打墙”
最典型的问题是 Agent 陷入无限循环或无效循环。比如,它可能反复搜索同一个关键词,或者在一个问题上不停绕圈子。
解决方案:
- 设置
max_iterations:这是底线。根据任务复杂度,通常设置为 5-10 轮。对于简单问答,3轮可能就够了。 - 优化系统提示词(System Prompt):在提示词中明确指令,例如“如果经过两轮搜索仍无法获得新信息,则基于已有信息给出最佳答案并说明局限性”。
- 使用更智能的停止条件:除了轮数,还可以检查连续几轮的
Action是否相同,或者Observation是否不再变化,从而提前终止。
4.2 处理工具失败:构建韧性
工具调用可能因网络、API 限制、权限等问题失败。一个脆弱的工具会导致整个 Agent 崩溃。
解决方案:
- 工具内部重试与降级:像我们之前代码那样,在工具函数内部进行
try...except捕获,并返回有意义的错误信息。对于网络请求,可以加入指数退避重试。 - Agent 层面的容错:在 Agent 的提示词中加入处理工具失败的指导,例如“如果工具调用失败,请分析失败原因并尝试另一种方法或告知用户”。
- 使用备用工具:为关键操作提供多个工具实现(如不同的搜索 API),当一个失败时,Agent 可以尝试另一个。
4.3 管理上下文:应对 Token 爆炸
每次循环,历史对话、工具结果都会被追加到上下文。几轮之后,很容易超出模型的上下文窗口,导致性能下降或成本激增。
解决方案:
- 选择性记忆:不要将全部历史都塞给模型。使用
ConversationSummaryMemory或ConversationBufferWindowMemory等记忆组件,只保留最近几轮或总结后的摘要。 - 利用 LangGraph 的状态管理:LangGraph 允许你定义更精细的状态结构,只将必要的状态传递给下一个节点,而不是整个对话历史。
- 精简工具输出:让工具返回最精炼的结果。例如,搜索工具可以返回摘要而非全文。
4.4 实现可观测性:给 Agent 装上“黑匣子”
没有可观测性,Agent 就是一个盲盒。你无法优化、无法调试、更无法信任。
解决方案:
- 集成 LangSmith:这是 LangChain 官方的可观测性平台。它能记录每一次 LLM 调用、工具调用、Token 消耗、耗时和完整的链式轨迹。通过分析这些轨迹(Traces),你能清楚地看到 Agent 的决策路径在哪里出现了偏差。
- 结构化日志:在代码中关键节点(如每次工具调用前后、每次循环开始结束)打入带有唯一 ID 的详细日志,方便在分布式系统中追踪单个请求的全链路。
- 定义评估指标:对于生产系统,你需要定义什么是“好”。是任务完成率?是平均循环轮次?还是用户满意度?建立评估体系,才能持续迭代优化 Agent。
4.5 LangChain vs. LangGraph:如何选型?
这是常见的困惑。简单来说:
- LangChain(高层):提供开箱即用的高级抽象,如
create_agent。它适合快速原型验证、标准化场景。你关注的是“做什么”,而不是“怎么做”。 - LangGraph(底层):是一个基于状态图的运行时引擎。它让你能通过定义节点(Node)和边(Edge)来精确控制复杂的工作流、循环和状态转移。适合需要精细控制、长时运行、多 Agent 协作的生产级应用。
实战建议:从 LangChain 的create_agent开始。当你的需求变得复杂,例如需要自定义循环逻辑、实现复杂分支、或管理持久化状态时,再深入使用 LangGraph。事实上,create_agent的底层就是由 LangGraph 驱动的。
5. 超越搜索:扩展你的 Agent 能力蓝图
搜索只是一个起点。Agent 的真正威力在于它能集成各种工具,成为一个“全能助手”。下面是一些扩展思路和代码片段。
5.1 集成多种工具
一个强大的 Agent 应该拥有一个工具箱。让我们给它加上计算和文件读写能力。
import math from datetime import datetime @tool def calculator(expression: str) -> str: """执行数学计算。输入是一个数学表达式字符串,例如 '3 + 5 * 2'。返回计算结果。""" try: # 警告:使用 eval 有安全风险,仅作示例。生产环境应使用更安全的解析库如 `ast.literal_eval` 或专用数学库。 result = eval(expression, {"__builtins__": None}, {"math": math}) return str(result) except Exception as e: return f"计算失败: {str(e)}" @tool def write_memo(content: str, filename: str = None) -> str: """将内容写入备忘录文件。输入是要写入的文本和可选的文件名。返回操作结果。""" if filename is None: filename = f"memo_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" try: with open(filename, 'w', encoding='utf-8') as f: f.write(content) return f"内容已成功写入文件: {filename}" except Exception as e: return f"写入文件失败: {str(e)}" # 创建拥有多工具的 Agent advanced_agent = create_agent( model=llm, tools=[web_search, calculator, write_memo], system_prompt="你是一个高级助手,可以搜索信息、进行数学计算和记录备忘录。请根据用户需求合理使用你的工具。", max_iterations=8 ) # 测试多工具 Agent result = advanced_agent.invoke({ "messages": [{"role": "user", "content": "请先计算圆周率乘以10的平方,然后将结果和今天的日期一起记入备忘录。"}] }) print(result["messages"][-1].content)这个 Agent 会先调用calculator计算math.pi * 10**2,然后调用write_memo将结果和日期写入文件。
5.2 处理复杂任务与状态管理
对于需要多步骤、有依赖关系的复杂任务,简单的循环可能不够。这时需要用到 LangGraph 来定义明确的工作流。
假设我们要构建一个“数据报告生成 Agent”:1. 从数据库取数据 2. 用 Python 分析 3. 生成图表 4. 撰写报告。
# 这是一个概念性示例,展示 LangGraph 的思路 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 定义状态结构 class ReportState(TypedDict): question: str raw_data: Annotated[list, operator.add] # 累积数据 analysis_result: str chart_path: str final_report: str # 2. 定义各个节点函数(工具) def fetch_data(state: ReportState): # 模拟数据库查询 state["raw_data"] = [{"x": i, "y": i*i} for i in range(10)] return state def analyze_data(state: ReportState): # 模拟数据分析 data = state["raw_data"] state["analysis_result"] = f"分析完成,共 {len(data)} 条数据。" return state def generate_chart(state: ReportState): # 模拟生成图表(这里用保存文本模拟) state["chart_path"] = "chart_20231027.png" return state def write_report(state: ReportState): # 综合所有信息写报告 state["final_report"] = f"""报告摘要: 问题:{state['question']} {state['analysis_result']} 图表已生成:{state['chart_path']} """ return state # 3. 构建图 workflow = StateGraph(ReportState) workflow.add_node("fetch", fetch_data) workflow.add_node("analyze", analyze_data) workflow.add_node("chart", generate_chart) workflow.add_node("report", write_report) # 4. 定义边(执行顺序) workflow.add_edge("fetch", "analyze") workflow.add_edge("analyze", "chart") workflow.add_edge("chart", "report") workflow.add_edge("report", END) # 5. 编译并运行图 app = workflow.compile() initial_state = {"question": "展示近期的增长趋势"} final_state = app.invoke(initial_state) print(final_state["final_report"])通过 LangGraph,你将 Agent 的工作流从“自由发挥”变成了“有剧本的表演”,更适合对流程有严格要求的复杂生产任务。
5.3 为 Agent 添加记忆
为了让 Agent 在多次交互中记住上下文,你需要给它加上记忆。
from langchain.memory import ConversationBufferMemory from langchain.agents import AgentExecutor from langchain.agents import create_react_agent from langchain import hub # 拉取一个标准的 ReAct 提示词模板 prompt = hub.pull("hwchase17/react") # 创建记忆 memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # 使用更底层的 AgentExecutor 来集成记忆 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) tools = [web_search, calculator] agent = create_react_agent(llm, tools, prompt) # 创建执行器,传入记忆 agent_executor = AgentExecutor( agent=agent, tools=tools, memory=memory, verbose=True, max_iterations=5 ) # 现在进行多轮对话 result1 = agent_executor.invoke({"input": "北京今天的天气怎么样?"}) print(result1["output"]) result2 = agent_executor.invoke({"input": "那上海呢?"}) # Agent 会记得上一轮在问天气 print(result2["output"])ConversationBufferMemory会将所有对话历史保存在内存中。对于长对话,可以考虑ConversationSummaryMemory来压缩历史,节省 Token。
6. 评估与迭代:如何判断你的 Agent 是否“好用”
构建出 Agent 只是第一步。如何衡量它的表现并持续改进,是更大的挑战。这没有银弹,但可以从以下几个维度入手:
- 任务完成率:给定一批测试任务,有多少被成功完成了?这是最基础的指标。
- 平均循环轮次(Steps):完成一个任务平均需要多少次 ReAct 循环?轮次越少,通常意味着效率越高、成本越低。
- 工具调用准确率:Agent 选择使用工具的决定是否正确?有没有该用没用,或不该用乱用的情况?
- 最终输出质量:答案是否准确、完整、有用?这可以通过人工评估或使用另一个 LLM 作为裁判(LLM-as-a-Judge)来打分。
- 成本与延迟:处理单个任务消耗的 Token 数和总耗时是多少?这直接关系到可用性和运营成本。
建立评估流水线:准备一个涵盖不同场景和难度的测试用例集。每次对 Agent 的提示词、工具或模型进行修改后,都在这个测试集上运行,并记录上述指标。使用 LangSmith 可以自动化地追踪每次运行的轨迹和成本,是进行这种评估的绝佳工具。
迭代循环:基于评估结果,分析失败案例的轨迹(Trace)。是提示词不清晰?是工具描述不准?还是模型能力不足?然后有针对性地进行调整。Agent 的开发是一个典型的“数据驱动”和“观测驱动”的迭代过程。
别再只停留在讨论和观望。真正的理解始于动手。从安装 LangChain、定义一个搜索工具、创建一个能自动完成“思考-行动”循环的 Agent 开始。你会立刻感受到,编程的范式正在从“告诉计算机每一步怎么做”转向“告诉计算机你想要什么结果”。在这个过程中,最大的挑战可能不是技术,而是思维方式的转变——从控制到赋能,从确定到引导。你的角色,正在从细枝末节的流程编码者,转变为为智能体设定目标和边界的架构师。
🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Claude 随心用,限时 5 折。 👉 点击领海量免费额度