1. 从零到一：为什么我们需要Agent、RAG与LangGraph？

如果你最近在AI应用开发圈子里待过，大概率会被这几个词刷屏：Agent、RAG、LangGraph。听起来很酷，但可能也让人有点懵：它们到底是什么？为什么突然这么火？更重要的是，我作为一个开发者，为什么要花时间学这个？

简单来说，这是构建下一代AI应用的三块核心拼图。过去，我们可能只是简单调用一下大模型的API，问个问题，拿个答案。但现在，用户的需求复杂了：他们希望AI能像人一样，拥有长期记忆，能调用工具（比如查数据库、发邮件），能根据复杂目标拆解任务并一步步执行。这就是“智能体”（Agent）的愿景。而RAG（检索增强生成）则是解决大模型“一本正经胡说八道”（幻觉问题）和知识过时问题的利器，它让AI在回答前，先去你的专属知识库（比如公司文档、产品手册）里查一查。LangGraph呢？你可以把它看作是给这些AI“零件”搭建工作流的“乐高积木”和“流程图工具”，它让多个AI步骤、工具调用、状态管理变得清晰、可控且可循环。

所以，这个系列的目标很明确：用15天时间，手把手带你从环境搭建开始，用Python和FastAPI，构建一个具备记忆、能使用工具、并能从专属知识库中获取信息的实用AI智能体。这不是一个理论课，而是一个完整的、可运行的代码实操项目。你会看到每一行代码，理解每一个设计决策背后的“为什么”，并最终得到一个可以部署、可以扩展的原型。

2. 环境准备与基础框架搭建：FastAPI + LangGraph

在开始构建复杂的AI逻辑之前，我们需要一个稳固的“地基”。这个地基就是我们的Web服务框架和核心的AI编排库。我选择FastAPI是因为它现代、快速，并且天生支持异步，这对于需要等待AI模型响应的应用来说至关重要。而LangGraph，则是我们编排Agent和RAG流程的“大脑”。

2.1 Python环境与核心依赖安装

首先，确保你有一个干净的Python环境（3.9以上）。我强烈建议使用venv或conda创建虚拟环境，避免包冲突。

# 创建并激活虚拟环境 python -m venv ai_agent_env source ai_agent_env/bin/activate # Linux/Mac # ai_agent_env\Scripts\activate # Windows # 安装核心依赖 pip install fastapi uvicorn langgraph langchain langchain-openai langchain-community pydantic

这里解释一下每个包的作用：

• fastapi&uvicorn: 我们的Web服务器框架和ASGI服务器。
• langgraph: 本次项目的核心，用于构建有状态的、多步骤的工作流图。
• langchain: 一个庞大的AI应用开发框架，提供了连接大模型、工具、记忆等组件的标准化接口。虽然我们主要用LangGraph编排，但很多底层组件（如Chat模型、文本分割器）来自LangChain生态。
• langchain-openai: 官方维护的OpenAI模型集成。
• langchain-community: 社区贡献的各种第三方集成（如向量数据库、工具）。
• pydantic: FastAPI和LangChain都重度依赖的数据验证库。

一个常见的坑是LangChain的版本兼容性。如果你在后续步骤中遇到导入错误，可以尝试固定一个较新的稳定版本，例如：pip install langchain==0.1.0。但通常使用最新版即可。

2.2 初始化FastAPI应用与LangGraph状态定义

我们的应用将提供一个API端点，接收用户的问题，然后通过我们定义的LangGraph工作流来处理，最后返回答案。首先，创建main.py。

# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Any, Optional import asyncio # 定义请求和响应的数据模型 class AgentRequest(BaseModel): question: str session_id: Optional[str] = None # 用于区分不同对话会话 class AgentResponse(BaseModel): answer: str session_id: str used_sources: Optional[List[str]] = [] # 如果用了RAG，这里可以返回引用的文档来源 # 初始化FastAPI应用 app = FastAPI(title="AI Agent with RAG & LangGraph") # 定义LangGraph的状态（State） # 这是LangGraph的核心概念，它定义了工作流在执行过程中携带和修改的数据 from typing import TypedDict, Annotated import operator class AgentState(TypedDict): # 用户输入的问题 question: str # 对话历史（用于记忆） messages: Annotated[List[Dict], operator.add] # 关键！这个注解告诉LangGraph如何合并这个字段（追加到列表） # 从知识库检索到的上下文 context: Optional[str] # 最终生成的答案 answer: Optional[str] # 当前决定要做什么（例如：“需要检索”，“直接回答”） next_step: str

