AI Agent实战指南：从LangChain基础到LangGraph复杂工作流构建

1. 从零到一：AI Agent 到底是什么？

最近在技术社区和招聘网站上，“AI Agent”这个词的热度持续攀升，很多开发者朋友都跃跃欲试，想抓住这波技术浪潮。但面对海量的概念、框架和教程，往往感觉无从下手：什么是Agent？它和普通的大模型调用有什么区别？LangChain和LangGraph又是什么关系？

本文旨在为你拨开迷雾，提供一份系统、可落地的AI Agent实战指南。我们将从一个最简单的“单步工具调用”Agent开始，逐步深入到使用LangGraph构建具备复杂工作流和记忆能力的多智能体系统。无论你是想快速入门了解核心概念，还是希望将Agent技术集成到自己的项目中，这篇文章都将提供清晰的路径和可直接运行的代码示例。学完本文，你将能够独立搭建一个具备规划、执行、反思能力的智能体应用。

AI Agent（智能体），简单来说，是一个能够感知环境、自主决策并执行行动以实现特定目标的程序实体。它不仅仅是调用一次大模型API获取回答，而是通过“思考-行动-观察”的循环，像人类一样完成任务。

我们可以通过一个对比来理解：

• 传统大模型调用：用户问“今天北京天气如何？”，模型基于训练数据生成一段描述性文字。它无法获取实时数据。
• AI Agent：用户提出同样问题。Agent内部会进行规划：“要回答这个问题，我需要调用天气查询工具。” 然后执行行动：调用一个联网搜索或天气API工具，获取实时数据。最后，它观察工具返回的结果，并组织成自然语言回复给用户：“根据实时数据，北京今天晴，气温25°C。”

其核心组件通常包括：

1. 规划（Planning）：分解任务，制定步骤序列。
2. 工具使用（Tool Use）：调用外部API、数据库、函数等扩展能力。
3. 记忆（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-dotenv

2.2 获取并配置API密钥

AI Agent的运行依赖于大模型和外部工具服务，因此需要配置相应的API密钥。强烈建议使用环境变量来管理这些敏感信息，不要硬编码在代码中。

1. OpenAI API Key：访问 OpenAI平台 创建。
2. 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_results=2) # 限制每次搜索返回2条结果 return tool

3.2 初始化大语言模型（LLM）

Agent的“大脑”是一个大语言模型。我们使用OpenAI的GPT-4o模型（也可用gpt-3.5-turbo）。

# llm_setup.py from langchain_openai import ChatOpenAI def get_llm(model_name="gpt-4o", temperature=0): """ 初始化OpenAI聊天模型。 :param model_name: 模型名称，如 'gpt-4o', 'gpt-3.5-turbo' :param temperature: 创造性，0表示更确定，1表示更多样。 :return: ChatOpenAI实例 """ llm = ChatOpenAI(model=model_name, temperature=temperature, api_key=OPENAI_API_KEY) return llm

3.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(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) 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: str

Annotated[List[BaseMessage], operator.add]是一个高级用法，它告诉LangGraph在更新messages字段时，使用append操作而不是覆盖，这对于累积对话历史至关重要。

4.3 创建具有记忆的图工作流

我们将创建一个包含以下节点的工作流：

1. 路由节点：判断用户问题是简单聊天还是需要工具调用。
2. 工具调用节点：调用搜索工具。
3. 反思节点：检查工具返回的结果是否回答了问题，如果没有，则修改问题重新搜索。
4. 回答节点：生成最终答案。

# 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(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY")) search_tool = TavilySearchResults(max_results=2, api_key=os.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(content=str(result), tool_call_id=tool_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(content=question)], "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这两个紧密相关的概念。理解它们的区别与联系至关重要。

