15天构建AI智能体：从RAG、LangGraph到工具调用的实战指南

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这里解释一下每个包的作用fastapiuvicorn: 我们的Web服务器框架和ASGI服务器。langgraph: 本次项目的核心用于构建有状态的、多步骤的工作流图。langchain: 一个庞大的AI应用开发框架提供了连接大模型、工具、记忆等组件的标准化接口。虽然我们主要用LangGraph编排但很多底层组件如Chat模型、文本分割器来自LangChain生态。langchain-openai: 官方维护的OpenAI模型集成。langchain-community: 社区贡献的各种第三方集成如向量数据库、工具。pydantic: FastAPI和LangChain都重度依赖的数据验证库。一个常见的坑是LangChain的版本兼容性。如果你在后续步骤中遇到导入错误可以尝试固定一个较新的稳定版本例如pip install langchain0.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(titleAI 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这里有几个关键点Pydantic模型AgentRequest和AgentResponse确保了API接口的输入输出格式清晰、可验证。LangGraph StateAgentState是一个TypedDict它定义了我们的工作流“记忆”了什么。Annotated[List[Dict], operator.add]是精髓所在。它告诉LangGraph当多个节点Node并行运行并试图修改messages时应该使用operator.add即列表的操作来合并结果而不是覆盖。这对于维护连贯的对话历史至关重要。next_step字段这是一个控制流变量。在简单的流程中可能不需要但当我们构建更复杂的、能自主决定“是先检索还是直接回答”的Agent时它会非常有用。2.3 构建第一个LangGraphHello 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(modelgpt-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_modelAgentResponse) 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( answerfinal_state[“answer”], session_idrequest.session_id or “default_session”, used_sources[] ) return response except Exception as e: raise HTTPException(status_code500, detailf”处理请求时出错: {str(e)}”) if __name__ “__main__”: import uvicorn uvicorn.run(app, host“0.0.0.0”, port8000)现在运行python main.py访问http://localhost:8000/docs你会看到自动生成的API文档。尝试调用/chat接口发送{“question”: “你好世界”}你应该会收到一个友好的中文回复。这个简单示例揭示了LangGraph的核心工作模式定义状态确定工作流需要处理哪些数据。创建节点函数每个节点是完成特定任务如调用LLM、检索文档的函数它读取和修改状态。构建图用StateGraph将节点连接起来定义执行顺序通过add_edge或条件分支后续会讲。编译与执行将图编译成一个可执行对象然后通过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_size1000, # 每个文本块的大小 chunk_overlap200, # 块之间的重叠避免上下文断裂 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_clsTextLoader) 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( documentssplits, embeddingself.embeddings, persist_directoryself.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_directoryself.persist_directory, embedding_functionself.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, kk) # 将检索到的文档内容合并成一个上下文字符串 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, k3) # 我们可以把来源信息也暂存起来方便最后输出 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: str3.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(contextcontext) # 构建完整的消息列表 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 EdgeLangGraph的强大之处在于它支持条件逻辑。我们可以根据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_modelAgentResponse) 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( answerfinal_state[“answer”], session_idrequest.session_id or “default_session”, used_sourcesfinal_state.get(“source_info”, []) # 返回引用的来源 ) return response except Exception as e: raise HTTPException(status_code500, detailf”处理请求时出错: {str(e)}”)现在重启你的FastAPI服务。尝试问两个问题“你好请介绍一下你自己。”– 这会走router - generate_response - END路径得到一个通用问候。“你们产品的核心功能是什么”– 假设你的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_modelAgentResponse) 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( answerfinal_state[“answer”], session_idsession_id, used_sourcesfinal_state.get(“source_info”, []) ) return response except Exception as e: raise HTTPException(status_code500, detailf”处理请求时出错: {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(tooltc[“name”], tool_inputtc[“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(contentstr(output), tool_call_idti.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实现了一个**ReActReasoning Acting**模式的基本循环模型思考决定是否用工具 - 执行工具 - 将结果返回给模型 - 模型再次思考给出最终答案。现在你的Agent就具备了记忆、知识库检索和工具调用的雏形。你可以通过不同的API端点来调用不同的图基础版和工具版或者将它们组合成一个更大的、更复杂的智能体工作流。这15天的旅程我们从零搭建了环境理解了LangGraph的状态和图概念实现了RAG知识库的构建与检索创建了具备条件路由的智能体流程并最终为其加上了记忆和工具调用的能力。这只是一个起点你可以在此基础上扩展集成更复杂的工具如网络搜索、数据库查询、优化路由逻辑使用小模型分类、实现流式输出、添加更持久化的记忆存储如向量数据库存储对话摘要以及设计更复杂的多智能体协作图。