这里有几个关键点：

1. Pydantic模型：AgentRequest和AgentResponse确保了API接口的输入输出格式清晰、可验证。
2. LangGraph State：AgentState是一个TypedDict，它定义了我们的工作流“记忆”了什么。Annotated[List[Dict], operator.add]是精髓所在。它告诉LangGraph，当多个节点（Node）并行运行并试图修改messages时，应该使用operator.add（即列表的+操作）来合并结果，而不是覆盖。这对于维护连贯的对话历史至关重要。
3. next_step字段：这是一个控制流变量。在简单的流程中可能不需要，但当我们构建更复杂的、能自主决定“是先检索还是直接回答”的Agent时，它会非常有用。

2.3 构建第一个LangGraph：Hello Agent

在深入RAG和复杂逻辑之前，我们先构建一个最简单的LangGraph，它只做一件事：调用大模型，生成一句问候语。这能帮助我们理解LangGraph的基本结构：图（Graph）、节点（Node）、边（Edge）。

# 在 main.py 中继续添加 from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI import os # 设置你的OpenAI API Key（实际项目中请使用环境变量或配置管理） os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 初始化一个聊天模型 llm = ChatOpenAI(model="gpt-3.5-turbo") def call_llm(state: AgentState): """节点函数：调用大模型生成回答""" # 从状态中获取最新的用户消息（假设最后一条是用户输入） last_message = state[“messages”][-1] user_input = last_message[“content”] # 构造一个简单的系统提示 system_prompt = “你是一个友好的助手。请用中文回答。” human_input = user_input # 调用模型 response = llm.invoke([ {“role”: “system”, “content”: system_prompt}, {“role”: “user”, “content”: human_input} ]) # 更新状态：将模型的回复添加到消息历史中 new_message = {“role”: “assistant”, “content”: response.content} return {“messages”: [new_message], “answer”: response.content} # 构建图 workflow = StateGraph(AgentState) # 添加节点。节点就是一个可调用的函数。 workflow.add_node(“generate_response”, call_llm) # 设置入口点。图从哪个节点开始执行。 workflow.set_entry_point(“generate_response”) # 设置出口点。执行完`generate_response`节点后，图就结束。 workflow.add_edge(“generate_response”, END) # 编译图，得到一个可执行的对象 app.graph = workflow.compile() @app.post(“/chat”, response_model=AgentResponse) async def chat_endpoint(request: AgentRequest): """处理用户聊天的API端点""" try: # 初始化状态。注意messages的格式，它需要符合ChatModel的调用约定。 initial_state: AgentState = { “question”: request.question, “messages”: [{“role”: “user”, “content”: request.question}], “context”: None, “answer”: None, “next_step”: “start” } # 执行编译好的图，传入初始状态 final_state = await app.graph.ainvoke(initial_state) # 构建响应 response = AgentResponse( answer=final_state[“answer”], session_id=request.session_id or “default_session”, used_sources=[] ) return response except Exception as e: raise HTTPException(status_code=500, detail=f”处理请求时出错: {str(e)}”) if __name__ == “__main__”: import uvicorn uvicorn.run(app, host=“0.0.0.0”, port=8000)

现在，运行python main.py，访问http://localhost:8000/docs，你会看到自动生成的API文档。尝试调用/chat接口，发送{“question”: “你好，世界！”}，你应该会收到一个友好的中文回复。

这个简单示例揭示了LangGraph的核心工作模式：

1. 定义状态：确定工作流需要处理哪些数据。
2. 创建节点函数：每个节点是完成特定任务（如调用LLM、检索文档）的函数，它读取和修改状态。
3. 构建图：用StateGraph将节点连接起来，定义执行顺序（通过add_edge）或条件分支（后续会讲）。
4. 编译与执行：将图编译成一个可执行对象，然后通过ainvoke（异步）或invoke（同步）传入初始状态来运行它。

3. 构建RAG知识库：从文档到向量检索

现在，我们的Agent能说话了，但它还是个“文盲”，对自己的知识库一无所知。接下来，我们给它装上“眼睛”和“记忆”——构建一个RAG系统。RAG的核心流程是：文档加载 -> 文本分割 -> 向量化 -> 存储 -> 检索。

3.1 文档加载与预处理

我们假设你的知识库是一些Markdown、PDF或TXT文件。这里以TXT为例。创建一个knowledge_base文件夹，里面放一些.txt文档。