• RAG（Retrieval-Augmented Generation）： 一种架构模式，用于解决大模型的“知识截止”和“幻觉”问题。其核心流程是：用户提问 → 从知识库（如向量数据库）检索相关文档片段 → 将片段和问题一起交给大模型生成答案。RAG更像是一个增强的“问答系统”。
• AI Agent： 一个更宏观的架构概念，指能自主完成任务的智能体。一个Agent可以使用RAG作为其内部的一个工具。例如，一个研究助手Agent，其任务可能是“撰写一篇关于量子计算的报告”。它会规划步骤：1) 搜索最新论文（调用搜索工具），2) 阅读公司内部文档（调用RAG工具查询向量数据库），3) 整理大纲，4) 撰写内容，5) 检查格式。在这里，RAG是Agent工具箱里的一把“专用扳手”。

LangChain vs. LangGraph：

• LangChain： 提供了构建AI应用（包括RAG系统和简单Agent）所需的标准化组件和连接器。它抽象了与LLM、向量数据库、工具等的交互，让你用统一的API来操作。它的AgentExecutor已经能处理简单的多步任务。
• LangGraph： 是构建复杂、有状态、多参与者工作流的框架。当你的Agent需要循环、条件分支、持久化状态、多角色协作（如一个分析师Agent和一个审核员Agent）时，LangGraph比基础的AgentExecutor更合适。它让你以“图”的视角来设计和调试工作流。

简单说：用LangChain快速搭建应用，用LangGraph设计复杂流程。很多复杂的RAG系统（如包含查询重写、混合检索、重排序等步骤）本身也可以用LangGraph来构建。

6. 实战：构建一个本地知识库问答Agent（RAG + 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_size=1000, # 每个片段约1000字符 chunk_overlap=200, # 片段间重叠200字符，保持上下文 separators=["\n\n", "\n", "。", "！", "？", "；", "，", "、", " "] ) splits = text_splitter.split_documents(documents) print(f"文档分割为 {len(splits)} 个片段。") # 3. 生成向量并存储 embeddings = OpenAIEmbeddings(api_key=OPENAI_API_KEY) # 如果目录已存在，可以加载现有库，否则创建新库 if os.path.exists(persist_directory): print(f"从 {persist_directory} 加载已有向量库...") vectorstore = Chroma(persist_directory=persist_directory, embedding_function=embeddings) # 添加新文档（可选，这里我们假设重新创建） # vectorstore.add_documents(splits) else: print(f"创建新的向量库并存储到 {persist_directory} ...") vectorstore = Chroma.from_documents( documents=splits, embedding=embeddings, persist_directory=persist_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_key=OPENAI_API_KEY) _vectorstore = Chroma(persist_directory=VECTOR_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(model="gpt-4o", temperature=0, api_key=OPENAI_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( agent=agent, tools=tools, verbose=True, handle_parsing_errors=True, max_iterations=5, # 限制最大迭代次数，防止死循环 early_stopping_method="generate" # 当连续两个动作为“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过程中，你一定会遇到各种问题。以下是一些常见问题及其解决方案。

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 学习路线与下一步

1. 夯实基础：彻底理解本文中的代码，尝试修改工具、调整提示词、构建不同的LangGraph工作流。
2. 探索高级模式：学习ReAct,Plan-and-Execute,AutoGen（微软的多智能体框架）等高级Agent架构。
3. 深入LangGraph：研究其Checkpointer实现持久化状态，Supervisor实现多智能体调度，以及Pregel并发执行模型。
4. 集成实际项目：将Agent能力嵌入到你的Web应用（用Flask/FastAPI）、聊天机器人或自动化流程中。
5. 关注开源生态：参与langchain-ai相关项目，关注CrewAI,AutoGen等新兴框架，保持对技术趋势的敏感。

AI Agent的开发是一场结合了软件工程、提示词艺术和LLM能力的探索。从今天这个能调用搜索和知识库的简单助手开始，逐步为其添加规划、记忆、协作和反思的能力，你就能构建出真正智能、有用的应用程序。