1. 从零到一AI Agent 到底是什么最近在技术社区和招聘网站上“AI Agent”这个词的热度持续攀升很多开发者朋友都跃跃欲试想抓住这波技术浪潮。但面对海量的概念、框架和教程往往感觉无从下手什么是Agent它和普通的大模型调用有什么区别LangChain和LangGraph又是什么关系本文旨在为你拨开迷雾提供一份系统、可落地的AI Agent实战指南。我们将从一个最简单的“单步工具调用”Agent开始逐步深入到使用LangGraph构建具备复杂工作流和记忆能力的多智能体系统。无论你是想快速入门了解核心概念还是希望将Agent技术集成到自己的项目中这篇文章都将提供清晰的路径和可直接运行的代码示例。学完本文你将能够独立搭建一个具备规划、执行、反思能力的智能体应用。AI Agent智能体简单来说是一个能够感知环境、自主决策并执行行动以实现特定目标的程序实体。它不仅仅是调用一次大模型API获取回答而是通过“思考-行动-观察”的循环像人类一样完成任务。我们可以通过一个对比来理解传统大模型调用用户问“今天北京天气如何”模型基于训练数据生成一段描述性文字。它无法获取实时数据。AI Agent用户提出同样问题。Agent内部会进行规划“要回答这个问题我需要调用天气查询工具。” 然后执行行动调用一个联网搜索或天气API工具获取实时数据。最后它观察工具返回的结果并组织成自然语言回复给用户“根据实时数据北京今天晴气温25°C。”其核心组件通常包括规划Planning分解任务制定步骤序列。工具使用Tool Use调用外部API、数据库、函数等扩展能力。记忆Memory保存对话历史、工具执行结果等上下文信息实现连贯交互。当前LangChain和LangGraph是构建AI Agent最主流的框架。LangChain提供了连接大模型、工具、记忆等组件的标准化接口而LangGraph则是在LangChain之上用于构建具有复杂、有状态工作流的图执行引擎。你可以把LangChain看作是乐高积木块而LangGraph就是指导你如何将这些积木组装成能动起来的机器人的说明书和控制器。2. 环境搭建与核心工具准备在开始编码之前我们需要准备好开发环境。本文将使用Python作为开发语言并聚焦于OpenAI的GPT系列模型也可替换为其他兼容API的模型。请确保你的Python版本在3.8以上。2.1 创建虚拟环境与安装依赖首先创建一个独立的项目目录并设置虚拟环境这是一个好的实践可以避免包依赖冲突。# 创建项目目录 mkdir ai-agent-tutorial cd ai-agent-tutorial # 创建并激活虚拟环境 (以venv为例也可使用conda) python -m venv venv # 在Windows上激活 venv\Scripts\activate # 在macOS/Linux上激活 source venv/bin/activate激活虚拟环境后命令行提示符前通常会显示(venv)。接下来安装核心依赖。# 安装LangChain及其社区工具包、LangGraph pip install langchain langchain-community langgraph # 安装OpenAI SDK (用于调用GPT模型) pip install openai # 安装Tavily SDK (我们将用它作为一个联网搜索工具的示例) # 注意Tavily需要API Key请前往其官网注册获取 pip install tavily-python # 可选但推荐安装环境变量管理库 pip install python-dotenv2.2 获取并配置API密钥AI Agent的运行依赖于大模型和外部工具服务因此需要配置相应的API密钥。强烈建议使用环境变量来管理这些敏感信息不要硬编码在代码中。OpenAI API Key访问 OpenAI平台 创建。Tavily API Key访问 Tavily官网 注册获取。在项目根目录下创建一个名为.env的文件将你的密钥填入# .env 文件 OPENAI_API_KEY你的-openai-api-key TAVILY_API_KEY你的-tavily-api-key然后在Python代码中通过dotenv加载这些变量。# config.py 或直接在代码开头 import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. 构建你的第一个基础AI Agent让我们从一个最简单的Agent开始一个能使用搜索工具回答问题的智能体。我们将使用LangChain的“AgentExecutor”模式。3.1 定义工具Tool工具是Agent延伸其能力的“手脚”。这里我们定义一个调用Tavily搜索API的工具。# tools.py from langchain_community.tools.tavily_search import TavilySearchResults def get_search_tool(): 创建并返回一个Tavily搜索工具实例。 该工具允许Agent在互联网上搜索最新信息。 # 确保已设置TAVILY_API_KEY环境变量 tool TavilySearchResults(max_results2) # 限制每次搜索返回2条结果 return tool3.2 初始化大语言模型LLMAgent的“大脑”是一个大语言模型。我们使用OpenAI的GPT-4o模型也可用gpt-3.5-turbo。# llm_setup.py from langchain_openai import ChatOpenAI def get_llm(model_namegpt-4o, temperature0): 初始化OpenAI聊天模型。 :param model_name: 模型名称如 gpt-4o, gpt-3.5-turbo :param temperature: 创造性0表示更确定1表示更多样。 :return: ChatOpenAI实例 llm ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyOPENAI_API_KEY) return llm3.3 创建并运行简单Agent现在我们将工具和大脑组装起来形成一个可以执行“思考-行动”循环的Agent。# simple_agent.py import sys sys.path.append(.) # 确保可以导入当前目录的模块 from llm_setup import get_llm from tools import get_search_tool from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # LangChain Hub用于拉取预定义的提示词 def run_simple_agent(question): 运行一个简单的ReAct范式Agent来回答问题。 # 1. 准备组件 llm get_llm() tools [get_search_tool()] # 2. 从LangChain Hub拉取一个为ReAct Agent设计好的提示词模板 # 这个提示词会指导LLM按照“Thought/Action/Action Input/Observation”的格式进行推理 prompt hub.pull(hwchase17/react) # 3. 使用工具和提示词创建Agent agent create_react_agent(llm, tools, prompt) # 4. 创建执行器它负责管理Agent的运行循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) print(f用户问题: {question}) print(*50) # 5. 执行 try: result agent_executor.invoke({input: question}) print(f\n最终答案: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) if __name__ __main__: # 测试一个需要最新信息的问题 question 2024年巴黎奥运会中国代表团获得了多少枚金牌 run_simple_agent(question)运行与观察 在终端执行python simple_agent.py。你会看到类似以下的详细输出清晰地展示了Agent的思考过程用户问题: 2024年巴黎奥运会中国代表团获得了多少枚金牌 Entering new AgentExecutor chain... Thought: 用户想知道2024年巴黎奥运会中国代表团的金牌数。这是一个需要最新信息的问题因为2024年奥运会尚未发生当前是2023年。我需要搜索确认一下。 Action: tavily_search_results_json Action Input: {query: 2024巴黎奥运会 中国 金牌数 最新} Observation: [{title: 2024年夏季奥林匹克运动会中国代表团 - 维基百科, url: https://zh.wikipedia.org/wiki/2024%E5%B9%B4%E5%A4%8F%E5%AD%A3%E5%A5%A5%E6%9E%97%E5%8C%B9%E5%85%8B%E8%BF%90%E5%8A%A8%E4%BC%9A%E4%B8%AD%E5%9B%BD%E4%BB%A3%E8%A1%A8%E5%9B%A2, content: 2024年夏季奥林匹克运动会中国代表团是中华人民共和国派出的...截至巴黎奥运会闭幕中国代表团共获得40枚金牌、27枚银牌、24枚铜牌位列金牌榜第一。}, ...] Thought: 根据搜索结果截至巴黎奥运会闭幕中国代表团共获得40枚金牌。 Action: Answer Action Input: 40枚金牌 Finished chain. 最终答案: 截至2024年巴黎奥运会闭幕中国代表团共获得了40枚金牌。这个简单的Agent已经具备了关键能力它识别出问题需要实时信息Thought决定使用搜索工具Action传入搜索词Action Input获取结果Observation最后提炼出答案并输出。4. 进阶使用LangGraph构建有状态工作流Agent基础的AgentExecutor适合简单任务但对于需要复杂状态管理、多角色协作或自定义工作流的场景LangGraph是更强大的工具。它允许你将Agent的工作流定义为一个“图”Graph其中节点是函数或工具调用边是控制流逻辑。4.1 LangGraph核心概念状态State与节点Node在LangGraph中一个工作流围绕一个共享的状态State字典运行。每个节点Node是一个函数它读取并修改这个状态。边Edge决定下一个执行哪个节点。让我们构建一个更复杂的Agent它具备长期记忆将对话历史保存到数据库和反思Reflection能力在任务失败时分析原因并重试。4.2 定义状态结构首先我们定义工作流中需要传递的所有信息。# graph_state.py from typing import TypedDict, List, Annotated import operator from langchain_core.messages import BaseMessage class AgentState(TypedDict): 定义LangGraph工作流的状态结构。 # 消息列表存储用户输入、AI回复、工具结果等所有消息 messages: Annotated[List[BaseMessage], operator.add] # 用户的最新问题 question: str # 记录工具调用的次数用于防止死循环 tool_call_count: int # 最终答案 final_answer: strAnnotated[List[BaseMessage], operator.add]是一个高级用法它告诉LangGraph在更新messages字段时使用append操作而不是覆盖这对于累积对话历史至关重要。4.3 创建具有记忆的图工作流我们将创建一个包含以下节点的工作流路由节点判断用户问题是简单聊天还是需要工具调用。工具调用节点调用搜索工具。反思节点检查工具返回的结果是否回答了问题如果没有则修改问题重新搜索。回答节点生成最终答案。# reflective_agent_graph.py import sys sys.path.append(.) from typing import Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_community.chat_message_histories import ChatMessageHistory from langchain_community.tools.tavily_search import TavilySearchResults from langchain_openai import ChatOpenAI from graph_state import AgentState from dotenv import load_dotenv import os load_dotenv() # 初始化组件 llm ChatOpenAI(modelgpt-4o, api_keyos.getenv(OPENAI_API_KEY)) search_tool TavilySearchResults(max_results2, api_keyos.getenv(TAVILY_API_KEY)) # 将工具绑定到LLM使其知道可以调用什么工具 llm_with_tools llm.bind_tools([search_tool]) def should_use_tool(state: AgentState) - Literal[call_tool, direct_answer]: 路由函数判断是否需要使用工具。 这是一个条件边Conditional Edge的判断逻辑。 messages state[messages] last_message messages[-1] # 简单规则如果问题包含“最新”、“今天”、“搜索”等词则使用工具 # 在实际应用中可以用一个更智能的LLM来路由 question state.get(question, ).lower() keywords [最新, 今天, 搜索, 查询, how many, what is the current] if any(keyword in question for keyword in keywords): print([路由] 判断为需要工具调用。) return call_tool else: print([路由] 判断为直接回答。) return direct_answer def call_tool_node(state: AgentState) - AgentState: 工具调用节点执行搜索并记录结果。 print([节点] 进入工具调用节点。) messages state[messages] question state[question] # 1. 让LLM根据对话历史决定搜索词 ai_msg llm_with_tools.invoke(messages) tool_calls ai_msg.tool_calls if not tool_calls: # 如果LLM没有生成工具调用直接返回 new_messages messages [ai_msg] return {messages: new_messages} # 2. 执行工具调用 tool_call tool_calls[0] tool_name tool_call[name] tool_args tool_call[args] print(f[工具调用] 调用 {tool_name}, 参数: {tool_args}) result search_tool.invoke(tool_args) # 3. 将工具执行结果作为 ToolMessage 添加到历史中 tool_message ToolMessage(contentstr(result), tool_call_idtool_call[id]) new_messages messages [ai_msg, tool_message] # 4. 更新工具调用计数 new_tool_call_count state.get(tool_call_count, 0) 1 return {messages: new_messages, tool_call_count: new_tool_call_count} def reflection_node(state: AgentState) - Literal[call_tool, finalize]: 反思节点检查工具返回的结果是否足够好。 如果不够好并且调用次数未超限则修改问题重新搜索。 print([节点] 进入反思节点。) messages state[messages] tool_call_count state.get(tool_call_count, 0) if tool_call_count 3: print([反思] 工具调用已达3次停止重试。) return finalize # 让LLM判断最后一次工具调用的结果是否充分回答了原始问题 reflection_prompt f 你是一个质量控制助手。请评估以下对话中工具返回的信息是否充分、准确地回答了用户的原始问题。 原始问题{state[question]} 工具返回的信息{messages[-1].content} 这是最后一次工具调用的结果 请只输出一个单词 - 如果信息充分且准确输出 SUFFICIENT。 - 如果信息不充分、不相关或不准确输出 INSUFFICIENT。 judgment llm.invoke(reflection_prompt).content.strip() if judgment SUFFICIENT: print([反思] 结果充分准备生成最终答案。) return finalize else: print([反思] 结果不充分将修改问题重新搜索。) # 可以在这里添加逻辑让LLM基于现有结果生成一个更精确的搜索问题 # 为了简化我们直接返回重新调用工具 return call_tool def direct_answer_node(state: AgentState) - AgentState: 直接回答节点处理无需工具调用的简单对话。 print([节点] 进入直接回答节点。) messages state[messages] response llm.invoke(messages) new_messages messages [response] return {messages: new_messages, final_answer: response.content} def finalize_node(state: AgentState) - AgentState: 最终回答节点基于所有消息生成最终答案。 print([节点] 进入最终回答节点。) messages state[messages] # 让LLM基于完整的对话历史包含工具结果生成面向用户的友好答案 final_response llm.invoke(messages) new_messages messages [final_response] return {messages: new_messages, final_answer: final_response.content} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(router, should_use_tool) # 注意路由函数本身作为节点但其返回值决定边 workflow.add_node(call_tool, call_tool_node) workflow.add_node(reflect, reflection_node) workflow.add_node(direct_answer, direct_answer_node) workflow.add_node(finalize, finalize_node) # 设置入口点 workflow.set_entry_point(router) # 添加边包括条件边 # 从 router 出发的条件边 workflow.add_conditional_edges( router, should_use_tool, # 这个函数返回下一个节点的名称 { call_tool: call_tool, direct_answer: direct_answer } ) # 从 direct_answer 直接到结束 workflow.add_edge(direct_answer, END) # 从 call_tool 到 reflect workflow.add_edge(call_tool, reflect) # 从 reflect 出发的条件边 workflow.add_conditional_edges( reflect, reflection_node, { call_tool: call_tool, finalize: finalize } ) # 从 finalize 到结束 workflow.add_edge(finalize, END) # 编译图 app workflow.compile() # 运行图 def run_reflective_agent(question: str): 运行具备反思能力的图工作流Agent。 print(f\n{*60}) print(f开始处理问题: {question}) print(*60) # 初始化状态 initial_state: AgentState { messages: [HumanMessage(contentquestion)], question: question, tool_call_count: 0, final_answer: } # 执行图 final_state app.invoke(initial_state) print(f\n{*60}) print(工作流执行完毕。) print(f最终答案: {final_state.get(final_answer, 未生成答案)}) print(*60) # 打印完整的消息流可选用于调试 # for msg in final_state[messages]: # print(f{type(msg).__name__}: {msg.content[:200]}...) if __name__ __main__: # 测试一个可能需要多次搜索或反思的问题 test_question 对比一下特斯拉Model 3和比亚迪汉EV的最新款在续航和智能驾驶方面的差异。 run_reflective_agent(test_question)这个示例展示了LangGraph的强大之处你可以清晰地定义工作流的每个步骤和决策点。reflection_node实现了简单的自我纠正机制如果第一次搜索效果不好Agent会尝试重新规划。在实际项目中你可以将这个图扩展得更复杂例如加入验证节点、多专家协作节点等。5. 核心概念深入Agent、RAG与LangGraph的关系在学习和开发过程中你一定会遇到RAG检索增强生成和Agent这两个紧密相关的概念。理解它们的区别与联系至关重要。RAGRetrieval-Augmented Generation 一种架构模式用于解决大模型的“知识截止”和“幻觉”问题。其核心流程是用户提问 → 从知识库如向量数据库检索相关文档片段 → 将片段和问题一起交给大模型生成答案。RAG更像是一个增强的“问答系统”。AI Agent 一个更宏观的架构概念指能自主完成任务的智能体。一个Agent可以使用RAG作为其内部的一个工具。例如一个研究助手Agent其任务可能是“撰写一篇关于量子计算的报告”。它会规划步骤1) 搜索最新论文调用搜索工具2) 阅读公司内部文档调用RAG工具查询向量数据库3) 整理大纲4) 撰写内容5) 检查格式。在这里RAG是Agent工具箱里的一把“专用扳手”。LangChain vs. LangGraphLangChain 提供了构建AI应用包括RAG系统和简单Agent所需的标准化组件和连接器。它抽象了与LLM、向量数据库、工具等的交互让你用统一的API来操作。它的AgentExecutor已经能处理简单的多步任务。LangGraph 是构建复杂、有状态、多参与者工作流的框架。当你的Agent需要循环、条件分支、持久化状态、多角色协作如一个分析师Agent和一个审核员Agent时LangGraph比基础的AgentExecutor更合适。它让你以“图”的视角来设计和调试工作流。简单说用LangChain快速搭建应用用LangGraph设计复杂流程。很多复杂的RAG系统如包含查询重写、混合检索、重排序等步骤本身也可以用LangGraph来构建。6. 实战构建一个本地知识库问答AgentRAG Agent让我们结合RAG和Agent构建一个能回答特定领域问题的智能体。假设我们有一个公司内部的技术文档PDF格式我们要创建一个Agent它能理解用户问题并从这些文档中查找信息来回答。6.1 项目结构local_rag_agent/ ├── data/ # 存放原始文档 │ └── company_handbook.pdf ├── vector_store/ # 存放向量数据库由程序生成 ├── tools/ # 自定义工具 │ └── rag_tool.py ├── agents/ # Agent定义 │ └── doc_qa_agent.py ├── config.py # 配置 ├── ingest.py # 文档加载与向量化脚本 └── main.py # 主程序入口6.2 文档加载与向量化知识库构建首先我们需要将PDF文档处理成向量并存储起来。# ingest.py from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 使用Chroma作为向量数据库 import os from config import OPENAI_API_KEY def ingest_documents(pdf_path: str, persist_directory: str ./vector_store): 加载PDF文档分割文本生成向量并存储到Chroma数据库。 print(开始加载文档...) # 1. 加载文档 loader PyPDFLoader(pdf_path) documents loader.load() # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符保持上下文 separators[\n\n, \n, 。, , , , , 、, ] ) splits text_splitter.split_documents(documents) print(f文档分割为 {len(splits)} 个片段。) # 3. 生成向量并存储 embeddings OpenAIEmbeddings(api_keyOPENAI_API_KEY) # 如果目录已存在可以加载现有库否则创建新库 if os.path.exists(persist_directory): print(f从 {persist_directory} 加载已有向量库...) vectorstore Chroma(persist_directorypersist_directory, embedding_functionembeddings) # 添加新文档可选这里我们假设重新创建 # vectorstore.add_documents(splits) else: print(f创建新的向量库并存储到 {persist_directory} ...) vectorstore Chroma.from_documents( documentssplits, embeddingembeddings, persist_directorypersist_directory ) vectorstore.persist() # 持久化到磁盘 print(文档向量化完成) return vectorstore if __name__ __main__: # 处理你的PDF文档 pdf_file ./data/company_handbook.pdf if os.path.exists(pdf_file): ingest_documents(pdf_file) else: print(f文件 {pdf_file} 不存在请将PDF文档放入data目录。)6.3 创建RAG检索工具接下来我们创建一个工具让Agent在需要时能够查询这个本地知识库。# tools/rag_tool.py from langchain.tools import tool from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from config import OPENAI_API_KEY, VECTOR_STORE_PATH # 初始化向量库全局避免重复加载 _embeddings OpenAIEmbeddings(api_keyOPENAI_API_KEY) _vectorstore Chroma(persist_directoryVECTOR_STORE_PATH, embedding_function_embeddings) # 创建检索器 _retriever _vectorstore.as_retriever(search_kwargs{k: 3}) # 返回最相关的3个片段 tool def query_company_handbook(query: str) - str: 从公司内部知识库员工手册中检索与问题相关的信息。 当用户询问关于公司制度、流程、政策、技术规范等内部信息时使用此工具。 Args: query: 用户的查询问题必须是明确的自然语言。 Returns: 从知识库中检索到的相关文本内容。如果未找到返回“在知识库中未找到相关信息”。 print(f[RAG工具] 正在知识库中检索: {query}) docs _retriever.invoke(query) if not docs: return 在知识库中未找到相关信息。 # 将检索到的文档内容合并 context \n\n---\n\n.join([doc.page_content for doc in docs]) return f从公司知识库中检索到以下相关信息\n\n{context}6.4 构建多功能问答Agent现在我们创建一个Agent它既能查询互联网又能查询内部知识库。# agents/doc_qa_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain import hub from langchain_openai import ChatOpenAI from tools.rag_tool import query_company_handbook from tools import get_search_tool # 之前定义的搜索工具 from config import OPENAI_API_KEY def get_doc_qa_agent(): 创建一个结合了互联网搜索和内部知识库查询的Agent。 llm ChatOpenAI(modelgpt-4o, temperature0, api_keyOPENAI_API_KEY) # 定义工具列表 tools [ get_search_tool(), # 工具1互联网搜索 query_company_handbook, # 工具2内部知识库查询 ] # 使用ReAct提示词模板 prompt hub.pull(hwchase17/react) # 创建Agent agent create_react_agent(llm, tools, prompt) # 创建执行器设置详细日志和错误处理 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate # 当连续两个动作为“Final Answer”时停止 ) return agent_executor def ask_agent(agent_executor, question): 向Agent提问并打印结果。 print(f\n用户: {question}) print(- * 40) try: result agent_executor.invoke({input: question}) print(f\nAgent: {result[output]}) except Exception as e: print(f执行出错: {e}) if __name__ __main__: agent get_doc_qa_agent() # 测试不同类型的问题 questions [ 我们公司的年假制度是怎样的, # 应触发内部知识库工具 今天纽约的天气怎么样, # 应触发互联网搜索工具 根据员工手册报销流程需要哪些材料同时帮我查一下最近AI芯片有什么新闻。 # 可能触发两个工具 ] for q in questions: ask_agent(agent, q) print(\n *60 \n)运行这个程序你会看到Agent如何根据问题类型智能地选择调用不同的工具。对于公司制度问题它会使用query_company_handbook工具对于实时天气问题它会使用互联网搜索工具。这正是一个初级AI Agent的典型应用。7. 常见问题与排查指南FAQ在开发AI Agent过程中你一定会遇到各种问题。以下是一些常见问题及其解决方案。问题现象可能原因排查思路与解决方案ModuleNotFoundError: No module named langchain_community依赖未正确安装或虚拟环境未激活。1. 确认已激活虚拟环境 (venv\Scripts\activate或source venv/bin/activate)。2. 运行pip install langchain-community。openai.AuthenticationError: Incorrect API key providedOpenAI API Key 错误或未设置。1. 检查.env文件中的OPENAI_API_KEY是否正确。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认已加载。3. 确保Key有余额和相应权限。Agent陷入死循环不断调用工具Agent无法从工具结果中提炼出最终答案或提示词未明确要求其最终输出。1. 在AgentExecutor中设置max_iterations参数如5。2. 检查工具的返回格式是否清晰便于LLM理解。3. 优化提示词Prompt明确要求“在得到足够信息后必须给出最终答案”。工具调用失败返回Invalid tool call工具定义与LLM绑定的工具描述不匹配或工具参数格式错误。1. 使用llm.bind_tools(tools)确保LLM知道工具的准确名称和参数。2. 检查工具函数的docstringLangChain会用它生成工具描述。3. 在AgentExecutor中设置handle_parsing_errorsTrue以捕获解析错误。RAG检索结果不相关文本分割策略不佳或检索器配置不当。1. 调整RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。2. 尝试不同的嵌入模型Embedding Model。3. 在检索时调整search_kwargs如{“k”: 5}返回更多结果或使用MMR搜索类型来平衡相关性与多样性。LangGraph图编译或运行出错状态State结构定义错误或节点函数返回值不符合预期。1. 仔细检查TypedDict的定义确保与节点函数返回的字典键匹配。2. 使用app.get_graph().draw_mermaid()输出图结构可视化检查节点和边是否正确连接。3. 在每个节点函数内打印日志跟踪状态的变化。程序运行慢频繁调用LLM或嵌入模型网络延迟高。1. 对于RAG考虑将向量数据库本地化如Chroma、FAISS避免每次查询都调用云端嵌入API。2. 使用缓存机制例如langchain.cache缓存LLM响应。3. 对于简单路由判断可尝试用规则关键词代替LLM调用。8. 最佳实践与项目进阶建议掌握了基础之后要打造一个健壮、可用的AI Agent系统还需要关注以下工程化实践8.1 提示词Prompt工程清晰的角色与指令在系统提示词System Prompt中明确Agent的角色、能力和约束。例如“你是一个专业的研究助手必须使用工具获取最新信息并在回答时引用来源。”少样本Few-Shot学习在提示词中提供1-2个高质量的输入输出示例能显著提升Agent执行复杂任务的准确性。结构化输出要求LLM以JSON等特定格式输出便于后续程序解析。这在多智能体协作中尤其重要。8.2 工具设计单一职责每个工具应只做一件事并做好。避免创建功能臃肿的“万能工具”。健壮的错误处理工具函数内部应有完善的try-except并返回结构化的错误信息让Agent能理解并采取补救措施。详细的描述工具的docstring至关重要LLM依靠它来决定何时以及如何调用工具。描述应清晰说明工具的用途、输入参数格式和输出示例。8.3 记忆Memory管理短期记忆使用ConversationBufferMemory或ConversationSummaryMemory来维护对话上下文。注意上下文长度限制对于长对话摘要记忆SummaryMemory是更好的选择。长期记忆对于需要记住跨会话信息的Agent可以将关键信息向量化后存入数据库在需要时通过RAG方式检索。这就是构建“数字分身”或“个性化助手”的基础。8.4 评估与监控构建测试集针对你的Agent常见任务准备一批标准问题及答案定期运行测试评估其准确性和稳定性。记录与审计记录每次Agent运行的完整链条Thought, Action, Observation这对于调试和优化至关重要。LangSmith是LangChain官方提供的优秀监控平台。人工反馈循环HITL在关键决策点引入人工审核。LangGraph原生支持“Human-in-the-Loop”节点可以在工作流中暂停并等待人工输入。8.5 学习路线与下一步夯实基础彻底理解本文中的代码尝试修改工具、调整提示词、构建不同的LangGraph工作流。探索高级模式学习ReAct,Plan-and-Execute,AutoGen微软的多智能体框架等高级Agent架构。深入LangGraph研究其Checkpointer实现持久化状态Supervisor实现多智能体调度以及Pregel并发执行模型。集成实际项目将Agent能力嵌入到你的Web应用用Flask/FastAPI、聊天机器人或自动化流程中。关注开源生态参与langchain-ai相关项目关注CrewAI,AutoGen等新兴框架保持对技术趋势的敏感。AI Agent的开发是一场结合了软件工程、提示词艺术和LLM能力的探索。从今天这个能调用搜索和知识库的简单助手开始逐步为其添加规划、记忆、协作和反思的能力你就能构建出真正智能、有用的应用程序。
1. 从零到一AI Agent 到底是什么最近在技术社区和招聘网站上“AI Agent”这个词的热度持续攀升很多开发者朋友都跃跃欲试想抓住这波技术浪潮。但面对海量的概念、框架和教程往往感觉无从下手什么是Agent它和普通的大模型调用有什么区别LangChain和LangGraph又是什么关系本文旨在为你拨开迷雾提供一份系统、可落地的AI Agent实战指南。我们将从一个最简单的“单步工具调用”Agent开始逐步深入到使用LangGraph构建具备复杂工作流和记忆能力的多智能体系统。无论你是想快速入门了解核心概念还是希望将Agent技术集成到自己的项目中这篇文章都将提供清晰的路径和可直接运行的代码示例。学完本文你将能够独立搭建一个具备规划、执行、反思能力的智能体应用。AI Agent智能体简单来说是一个能够感知环境、自主决策并执行行动以实现特定目标的程序实体。它不仅仅是调用一次大模型API获取回答而是通过“思考-行动-观察”的循环像人类一样完成任务。我们可以通过一个对比来理解传统大模型调用用户问“今天北京天气如何”模型基于训练数据生成一段描述性文字。它无法获取实时数据。AI Agent用户提出同样问题。Agent内部会进行规划“要回答这个问题我需要调用天气查询工具。” 然后执行行动调用一个联网搜索或天气API工具获取实时数据。最后它观察工具返回的结果并组织成自然语言回复给用户“根据实时数据北京今天晴气温25°C。”其核心组件通常包括规划Planning分解任务制定步骤序列。工具使用Tool Use调用外部API、数据库、函数等扩展能力。记忆Memory保存对话历史、工具执行结果等上下文信息实现连贯交互。当前LangChain和LangGraph是构建AI Agent最主流的框架。LangChain提供了连接大模型、工具、记忆等组件的标准化接口而LangGraph则是在LangChain之上用于构建具有复杂、有状态工作流的图执行引擎。你可以把LangChain看作是乐高积木块而LangGraph就是指导你如何将这些积木组装成能动起来的机器人的说明书和控制器。2. 环境搭建与核心工具准备在开始编码之前我们需要准备好开发环境。本文将使用Python作为开发语言并聚焦于OpenAI的GPT系列模型也可替换为其他兼容API的模型。请确保你的Python版本在3.8以上。2.1 创建虚拟环境与安装依赖首先创建一个独立的项目目录并设置虚拟环境这是一个好的实践可以避免包依赖冲突。# 创建项目目录 mkdir ai-agent-tutorial cd ai-agent-tutorial # 创建并激活虚拟环境 (以venv为例也可使用conda) python -m venv venv # 在Windows上激活 venv\Scripts\activate # 在macOS/Linux上激活 source venv/bin/activate激活虚拟环境后命令行提示符前通常会显示(venv)。接下来安装核心依赖。# 安装LangChain及其社区工具包、LangGraph pip install langchain langchain-community langgraph # 安装OpenAI SDK (用于调用GPT模型) pip install openai # 安装Tavily SDK (我们将用它作为一个联网搜索工具的示例) # 注意Tavily需要API Key请前往其官网注册获取 pip install tavily-python # 可选但推荐安装环境变量管理库 pip install python-dotenv2.2 获取并配置API密钥AI Agent的运行依赖于大模型和外部工具服务因此需要配置相应的API密钥。强烈建议使用环境变量来管理这些敏感信息不要硬编码在代码中。OpenAI API Key访问 OpenAI平台 创建。Tavily API Key访问 Tavily官网 注册获取。在项目根目录下创建一个名为.env的文件将你的密钥填入# .env 文件 OPENAI_API_KEY你的-openai-api-key TAVILY_API_KEY你的-tavily-api-key然后在Python代码中通过dotenv加载这些变量。# config.py 或直接在代码开头 import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. 构建你的第一个基础AI Agent让我们从一个最简单的Agent开始一个能使用搜索工具回答问题的智能体。我们将使用LangChain的“AgentExecutor”模式。3.1 定义工具Tool工具是Agent延伸其能力的“手脚”。这里我们定义一个调用Tavily搜索API的工具。# tools.py from langchain_community.tools.tavily_search import TavilySearchResults def get_search_tool(): 创建并返回一个Tavily搜索工具实例。 该工具允许Agent在互联网上搜索最新信息。 # 确保已设置TAVILY_API_KEY环境变量 tool TavilySearchResults(max_results2) # 限制每次搜索返回2条结果 return tool3.2 初始化大语言模型LLMAgent的“大脑”是一个大语言模型。我们使用OpenAI的GPT-4o模型也可用gpt-3.5-turbo。# llm_setup.py from langchain_openai import ChatOpenAI def get_llm(model_namegpt-4o, temperature0): 初始化OpenAI聊天模型。 :param model_name: 模型名称如 gpt-4o, gpt-3.5-turbo :param temperature: 创造性0表示更确定1表示更多样。 :return: ChatOpenAI实例 llm ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyOPENAI_API_KEY) return llm3.3 创建并运行简单Agent现在我们将工具和大脑组装起来形成一个可以执行“思考-行动”循环的Agent。# simple_agent.py import sys sys.path.append(.) # 确保可以导入当前目录的模块 from llm_setup import get_llm from tools import get_search_tool from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # LangChain Hub用于拉取预定义的提示词 def run_simple_agent(question): 运行一个简单的ReAct范式Agent来回答问题。 # 1. 准备组件 llm get_llm() tools [get_search_tool()] # 2. 从LangChain Hub拉取一个为ReAct Agent设计好的提示词模板 # 这个提示词会指导LLM按照“Thought/Action/Action Input/Observation”的格式进行推理 prompt hub.pull(hwchase17/react) # 3. 使用工具和提示词创建Agent agent create_react_agent(llm, tools, prompt) # 4. 创建执行器它负责管理Agent的运行循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) print(f用户问题: {question}) print(*50) # 5. 执行 try: result agent_executor.invoke({input: question}) print(f\n最终答案: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) if __name__ __main__: # 测试一个需要最新信息的问题 question 2024年巴黎奥运会中国代表团获得了多少枚金牌 run_simple_agent(question)运行与观察 在终端执行python simple_agent.py。你会看到类似以下的详细输出清晰地展示了Agent的思考过程用户问题: 2024年巴黎奥运会中国代表团获得了多少枚金牌 Entering new AgentExecutor chain... Thought: 用户想知道2024年巴黎奥运会中国代表团的金牌数。这是一个需要最新信息的问题因为2024年奥运会尚未发生当前是2023年。我需要搜索确认一下。 Action: tavily_search_results_json Action Input: {query: 2024巴黎奥运会 中国 金牌数 最新} Observation: [{title: 2024年夏季奥林匹克运动会中国代表团 - 维基百科, url: https://zh.wikipedia.org/wiki/2024%E5%B9%B4%E5%A4%8F%E5%AD%A3%E5%A5%A5%E6%9E%97%E5%8C%B9%E5%85%8B%E8%BF%90%E5%8A%A8%E4%BC%9A%E4%B8%AD%E5%9B%BD%E4%BB%A3%E8%A1%A8%E5%9B%A2, content: 2024年夏季奥林匹克运动会中国代表团是中华人民共和国派出的...截至巴黎奥运会闭幕中国代表团共获得40枚金牌、27枚银牌、24枚铜牌位列金牌榜第一。}, ...] Thought: 根据搜索结果截至巴黎奥运会闭幕中国代表团共获得40枚金牌。 Action: Answer Action Input: 40枚金牌 Finished chain. 最终答案: 截至2024年巴黎奥运会闭幕中国代表团共获得了40枚金牌。这个简单的Agent已经具备了关键能力它识别出问题需要实时信息Thought决定使用搜索工具Action传入搜索词Action Input获取结果Observation最后提炼出答案并输出。4. 进阶使用LangGraph构建有状态工作流Agent基础的AgentExecutor适合简单任务但对于需要复杂状态管理、多角色协作或自定义工作流的场景LangGraph是更强大的工具。它允许你将Agent的工作流定义为一个“图”Graph其中节点是函数或工具调用边是控制流逻辑。4.1 LangGraph核心概念状态State与节点Node在LangGraph中一个工作流围绕一个共享的状态State字典运行。每个节点Node是一个函数它读取并修改这个状态。边Edge决定下一个执行哪个节点。让我们构建一个更复杂的Agent它具备长期记忆将对话历史保存到数据库和反思Reflection能力在任务失败时分析原因并重试。4.2 定义状态结构首先我们定义工作流中需要传递的所有信息。# graph_state.py from typing import TypedDict, List, Annotated import operator from langchain_core.messages import BaseMessage class AgentState(TypedDict): 定义LangGraph工作流的状态结构。 # 消息列表存储用户输入、AI回复、工具结果等所有消息 messages: Annotated[List[BaseMessage], operator.add] # 用户的最新问题 question: str # 记录工具调用的次数用于防止死循环 tool_call_count: int # 最终答案 final_answer: strAnnotated[List[BaseMessage], operator.add]是一个高级用法它告诉LangGraph在更新messages字段时使用append操作而不是覆盖这对于累积对话历史至关重要。4.3 创建具有记忆的图工作流我们将创建一个包含以下节点的工作流路由节点判断用户问题是简单聊天还是需要工具调用。工具调用节点调用搜索工具。反思节点检查工具返回的结果是否回答了问题如果没有则修改问题重新搜索。回答节点生成最终答案。# reflective_agent_graph.py import sys sys.path.append(.) from typing import Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_community.chat_message_histories import ChatMessageHistory from langchain_community.tools.tavily_search import TavilySearchResults from langchain_openai import ChatOpenAI from graph_state import AgentState from dotenv import load_dotenv import os load_dotenv() # 初始化组件 llm ChatOpenAI(modelgpt-4o, api_keyos.getenv(OPENAI_API_KEY)) search_tool TavilySearchResults(max_results2, api_keyos.getenv(TAVILY_API_KEY)) # 将工具绑定到LLM使其知道可以调用什么工具 llm_with_tools llm.bind_tools([search_tool]) def should_use_tool(state: AgentState) - Literal[call_tool, direct_answer]: 路由函数判断是否需要使用工具。 这是一个条件边Conditional Edge的判断逻辑。 messages state[messages] last_message messages[-1] # 简单规则如果问题包含“最新”、“今天”、“搜索”等词则使用工具 # 在实际应用中可以用一个更智能的LLM来路由 question state.get(question, ).lower() keywords [最新, 今天, 搜索, 查询, how many, what is the current] if any(keyword in question for keyword in keywords): print([路由] 判断为需要工具调用。) return call_tool else: print([路由] 判断为直接回答。) return direct_answer def call_tool_node(state: AgentState) - AgentState: 工具调用节点执行搜索并记录结果。 print([节点] 进入工具调用节点。) messages state[messages] question state[question] # 1. 让LLM根据对话历史决定搜索词 ai_msg llm_with_tools.invoke(messages) tool_calls ai_msg.tool_calls if not tool_calls: # 如果LLM没有生成工具调用直接返回 new_messages messages [ai_msg] return {messages: new_messages} # 2. 执行工具调用 tool_call tool_calls[0] tool_name tool_call[name] tool_args tool_call[args] print(f[工具调用] 调用 {tool_name}, 参数: {tool_args}) result search_tool.invoke(tool_args) # 3. 将工具执行结果作为 ToolMessage 添加到历史中 tool_message ToolMessage(contentstr(result), tool_call_idtool_call[id]) new_messages messages [ai_msg, tool_message] # 4. 更新工具调用计数 new_tool_call_count state.get(tool_call_count, 0) 1 return {messages: new_messages, tool_call_count: new_tool_call_count} def reflection_node(state: AgentState) - Literal[call_tool, finalize]: 反思节点检查工具返回的结果是否足够好。 如果不够好并且调用次数未超限则修改问题重新搜索。 print([节点] 进入反思节点。) messages state[messages] tool_call_count state.get(tool_call_count, 0) if tool_call_count 3: print([反思] 工具调用已达3次停止重试。) return finalize # 让LLM判断最后一次工具调用的结果是否充分回答了原始问题 reflection_prompt f 你是一个质量控制助手。请评估以下对话中工具返回的信息是否充分、准确地回答了用户的原始问题。 原始问题{state[question]} 工具返回的信息{messages[-1].content} 这是最后一次工具调用的结果 请只输出一个单词 - 如果信息充分且准确输出 SUFFICIENT。 - 如果信息不充分、不相关或不准确输出 INSUFFICIENT。 judgment llm.invoke(reflection_prompt).content.strip() if judgment SUFFICIENT: print([反思] 结果充分准备生成最终答案。) return finalize else: print([反思] 结果不充分将修改问题重新搜索。) # 可以在这里添加逻辑让LLM基于现有结果生成一个更精确的搜索问题 # 为了简化我们直接返回重新调用工具 return call_tool def direct_answer_node(state: AgentState) - AgentState: 直接回答节点处理无需工具调用的简单对话。 print([节点] 进入直接回答节点。) messages state[messages] response llm.invoke(messages) new_messages messages [response] return {messages: new_messages, final_answer: response.content} def finalize_node(state: AgentState) - AgentState: 最终回答节点基于所有消息生成最终答案。 print([节点] 进入最终回答节点。) messages state[messages] # 让LLM基于完整的对话历史包含工具结果生成面向用户的友好答案 final_response llm.invoke(messages) new_messages messages [final_response] return {messages: new_messages, final_answer: final_response.content} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(router, should_use_tool) # 注意路由函数本身作为节点但其返回值决定边 workflow.add_node(call_tool, call_tool_node) workflow.add_node(reflect, reflection_node) workflow.add_node(direct_answer, direct_answer_node) workflow.add_node(finalize, finalize_node) # 设置入口点 workflow.set_entry_point(router) # 添加边包括条件边 # 从 router 出发的条件边 workflow.add_conditional_edges( router, should_use_tool, # 这个函数返回下一个节点的名称 { call_tool: call_tool, direct_answer: direct_answer } ) # 从 direct_answer 直接到结束 workflow.add_edge(direct_answer, END) # 从 call_tool 到 reflect workflow.add_edge(call_tool, reflect) # 从 reflect 出发的条件边 workflow.add_conditional_edges( reflect, reflection_node, { call_tool: call_tool, finalize: finalize } ) # 从 finalize 到结束 workflow.add_edge(finalize, END) # 编译图 app workflow.compile() # 运行图 def run_reflective_agent(question: str): 运行具备反思能力的图工作流Agent。 print(f\n{*60}) print(f开始处理问题: {question}) print(*60) # 初始化状态 initial_state: AgentState { messages: [HumanMessage(contentquestion)], question: question, tool_call_count: 0, final_answer: } # 执行图 final_state app.invoke(initial_state) print(f\n{*60}) print(工作流执行完毕。) print(f最终答案: {final_state.get(final_answer, 未生成答案)}) print(*60) # 打印完整的消息流可选用于调试 # for msg in final_state[messages]: # print(f{type(msg).__name__}: {msg.content[:200]}...) if __name__ __main__: # 测试一个可能需要多次搜索或反思的问题 test_question 对比一下特斯拉Model 3和比亚迪汉EV的最新款在续航和智能驾驶方面的差异。 run_reflective_agent(test_question)这个示例展示了LangGraph的强大之处你可以清晰地定义工作流的每个步骤和决策点。reflection_node实现了简单的自我纠正机制如果第一次搜索效果不好Agent会尝试重新规划。在实际项目中你可以将这个图扩展得更复杂例如加入验证节点、多专家协作节点等。5. 核心概念深入Agent、RAG与LangGraph的关系在学习和开发过程中你一定会遇到RAG检索增强生成和Agent这两个紧密相关的概念。理解它们的区别与联系至关重要。RAGRetrieval-Augmented Generation 一种架构模式用于解决大模型的“知识截止”和“幻觉”问题。其核心流程是用户提问 → 从知识库如向量数据库检索相关文档片段 → 将片段和问题一起交给大模型生成答案。RAG更像是一个增强的“问答系统”。AI Agent 一个更宏观的架构概念指能自主完成任务的智能体。一个Agent可以使用RAG作为其内部的一个工具。例如一个研究助手Agent其任务可能是“撰写一篇关于量子计算的报告”。它会规划步骤1) 搜索最新论文调用搜索工具2) 阅读公司内部文档调用RAG工具查询向量数据库3) 整理大纲4) 撰写内容5) 检查格式。在这里RAG是Agent工具箱里的一把“专用扳手”。LangChain vs. LangGraphLangChain 提供了构建AI应用包括RAG系统和简单Agent所需的标准化组件和连接器。它抽象了与LLM、向量数据库、工具等的交互让你用统一的API来操作。它的AgentExecutor已经能处理简单的多步任务。LangGraph 是构建复杂、有状态、多参与者工作流的框架。当你的Agent需要循环、条件分支、持久化状态、多角色协作如一个分析师Agent和一个审核员Agent时LangGraph比基础的AgentExecutor更合适。它让你以“图”的视角来设计和调试工作流。简单说用LangChain快速搭建应用用LangGraph设计复杂流程。很多复杂的RAG系统如包含查询重写、混合检索、重排序等步骤本身也可以用LangGraph来构建。6. 实战构建一个本地知识库问答AgentRAG Agent让我们结合RAG和Agent构建一个能回答特定领域问题的智能体。假设我们有一个公司内部的技术文档PDF格式我们要创建一个Agent它能理解用户问题并从这些文档中查找信息来回答。6.1 项目结构local_rag_agent/ ├── data/ # 存放原始文档 │ └── company_handbook.pdf ├── vector_store/ # 存放向量数据库由程序生成 ├── tools/ # 自定义工具 │ └── rag_tool.py ├── agents/ # Agent定义 │ └── doc_qa_agent.py ├── config.py # 配置 ├── ingest.py # 文档加载与向量化脚本 └── main.py # 主程序入口6.2 文档加载与向量化知识库构建首先我们需要将PDF文档处理成向量并存储起来。# ingest.py from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 使用Chroma作为向量数据库 import os from config import OPENAI_API_KEY def ingest_documents(pdf_path: str, persist_directory: str ./vector_store): 加载PDF文档分割文本生成向量并存储到Chroma数据库。 print(开始加载文档...) # 1. 加载文档 loader PyPDFLoader(pdf_path) documents loader.load() # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符保持上下文 separators[\n\n, \n, 。, , , , , 、, ] ) splits text_splitter.split_documents(documents) print(f文档分割为 {len(splits)} 个片段。) # 3. 生成向量并存储 embeddings OpenAIEmbeddings(api_keyOPENAI_API_KEY) # 如果目录已存在可以加载现有库否则创建新库 if os.path.exists(persist_directory): print(f从 {persist_directory} 加载已有向量库...) vectorstore Chroma(persist_directorypersist_directory, embedding_functionembeddings) # 添加新文档可选这里我们假设重新创建 # vectorstore.add_documents(splits) else: print(f创建新的向量库并存储到 {persist_directory} ...) vectorstore Chroma.from_documents( documentssplits, embeddingembeddings, persist_directorypersist_directory ) vectorstore.persist() # 持久化到磁盘 print(文档向量化完成) return vectorstore if __name__ __main__: # 处理你的PDF文档 pdf_file ./data/company_handbook.pdf if os.path.exists(pdf_file): ingest_documents(pdf_file) else: print(f文件 {pdf_file} 不存在请将PDF文档放入data目录。)6.3 创建RAG检索工具接下来我们创建一个工具让Agent在需要时能够查询这个本地知识库。# tools/rag_tool.py from langchain.tools import tool from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from config import OPENAI_API_KEY, VECTOR_STORE_PATH # 初始化向量库全局避免重复加载 _embeddings OpenAIEmbeddings(api_keyOPENAI_API_KEY) _vectorstore Chroma(persist_directoryVECTOR_STORE_PATH, embedding_function_embeddings) # 创建检索器 _retriever _vectorstore.as_retriever(search_kwargs{k: 3}) # 返回最相关的3个片段 tool def query_company_handbook(query: str) - str: 从公司内部知识库员工手册中检索与问题相关的信息。 当用户询问关于公司制度、流程、政策、技术规范等内部信息时使用此工具。 Args: query: 用户的查询问题必须是明确的自然语言。 Returns: 从知识库中检索到的相关文本内容。如果未找到返回“在知识库中未找到相关信息”。 print(f[RAG工具] 正在知识库中检索: {query}) docs _retriever.invoke(query) if not docs: return 在知识库中未找到相关信息。 # 将检索到的文档内容合并 context \n\n---\n\n.join([doc.page_content for doc in docs]) return f从公司知识库中检索到以下相关信息\n\n{context}6.4 构建多功能问答Agent现在我们创建一个Agent它既能查询互联网又能查询内部知识库。# agents/doc_qa_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain import hub from langchain_openai import ChatOpenAI from tools.rag_tool import query_company_handbook from tools import get_search_tool # 之前定义的搜索工具 from config import OPENAI_API_KEY def get_doc_qa_agent(): 创建一个结合了互联网搜索和内部知识库查询的Agent。 llm ChatOpenAI(modelgpt-4o, temperature0, api_keyOPENAI_API_KEY) # 定义工具列表 tools [ get_search_tool(), # 工具1互联网搜索 query_company_handbook, # 工具2内部知识库查询 ] # 使用ReAct提示词模板 prompt hub.pull(hwchase17/react) # 创建Agent agent create_react_agent(llm, tools, prompt) # 创建执行器设置详细日志和错误处理 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate # 当连续两个动作为“Final Answer”时停止 ) return agent_executor def ask_agent(agent_executor, question): 向Agent提问并打印结果。 print(f\n用户: {question}) print(- * 40) try: result agent_executor.invoke({input: question}) print(f\nAgent: {result[output]}) except Exception as e: print(f执行出错: {e}) if __name__ __main__: agent get_doc_qa_agent() # 测试不同类型的问题 questions [ 我们公司的年假制度是怎样的, # 应触发内部知识库工具 今天纽约的天气怎么样, # 应触发互联网搜索工具 根据员工手册报销流程需要哪些材料同时帮我查一下最近AI芯片有什么新闻。 # 可能触发两个工具 ] for q in questions: ask_agent(agent, q) print(\n *60 \n)运行这个程序你会看到Agent如何根据问题类型智能地选择调用不同的工具。对于公司制度问题它会使用query_company_handbook工具对于实时天气问题它会使用互联网搜索工具。这正是一个初级AI Agent的典型应用。7. 常见问题与排查指南FAQ在开发AI Agent过程中你一定会遇到各种问题。以下是一些常见问题及其解决方案。问题现象可能原因排查思路与解决方案ModuleNotFoundError: No module named langchain_community依赖未正确安装或虚拟环境未激活。1. 确认已激活虚拟环境 (venv\Scripts\activate或source venv/bin/activate)。2. 运行pip install langchain-community。openai.AuthenticationError: Incorrect API key providedOpenAI API Key 错误或未设置。1. 检查.env文件中的OPENAI_API_KEY是否正确。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认已加载。3. 确保Key有余额和相应权限。Agent陷入死循环不断调用工具Agent无法从工具结果中提炼出最终答案或提示词未明确要求其最终输出。1. 在AgentExecutor中设置max_iterations参数如5。2. 检查工具的返回格式是否清晰便于LLM理解。3. 优化提示词Prompt明确要求“在得到足够信息后必须给出最终答案”。工具调用失败返回Invalid tool call工具定义与LLM绑定的工具描述不匹配或工具参数格式错误。1. 使用llm.bind_tools(tools)确保LLM知道工具的准确名称和参数。2. 检查工具函数的docstringLangChain会用它生成工具描述。3. 在AgentExecutor中设置handle_parsing_errorsTrue以捕获解析错误。RAG检索结果不相关文本分割策略不佳或检索器配置不当。1. 调整RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。2. 尝试不同的嵌入模型Embedding Model。3. 在检索时调整search_kwargs如{“k”: 5}返回更多结果或使用MMR搜索类型来平衡相关性与多样性。LangGraph图编译或运行出错状态State结构定义错误或节点函数返回值不符合预期。1. 仔细检查TypedDict的定义确保与节点函数返回的字典键匹配。2. 使用app.get_graph().draw_mermaid()输出图结构可视化检查节点和边是否正确连接。3. 在每个节点函数内打印日志跟踪状态的变化。程序运行慢频繁调用LLM或嵌入模型网络延迟高。1. 对于RAG考虑将向量数据库本地化如Chroma、FAISS避免每次查询都调用云端嵌入API。2. 使用缓存机制例如langchain.cache缓存LLM响应。3. 对于简单路由判断可尝试用规则关键词代替LLM调用。8. 最佳实践与项目进阶建议掌握了基础之后要打造一个健壮、可用的AI Agent系统还需要关注以下工程化实践8.1 提示词Prompt工程清晰的角色与指令在系统提示词System Prompt中明确Agent的角色、能力和约束。例如“你是一个专业的研究助手必须使用工具获取最新信息并在回答时引用来源。”少样本Few-Shot学习在提示词中提供1-2个高质量的输入输出示例能显著提升Agent执行复杂任务的准确性。结构化输出要求LLM以JSON等特定格式输出便于后续程序解析。这在多智能体协作中尤其重要。8.2 工具设计单一职责每个工具应只做一件事并做好。避免创建功能臃肿的“万能工具”。健壮的错误处理工具函数内部应有完善的try-except并返回结构化的错误信息让Agent能理解并采取补救措施。详细的描述工具的docstring至关重要LLM依靠它来决定何时以及如何调用工具。描述应清晰说明工具的用途、输入参数格式和输出示例。8.3 记忆Memory管理短期记忆使用ConversationBufferMemory或ConversationSummaryMemory来维护对话上下文。注意上下文长度限制对于长对话摘要记忆SummaryMemory是更好的选择。长期记忆对于需要记住跨会话信息的Agent可以将关键信息向量化后存入数据库在需要时通过RAG方式检索。这就是构建“数字分身”或“个性化助手”的基础。8.4 评估与监控构建测试集针对你的Agent常见任务准备一批标准问题及答案定期运行测试评估其准确性和稳定性。记录与审计记录每次Agent运行的完整链条Thought, Action, Observation这对于调试和优化至关重要。LangSmith是LangChain官方提供的优秀监控平台。人工反馈循环HITL在关键决策点引入人工审核。LangGraph原生支持“Human-in-the-Loop”节点可以在工作流中暂停并等待人工输入。8.5 学习路线与下一步夯实基础彻底理解本文中的代码尝试修改工具、调整提示词、构建不同的LangGraph工作流。探索高级模式学习ReAct,Plan-and-Execute,AutoGen微软的多智能体框架等高级Agent架构。深入LangGraph研究其Checkpointer实现持久化状态Supervisor实现多智能体调度以及Pregel并发执行模型。集成实际项目将Agent能力嵌入到你的Web应用用Flask/FastAPI、聊天机器人或自动化流程中。关注开源生态参与langchain-ai相关项目关注CrewAI,AutoGen等新兴框架保持对技术趋势的敏感。AI Agent的开发是一场结合了软件工程、提示词艺术和LLM能力的探索。从今天这个能调用搜索和知识库的简单助手开始逐步为其添加规划、记忆、协作和反思的能力你就能构建出真正智能、有用的应用程序。
)