# rag_processor.py from langchain_community.document_loaders import TextLoader, DirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 我们使用轻量级的Chroma作为向量数据库 import os class RAGKnowledgeBase: def __init__(self, persist_directory=“./chroma_db”): self.persist_directory = persist_directory self.embeddings = OpenAIEmbeddings(model=“text-embedding-3-small”) # 使用OpenAI的嵌入模型 self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个文本块的大小 chunk_overlap=200, # 块之间的重叠，避免上下文断裂 separators=[“\n\n”, “\n”, “。”, “！”, “？”, “,”, “ “, “”] # 中文友好的分隔符 ) self.vector_store = None def load_and_split_documents(self, data_path=“./knowledge_base”): """加载目录下的所有文本文件并进行分割""" # 使用通配符加载所有txt文件 loader = DirectoryLoader(data_path, glob=“**/*.txt”, loader_cls=TextLoader) documents = loader.load() print(f”已加载 {len(documents)} 个文档。”) # 分割文档 splits = self.text_splitter.split_documents(documents) print(f”文档被分割成 {len(splits)} 个文本块。”) return splits def create_vector_store(self, splits): """创建向量存储并持久化""" # 使用Chroma.from_documents，它会自动将文本转换为向量并存储 self.vector_store = Chroma.from_documents( documents=splits, embedding=self.embeddings, persist_directory=self.persist_directory ) self.vector_store.persist() # 持久化到磁盘 print(f”向量数据库已创建并保存至 {self.persist_directory}”) def load_existing_vector_store(self): """加载已存在的向量数据库""" if os.path.exists(self.persist_directory): self.vector_store = Chroma( persist_directory=self.persist_directory, embedding_function=self.embeddings ) print(“已加载现有向量数据库。”) return True else: print(“未找到已有的向量数据库。”) return False def similarity_search(self, query: str, k: int = 4): """在知识库中进行相似性检索""" if self.vector_store is None: if not self.load_existing_vector_store(): raise ValueError(“向量数据库未初始化且不存在。”) docs = self.vector_store.similarity_search(query, k=k) # 将检索到的文档内容合并成一个上下文字符串 context = “\n\n”.join([doc.page_content for doc in docs]) return context, docs # 返回上下文和原始文档对象（可用于引用溯源）

关键参数解析与避坑指南：

• chunk_size和chunk_overlap：这是RAG效果的关键。chunk_size太大，检索到的信息可能不精准；太小，则可能丢失重要上下文。chunk_overlap用于保持语义连贯。对于中文，1000-1500的chunk_size和200的overlap是个不错的起点。
• separators：默认的分隔符是针对英文的。我们加入了中文标点，如“。”、“！”，这样能更好地按句子或意群分割中文文本。
• 向量数据库选择：Chroma轻量、易用，适合本地开发和原型。生产环境可以考虑Weaviate、Qdrant或Pinecone等。
• 嵌入模型：text-embedding-3-small在效果和成本间取得了很好的平衡。确保你的OpenAI账户有该模型的权限。

3.2 将RAG集成到LangGraph节点中

现在，我们需要创建一个LangGraph节点，专门负责检索。修改我们的AgentState，加入context字段来承载检索结果。

# 在 main.py 中更新和添加 from rag_processor import RAGKnowledgeBase # 初始化RAG知识库 rag_kb = RAGKnowledgeBase() # 假设第一次运行，需要构建索引（后续可以注释掉） # splits = rag_kb.load_and_split_documents() # rag_kb.create_vector_store(splits) # 或者直接加载现有的 rag_kb.load_existing_vector_store() def retrieve_context(state: AgentState): """节点函数：从知识库中检索相关上下文""" question = state[“question”] print(f”正在检索与问题相关的内容: {question}”) try: # 调用我们上面写的检索函数 context, source_docs = rag_kb.similarity_search(question, k=3) # 我们可以把来源信息也暂存起来，方便最后输出 source_info = [doc.metadata.get(“source”, “Unknown”) for doc in source_docs] return { “context”: context, “source_info”: source_info, # 注意，我们需要在State中定义这个字段才能用 “next_step”: “generate_with_context” # 检索完后，下一步是生成 } except Exception as e: print(f”检索失败: {e}”) # 即使检索失败，也继续流程，但上下文为空 return {“context”: “”, “next_step”: “generate_with_context”}

我们需要更新AgentState以包含source_info。

class AgentState(TypedDict): question: str messages: Annotated[List[Dict], operator.add] context: Optional[str] source_info: Optional[List[str]] # 新增：检索到的文档来源 answer: Optional[str] next_step: str

