在上一篇文章中,我们探讨了大语言模型(LLM)工程的基础概念、核心组件以及RAG(检索增强生成)架构的搭建。如果你已经成功搭建了一个能够回答问题的知识库,那么恭喜你,你已经迈出了坚实的第一步。然而,一个真正智能、能在复杂环境中自主行动的AI系统,远不止于“问答”。它需要能够感知环境、规划任务、调用工具并执行决策——这就是智能体(Agent)的世界。
本文将作为“大语言模型工程”系列的第二部分,我们将深入智能体开发的核心。从LangChain的AgentExecutor到新兴的LangGraph工作流,从简单的工具调用到复杂的多智能体协作,我们将手把手带你构建一个能够自主处理现实任务的智能体系统。无论你是希望将AI能力集成到现有业务中,还是想探索下一代AI应用的可能性,本文都将提供从理论到实践的完整路径。
1. 从问答到行动:理解智能体(Agent)
在传统的LLM应用中,模型更像一个“知识渊博的顾问”,你提问,它回答。但智能体赋予了LLM“手和脚”,使其成为一个能够主动执行任务的“智能员工”。
1.1 智能体的核心概念
一个智能体通常由以下几个核心部分组成:
- 大脑(LLM Core):负责理解指令、进行推理和做出决策。这是智能体的“思考”部分。
- 工具(Tools):智能体可以调用的外部函数或API。例如:搜索网络、查询数据库、执行计算、调用其他服务等。这是智能体的“手”。
- 记忆(Memory):存储与当前会话或任务相关的历史信息,包括对话历史、工具调用结果、中间状态等。这是智能体的“经验”。
- 规划器(Planner):对于复杂任务,智能体需要将其分解为一系列子步骤。规划器负责制定这个执行计划。
- 执行器(Executor):按照规划器的步骤,依次调用工具、处理结果,并决定下一步行动。
1.2 智能体 vs. 传统程序 vs. RAG
为了更清晰地理解智能体的定位,我们可以做一个简单的对比:
| 特性 | 传统程序 | RAG 系统 | 智能体系统 |
|---|---|---|---|
| 决策方式 | 确定性逻辑,if-else规则 | 基于检索内容的生成,一次调用 | 基于LLM推理的循环决策 |
| 灵活性 | 低,规则固定 | 中,回答基于检索内容 | 高,能根据环境动态调整策略 |
| 可执行动作 | 程序内定义的固定操作 | 仅文本生成 | 可调用任意外部工具 |
| 适用场景 | 流程固定、逻辑明确的任务 | 知识密集型问答、文档总结 | 复杂、多步骤、需交互的任务 |
| 示例 | 计算器、数据ETL脚本 | 公司知识库客服、论文解读助手 | 自动数据分析报告生成、旅行规划助手、自动化运维机器人 |
智能体的本质是将LLM的推理能力与外部工具的执行能力相结合,通过循环的“思考-行动-观察”过程,最终完成一个用户目标。
2. 环境准备与核心框架选择
在开始构建智能体之前,我们需要搭建开发环境并选择合适的框架。目前,LangChain和LangGraph是构建智能体最流行和强大的生态系统。
2.1 环境与依赖安装
我们使用Python进行开发。建议使用Python 3.10或更高版本,并创建一个虚拟环境。
# 创建并激活虚拟环境 (可选) python -m venv llm-agent-env source llm-agent-env/bin/activate # Linux/Mac # llm-agent-env\Scripts\activate # Windows # 安装核心依赖 pip install langchain langchain-community langgraph # 安装一个LLM提供商,这里以OpenAI为例(需自行准备API Key) pip install openai # 安装一些可能用到的工具库 pip install requests duckduckgo-search wikipedia2.2 框架简介:LangChain 与 LangGraph
- LangChain:提供了构建LLM应用的基础模块,如模型I/O、提示模板、链、记忆、索引以及智能体(Agent)。其
AgentExecutor是早期构建智能体的标准方式。 - LangGraph:建立在LangChain之上,用于构建有状态、多智能体(Multi-Agent)工作流。它用图(Graph)的概念来定义智能体之间的交互和状态流转,非常适合复杂、循环的任务。
如何选择?
- 对于单一智能体、任务相对线性的场景(如:先搜索再总结),使用LangChain的
AgentExecutor足够简单高效。 - 对于需要复杂编排、多角色协作、循环审批或具有严格状态管理的场景(如:一个智能体写代码,另一个智能体评审),LangGraph是更现代、更强大的选择。本文后续将重点介绍LangGraph。
2.3 配置LLM模型
你需要一个LLM的API Key。本文以OpenAI GPT-4为例,但你也可以轻松替换为Anthropic Claude、Google Gemini或本地部署的Ollama模型。
# 文件:config.py 或直接在代码开头设置 import os from langchain_openai import ChatOpenAI # 设置你的OpenAI API Key os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 初始化LLM。对于智能体任务,建议使用能力更强的模型,如gpt-4-turbo llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0) # temperature=0 使输出更确定,适合执行任务。可以调高以获得更多创造性。3. 构建你的第一个智能体:LangChain AgentExecutor
让我们从经典的LangChain智能体开始,直观感受“思考-行动”循环。
3.1 定义工具(Tools)
工具是智能体能力的延伸。我们定义两个简单的工具:一个计算器和一个网络搜索器。
# 文件:simple_agent.py from langchain.agents import tool from langchain_community.tools import DuckDuckGoSearchRun import math # 1. 自定义工具:计算器 @tool def calculator(expression: str) -> str: """执行数学计算。输入是一个数学表达式字符串,例如 ‘(3 + 5) * 2‘。""" try: # 警告:使用eval存在安全风险,仅用于演示。生产环境应使用安全库如`ast.literal_eval`或专门数学库。 result = eval(expression, {"__builtins__": None}, {"math": math}) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}" # 2. 使用社区工具:网络搜索 search = DuckDuckGoSearchRun() # 将工具放入列表 tools = [calculator, search]3.2 创建智能体(Agent)
我们使用LangChain提供的create_react_agent函数。ReAct(Reasoning + Acting)是一种经典的智能体范式,让模型在思考中显式地规划行动。
# 续 simple_agent.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor # 从LangChain Hub拉取一个适合ReAct的提示模板 prompt = hub.pull("hwchase17/react") # 创建智能体 agent = create_react_agent(llm, tools, prompt) # 创建执行器,它负责运行智能体的循环 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # verbose=True 会打印出详细的思考过程,便于调试。3.3 运行与观察
现在,让我们向智能体提问一个需要组合使用工具的问题。
# 续 simple_agent.py if __name__ == "__main__": question = "请先搜索‘Python之父的最新演讲主题’,然后用计算器计算这个主题标题的字符数乘以2。" try: result = agent_executor.invoke({"input": question}) print("\n=== 最终回答 ===") print(result["output"]) except Exception as e: print(f"执行出错: {e}")运行结果与过程分析(verbose输出节选):
> Entering new AgentExecutor chain... 我需要回答一个包含两个步骤的问题:先搜索,再计算。 我应该使用搜索工具来获取信息。 Action: DuckDuckGoSearchRun Action Input: Python之父的最新演讲主题 Observation: 根据搜索结果,Python之父Guido van Rossum最近的一次公开演讲主题可能是“Python in 2024: The State of the Union”... Thought: 我得到了演讲主题。现在需要计算这个主题的字符数,然后乘以2。主题是“Python in 2024: The State of the Union”。让我先计算字符数(包括空格和标点)。 Action: calculator Action Input: len('Python in 2024: The State of the Union') * 2 Observation: 计算结果: 78 Thought: 我已经完成了所有步骤。现在可以给出最终答案。 Final Answer: Python之父Guido van Rossum近期的一个演讲主题是“Python in 2024: The State of the Union”。该主题标题的字符数(包括空格和标点)为39,乘以2后结果是78。 === 最终回答 === Python之父Guido van Rossum近期的一个演讲主题是“Python in 2024: The State of the Union”。该主题标题的字符数(包括空格和标点)为39,乘以2后结果是78。你可以清晰地看到智能体的“思考(Thought)”、“行动(Action)”、“观察(Observation)”循环。这就是智能体工作的核心机制。
4. 进阶:使用LangGraph构建可控工作流智能体
AgentExecutor虽然简单,但对于复杂流程的控制力较弱。LangGraph通过“图”和“状态”的概念,让你能像设计流程图一样设计智能体的行为。
4.1 LangGraph核心概念:StateGraph 与 Nodes
在LangGraph中,你定义一个State(状态),它是一个字典,包含任务执行过程中的所有信息(如用户输入、中间结果、对话历史)。然后你创建多个Node(节点),每个节点是一个函数,负责处理状态的一部分。最后,你用StateGraph将这些节点连接起来,并定义流转条件(Edges)。
4.2 实战:构建一个带“人工审核”的写作智能体
假设我们需要一个智能体来撰写技术博客大纲,但在发布前需要经过“人工审核”节点(模拟人类干预)。
第1步:定义状态(State)
# 文件:writing_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END # 定义状态结构 class AgentState(TypedDict): # 用户原始需求 user_request: str # 由规划器生成的任务列表 task_list: List[str] # 当前执行的任务索引 current_task_index: int # 收集每个任务的执行结果 task_results: Annotated[List[str], operator.add] # 最终生成的大纲 outline: str # 审核意见 review_feedback: str # 最终输出 final_output: str第2步:创建节点(Nodes)
我们将创建四个节点:规划器(Planner)、执行器(Executor)、审核员(Reviewer)、修订器(Revisor)。
# 续 writing_agent.py from langchain_core.messages import HumanMessage, SystemMessage # 节点1:规划器 - 将用户需求分解为具体任务 def planner_node(state: AgentState): """分析用户需求,拆解成可执行的步骤。""" print(f"\n[规划器] 正在分析需求: {state['user_request']}") prompt = f""" 用户希望撰写一篇技术博客。需求是:{state['user_request']} 请将整个写作过程分解为3-4个清晰的、顺序执行的任务步骤。 以列表形式返回,例如: 1. 确定博客核心主题与目标读者。 2. 设计博客的章节结构(大纲)。 3. 为每个章节搜集关键要点。 4. 撰写开场白和结尾总结。 """ messages = [SystemMessage(content="你是一个专业的写作规划助手。"), HumanMessage(content=prompt)] response = llm.invoke(messages) task_list = [line.strip() for line in response.content.split('\n') if line.strip() and line[0].isdigit()] print(f"[规划器] 生成任务列表: {task_list}") return {"task_list": task_list, "current_task_index": 0, "task_results": []} # 节点2:执行器 - 执行当前任务 def executor_node(state: AgentState): """执行当前索引指向的任务。""" idx = state['current_task_index'] task = state['task_list'][idx] print(f"\n[执行器] 正在执行任务 {idx+1}: {task}") # 这里简单模拟执行,实际可以调用更复杂的工具或LLM prompt = f"基于写作需求‘{state['user_request']}’,请完成以下子任务:{task}。请提供详细、可操作的结果。" response = llm.invoke([HumanMessage(content=prompt)]) result = response.content print(f"[执行器] 任务结果: {result[:100]}...") # 打印前100字符 return {"task_results": [result], "current_task_index": idx + 1} # 节点3:审核员 - 模拟人工审核 def reviewer_node(state: AgentState): """审核生成的大纲,提出修改意见。""" outline = state['outline'] print(f"\n[审核员] 正在审核大纲...") prompt = f""" 你是一名资深技术编辑。请审核以下博客大纲: {outline} 请提供具体的修改意见,例如:结构是否合理?重点是否突出?有无遗漏关键点? 如果大纲质量很好,可以直接说‘通过审核’。 """ response = llm.invoke([HumanMessage(content=prompt)]) feedback = response.content print(f"[审核员] 审核意见: {feedback[:150]}...") return {"review_feedback": feedback} # 节点4:修订器 - 根据审核意见修改大纲 def revisor_node(state: AgentState): """根据审核意见修订大纲。""" outline = state['outline'] feedback = state['review_feedback'] print(f"\n[修订器] 正在根据意见修订大纲...") prompt = f""" 原始大纲: {outline} 审核意见: {feedback} 请根据审核意见,输出修订后的完整大纲。 """ response = llm.invoke([HumanMessage(content=prompt)]) revised_outline = response.content print(f"[修订器] 修订完成。") return {"outline": revised_outline, "final_output": revised_outline}第3步:构建图(Graph)并定义流转逻辑
# 续 writing_agent.py # 初始化图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("planner", planner_node) workflow.add_node("executor", executor_node) workflow.add_node("reviewer", reviewer_node) workflow.add_node("revisor", revisor_node) # 设置入口点 workflow.set_entry_point("planner") # 定义边(条件流转) # 规划后,进入执行阶段 workflow.add_edge("planner", "executor") # 执行器节点后,判断是否所有任务都执行完了 def decide_after_execute(state: AgentState): if state['current_task_index'] >= len(state['task_list']): # 所有任务完成,汇总结果生成大纲,然后进入审核 print("\n[决策] 所有任务执行完毕,开始生成大纲并进入审核。") # 简单地将所有任务结果拼接作为大纲初稿 draft_outline = "\n".join([f"任务{i+1}结果:{res[:200]}" for i, res in enumerate(state['task_results'])]) # 可以在这里让LLM基于结果提炼一个正式大纲,此处为演示简化处理 state['outline'] = draft_outline return "reviewer" else: # 还有任务,继续执行 return "executor" workflow.add_conditional_edges( "executor", decide_after_execute, { "reviewer": "reviewer", "executor": "executor", } ) # 审核后,根据意见决定是结束还是修订 def decide_after_review(state: AgentState): feedback = state['review_feedback'] if "通过审核" in feedback or "很好" in feedback: print("\n[决策] 审核通过,流程结束。") return END else: print("\n[决策] 审核未通过,进入修订环节。") return "revisor" workflow.add_conditional_edges( "reviewer", decide_after_review, { END: END, "revisor": "revisor", } ) # 修订后,再次进入审核(形成循环) workflow.add_edge("revisor", "reviewer") # 编译图 app = workflow.compile()第4步:运行工作流
# 续 writing_agent.py if __name__ == "__main__": # 初始化状态 initial_state = AgentState( user_request="写一篇关于如何使用LangGraph构建AI智能体的教程博客", task_list=[], current_task_index=0, task_results=[], outline="", review_feedback="", final_output="" ) print("=== 启动写作智能体工作流 ===") # 运行图 final_state = app.invoke(initial_state, config={"recursion_limit": 10}) print("\n=== 工作流执行完毕 ===") print("\n--- 最终生成的大纲 ---") print(final_state.get("final_output", "无输出"))这个例子展示了LangGraph如何清晰地管理复杂状态和循环逻辑(审核-修订-再审核)。你可以通过修改decide_after_review函数中的条件,或增加更多节点(如“发布器”、“数据分析器”),来构建极其复杂和强大的多智能体工作流。
5. 工程化与最佳实践
构建玩具智能体很有趣,但要将其用于生产环境,必须考虑工程化问题。
5.1 智能体的稳定性与可靠性
- 错误处理与重试:工具调用可能失败(网络超时、API限流)。必须在工具函数和智能体循环中加入健壮的错误处理与指数退避重试机制。
from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) @tool def call_unstable_api(query: str) -> str: response = requests.post(api_url, json={"query": query}, timeout=10) response.raise_for_status() return response.text - 超时控制:为整个智能体调用或单个工具调用设置超时,避免无限循环或长时间等待。
- 验证与过滤:对智能体生成的行动指令(如要调用的函数名、参数)进行严格验证,防止注入攻击或非法操作。
5.2 提示工程(Prompt Engineering)
智能体的表现极度依赖提示词。
- 系统提示(System Prompt):明确智能体的角色、职责、约束和输出格式。例如:“你是一个谨慎的助手,在调用工具前必须充分思考。你不能执行任何破坏性操作。”
- 工具描述:为每个工具编写清晰、准确的描述,说明其功能、输入格式和输出示例。
- 少样本示例(Few-Shot):在提示词中提供几个“思考-行动-观察”的完整示例,能显著提升智能体使用工具的准确性。
5.3 记忆(Memory)管理
智能体需要记住上下文。LangChain提供了多种记忆方案:
- 对话记忆:存储用户和AI的对话历史。
ConversationBufferMemory,ConversationSummaryMemory。 - 向量记忆:将历史信息存入向量数据库,需要时检索。适用于长上下文。
- 自定义记忆:在LangGraph的
State中,你可以设计任何结构来存储需要记忆的信息。
5.4 监控、评估与成本控制
- 日志记录:详细记录每个智能体的输入、输出、中间思考步骤、工具调用及结果。这对于调试和审计至关重要。
- 评估指标:定义如何评估智能体的表现。是任务完成率?用户满意度?还是结果准确性?需要设计自动化或人工评估流程。
- 成本估算:智能体可能进行多次LLM调用和工具调用。必须估算单次任务的平均Token消耗和API成本,并设置预算上限。
6. 常见问题与排查思路
在开发智能体时,你可能会遇到以下典型问题:
| 问题现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
| 智能体陷入循环,不断重复同一个工具调用。 | 1. 提示词未明确停止条件。 2. 工具返回的结果无法让LLM做出新决策。 3. 图(LangGraph)的边逻辑有误,形成死循环。 | 1. 在系统提示中强调“在获得最终答案后,必须输出Final Answer:”。2. 检查工具输出是否清晰、格式正确。 3. 调试LangGraph的 conditional_edges函数,确保有指向END的路径。 |
| 智能体拒绝调用工具,直接给出猜测性答案。 | 1. 工具描述不清,LLM不理解何时使用。 2. 模型温度(temperature)过高,导致行为不稳定。 3. 示例(Few-Shot)不足。 | 1. 重写工具描述,明确使用场景和输入格式。 2. 将 temperature调低(如0.1)。3. 在提示词中加入正确调用工具的示例。 |
LangGraph报错RecursionLimit reached。 | 工作流图中存在无限循环,超过了默认的递归限制(通常为25)。 | 1. 检查图中是否存在未正确终止的循环(如审核-修订循环没有通过条件)。 2. 在 app.invoke()的config中增加{"recursion_limit": 50},但这只是临时解决,需找到循环根源。 |
| 工具调用失败(如网络错误)。 | 工具依赖的外部服务不稳定或参数错误。 | 1. 在工具函数内部实现重试和异常捕获。 2. 让智能体具备错误处理能力,在提示词中教导它“如果工具X失败,可以尝试工具Y或向用户报告”。 |
| 智能体输出的行动指令格式错误,无法解析。 | LLM没有严格按照要求的格式(如JSON,特定关键词)输出。 | 1. 使用LangChain的OutputParser来强制格式化输出。2. 在提示词中更严格地规定输出格式,并使用少样本示例。 |
7. 总结与展望:智能体开发的未来
通过本文,我们完成了从理解智能体概念,到使用LangChain构建简单智能体,再到利用LangGraph设计复杂工作流的完整旅程。智能体开发的核心在于将LLM的通用推理能力与专用工具的执行能力通过一个可编程的“循环控制器”有机结合。
下一步学习路线建议:
- 深入框架:研究LangGraph的更多特性,如持久化检查点(Checkpoint)、并行节点、分支与合并。
- 探索多智能体:构建多个具有不同角色的智能体(如“研究员”、“写手”、“评审员”),让它们通过消息传递进行协作。
- 集成真实工具:将智能体与你公司的内部API、数据库、云服务连接起来,解决实际业务问题。
- 评估与优化:建立一套评估体系,用数据驱动的方式迭代提示词、工具集和工作流设计。
- 关注新兴架构:了解如Microsoft的AutoGen、CrewAI等其他多智能体框架,以及模型上下文推理(MCP)等新协议。
智能体技术正在快速发展,从自动编码助手、数据分析机器人到完全自主的虚拟员工,其应用场景无限广阔。成功的智能体工程不再是简单的API调用,而是涉及提示工程、软件架构、可靠性设计和人机交互的综合学科。现在,你已经掌握了核心的构建方法,接下来就是发挥创意,用代码将智能体的潜力转化为现实价值的时候了。