3.3 创建“生成”节点：融合上下文进行回答

现在，我们有了上下文，需要一个新的生成节点，它能利用检索到的context来生成答案。

def generate_answer_with_context(state: AgentState): """节点函数：利用检索到的上下文生成答案""" question = state[“question”] context = state.get(“context”, “”) chat_history = state[“messages”][:-1] # 获取历史消息（可能包含多轮对话） # 构建一个增强版的系统提示，指导模型使用上下文 system_message = “””你是一个专业的助手，请根据用户的问题和提供的相关上下文来回答问题。 上下文信息如下： {context} 请注意： 1. 如果上下文信息足以回答问题，请严格基于上下文回答。 2. 如果上下文信息不足或与问题无关，你可以运用自己的知识来回答，但请说明这一点。 3. 回答请使用中文，并力求准确、清晰。 “””.format(context=context) # 构建完整的消息列表 messages_for_llm = [{“role”: “system”, “content”: system_message}] # 加入历史对话（如果有） messages_for_llm.extend(chat_history) # 加入当前用户问题 messages_for_llm.append({“role”: “user”, “content”: question}) # 调用模型 response = llm.invoke(messages_for_llm) # 更新状态 new_message = {“role”: “assistant”, “content”: response.content} return { “messages”: [new_message], “answer”: response.content, “next_step”: “end” # 生成完毕，进入结束状态 }

4. 组装智能体：用LangGraph编排RAG与生成流程

现在我们有三个核心节点了：retrieve_context、generate_answer_with_context，以及最初那个简单的generate_response。如何把它们智能地组织起来？我们可以设计一个简单的路由逻辑：先判断问题是否需要检索知识库。

4.1 创建“路由”节点

这个节点像一个调度员，根据用户问题的性质，决定走哪条路。

def route_question(state: AgentState): """节点函数：判断问题类型，决定下一步是检索还是直接回答""" question = state[“question”].lower() # 这里实现一个非常简单的路由逻辑 # 例如，如果问题包含特定关键词或是一般性问候，则直接回答 general_keywords = [“你好”, “hi”, “hello”, “你是谁”, “你的名字”] needs_retrieval_keywords = [“文档”, “知识库”, “产品”, “功能”, “如何”, “什么”, “为什么”] # 示例关键词 # 检查是否是通用问候 if any(keyword in question for keyword in general_keywords): return {“next_step”: “generate_response”} # 直接去简单生成节点 # 检查是否需要检索（这里逻辑可以很复杂，比如用一个小型分类模型） # 为了简单，我们假设需要检索关键词的问题就去检索 if any(keyword in question for keyword in needs_retrieval_keywords): return {“next_step”: “retrieve”} # 默认情况，也去检索，因为我们的Agent主要功能是问答知识库 return {“next_step”: “retrieve”}

4.2 构建条件边（Conditional Edge）

LangGraph的强大之处在于它支持条件逻辑。我们可以根据route_question节点输出的next_step值，动态决定下一个执行哪个节点。

# 在 main.py 中重构图的构建部分 from langgraph.graph import StateGraph, END # 重新构建图 workflow = StateGraph(AgentState) # 添加所有节点 workflow.add_node(“router”, route_question) # 路由节点 workflow.add_node(“retrieve”, retrieve_context) # 检索节点 workflow.add_node(“generate_with_context”, generate_answer_with_context) # 带上下文的生成节点 workflow.add_node(“generate_response”, call_llm) # 直接生成节点（用于简单问候） # 设置入口点为路由节点 workflow.set_entry_point(“router”) # 定义条件边：根据`state[‘next_step’]`的值决定下一步 def decide_next_step(state): return state[“next_step”] workflow.add_conditional_edges( “router”, # 源节点 decide_next_step, # 条件函数，返回下一个节点的名字 { “retrieve”: “retrieve”, # 如果返回”retrieve”，则跳转到”retrieve”节点 “generate_response”: “generate_response”, # 如果返回”generate_response”，则跳转到”generate_response”节点 # 理论上，router也可以直接指向”generate_with_context”，但这里我们设计为检索后生成 } ) # 添加普通边：检索完成后，必然去生成 workflow.add_edge(“retrieve”, “generate_with_context”) # 直接生成节点完成后，直接结束 workflow.add_edge(“generate_response”, END) # 带上下文的生成节点完成后，也结束 workflow.add_edge(“generate_with_context”, END) # 编译图 app.graph = workflow.compile()

4.3 更新API端点以支持完整流程

现在，我们的API端点将执行这个更复杂的图。

@app.post(“/chat”, response_model=AgentResponse) async def chat_endpoint(request: AgentRequest): try: initial_state: AgentState = { “question”: request.question, “messages”: [{“role”: “user”, “content”: request.question}], “context”: None, “source_info”: None, “answer”: None, “next_step”: “” # 初始为空，由router节点填充 } # 执行图 final_state = await app.graph.ainvoke(initial_state) response = AgentResponse( answer=final_state[“answer”], session_id=request.session_id or “default_session”, used_sources=final_state.get(“source_info”, []) # 返回引用的来源 ) return response except Exception as e: raise HTTPException(status_code=500, detail=f”处理请求时出错: {str(e)}”)

现在，重启你的FastAPI服务。尝试问两个问题：

1. “你好，请介绍一下你自己。”– 这会走router -> generate_response -> END路径，得到一个通用问候。
2. “你们产品的核心功能是什么？”– 假设你的knowledge_base文档里有产品介绍，这会走router -> retrieve -> generate_with_context -> END路径，模型会从你提供的文档中寻找答案并生成回复。

5. 为智能体添加记忆与工具调用能力

一个真正的智能体（Agent）不仅要有知识（RAG），还要有记忆（记住对话历史）和能力（调用外部工具）。让我们进一步完善它。

5.1 实现对话记忆

目前，我们的messages列表已经保存了历史，但在多轮对话中，我们需要确保每次新的问题都能看到完整的对话历史。当前的图设计在每次调用时都会用新的用户消息初始化messages，这丢失了历史。我们需要修改状态初始化和节点逻辑。

修改状态初始化（在API端点中）： 我们需要一个方式来存储和读取对话历史。一个简单的方法是用session_id作为键，在内存或外部存储（如Redis）中保存messages。这里为了演示，使用一个全局字典（生产环境请用数据库）。

# 简单的内存存储（非生产环境使用） conversation_memory = {} @app.post(“/chat”, response_model=AgentResponse) async def chat_endpoint(request: AgentRequest): try: session_id = request.session_id or “default_session” # 从内存中获取该会话的历史消息，如果没有则初始化 historical_messages = conversation_memory.get(session_id, []) # 将当前用户问题添加到历史中 updated_messages = historical_messages + [{“role”: “user”, “content”: request.question}] initial_state: AgentState = { “question”: request.question, “messages”: updated_messages, # 使用包含历史的消息列表 “context”: None, “source_info”: None, “answer”: None, “next_step”: “” } final_state = await app.graph.ainvoke(initial_state) # 将助手的回复也存入历史 assistant_message = {“role”: “assistant”, “content”: final_state[“answer”]} conversation_memory[session_id] = updated_messages + [assistant_message] # 注意：这里只保存最近N轮，避免内存爆炸，可以加个长度限制 response = AgentResponse( answer=final_state[“answer”], session_id=session_id, used_sources=final_state.get(“source_info”, []) ) return response except Exception as e: raise HTTPException(status_code=500, detail=f”处理请求时出错: {str(e)}”)

修改生成节点：generate_answer_with_context和call_llm节点现在会接收到包含完整历史的messages，它们需要正确处理。在我们的实现中，generate_answer_with_context已经使用了chat_history = state[“messages”][:-1]，这包含了之前所有的消息（用户和助手）。call_llm节点也需要类似调整，使其能处理多轮对话。

5.2 为智能体添加工具调用能力

工具（Tools）是Agent的“手”和“脚”。例如，让Agent能查询天气、计算数学、搜索网络。LangGraph和LangChain对工具有很好的支持。

首先，定义一个简单的工具：比如一个计算器工具。

from langchain.tools import tool from datetime import datetime @tool def get_current_time(query: str) -> str: “”“获取当前的日期和时间。当用户询问时间、日期、现在几点时使用此工具。输入参数通常可以忽略。”“” now = datetime.now() return now.strftime(“%Y-%m-%d %H:%M:%S”) @tool def simple_calculator(expression: str) -> str: “”“执行简单的数学计算。支持加减乘除（+ - * /）和括号。例如：’(3 + 5) * 2‘。”“” # 警告：使用eval有安全风险，此处仅用于演示。生产环境应使用安全的表达式解析库（如`ast.literal_eval`配合限制）。 try: # 非常简单的安全过滤（生产环境需要更严格的检查） allowed_chars = set(“0123456789+-*/(). “) if not all(c in allowed_chars for c in expression): return “错误：表达式中包含不允许的字符。” result = eval(expression) return str(result) except Exception as e: return f”计算错误: {e}”

然后，创建一个能使用工具的节点：我们需要修改图的结构，加入一个“决定是否使用工具”的节点，以及一个“执行工具”的节点。这通常通过让LLM生成一个包含工具调用请求的特定格式消息来实现，但LangGraph提供了更优雅的ToolNode和tools_condition。

为了简化，我们创建一个集成了工具调用能力的生成节点。这需要用到LangChain的bind_tools和with_structured_output等功能。这是一个更高级的集成，展示了如何让LLM主动选择工具。

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.prebuilt import ToolExecutor, ToolInvocation import json # 创建工具列表 tools = [get_current_time, simple_calculator] tool_executor = ToolExecutor(tools) # 工具执行器 def agent_node(state: AgentState): """一个更高级的Agent节点，可以决定是否调用工具""" messages = state[“messages”] # 1. 让LLM根据对话历史和工具定义，决定下一步 # 我们需要一个支持工具调用的模型调用 llm_with_tools = llm.bind_tools(tools) # 调用模型，它可能会返回一个包含ToolCall内容的AIMessage response = llm_with_tools.invoke(messages) # 2. 检查响应中是否包含工具调用 if response.tool_calls: # 如果有工具调用，我们需要执行它们 tool_invocations = [] for tc in response.tool_calls: # 构建工具调用请求 ti = ToolInvocation(tool=tc[“name”], tool_input=tc[“args”]) tool_invocations.append(ti) # 执行所有工具 tool_outputs = tool_executor.batch(tool_invocations) # 3. 将工具执行结果封装成ToolMessage，并添加到消息历史中 tool_messages = [] for ti, output in zip(tool_invocations, tool_outputs): tool_msg = ToolMessage(content=str(output), tool_call_id=ti.id) tool_messages.append(tool_msg) # 返回更新后的状态：包含模型的响应（含工具调用请求）和工具的执行结果 return {“messages”: [response] + tool_messages} else: # 如果没有工具调用，直接返回模型的回答 return {“messages”: [response], “answer”: response.content}

最后，重构我们的图：我们需要一个新的图，它循环运行agent_node，直到模型不再调用工具，给出最终答案。这需要引入“循环”的概念。

from langgraph.graph import StateGraph, END from langgraph.graph import MessagesState # 可以使用一个预定义的状态，简化消息处理 # 为了简化，我们使用LangGraph预定义的MessagesState，它专门为消息对话设计 from typing import TypedDict, List from langgraph.graph import add_messages class AgentStateWithTools(TypedDict): messages: Annotated[List, add_messages] # 使用add_messages操作符 # 可以添加其他需要的字段，如context # 创建新图 workflow_with_tools = StateGraph(AgentStateWithTools) # 添加节点 workflow_with_tools.add_node(“agent”, agent_node) # 这个agent节点集成了工具调用判断 # 定义条件边：检查上一步的最后一个消息是否是工具调用 def should_continue(state): messages = state[“messages”] last_message = messages[-1] # 如果最后一条消息是AIMessage且包含了工具调用，说明还需要继续（让模型基于工具结果再回答） if hasattr(last_message, ‘tool_calls’) and last_message.tool_calls: return “agent” # 继续循环 else: return END # 结束 workflow_with_tools.set_entry_point(“agent”) workflow_with_tools.add_conditional_edges(“agent”, should_continue) # 编译图 app.graph_with_tools = workflow_with_tools.compile()

这个新的图app.graph_with_tools实现了一个**ReAct（Reasoning + Acting）**模式的基本循环：模型思考（决定是否用工具） -> 执行工具 -> 将结果返回给模型 -> 模型再次思考给出最终答案。

现在，你的Agent就具备了记忆、知识库检索和工具调用的雏形。你可以通过不同的API端点来调用不同的图（基础版和工具版），或者将它们组合成一个更大的、更复杂的智能体工作流。

这15天的旅程，我们从零搭建了环境，理解了LangGraph的状态和图概念，实现了RAG知识库的构建与检索，创建了具备条件路由的智能体流程，并最终为其加上了记忆和工具调用的能力。这只是一个起点，你可以在此基础上扩展：集成更复杂的工具（如网络搜索、数据库查询）、优化路由逻辑（使用小模型分类）、实现流式输出、添加更持久化的记忆存储（如向量数据库存储对话摘要），以及设计更复杂的多智能体协作图。