从零构建AI智能体：原生Agent、RAG与LangGraph实战指南

1. 项目概述为什么是Agent、RAG与LangGraph如果你最近在关注AI应用开发尤其是想从“调用API”升级到“构建智能体”那么“原生Agent、RAG与LangGraph”这个组合几乎是你绕不开的核心技术栈。我花了半个月时间从零开始把这三个概念揉在一起做了一套完整的代码实操。这不仅仅是学习几个库而是理解如何让大语言模型LLM从“一个聪明的聊天机器人”变成“一个能自主使用工具、查询知识、并管理复杂工作流的智能代理”。简单来说Agent是大脑负责决策和规划RAG是外挂的知识库让大脑不再“一本正经地胡说八道”而LangGraph则是神经中枢用图Graph的方式清晰地定义大脑思考、行动、等待反馈的整个循环流程。单独学任何一个都只能解决局部问题。但当你把它们串联起来就能构建出真正实用、可落地的AI应用比如智能客服、数据分析助手、自动化报告生成工具等等。接下来的内容我会完全基于代码实操带你走过这15天的核心旅程。没有空洞的理论所有解释都会落在具体的代码行和运行结果上。无论你是刚入门Python的开发者还是已经用过LangChain想寻求更优解的工程师都能找到可以直接“抄作业”的路径。2. 核心架构与工具选型解析在动手写第一行代码之前搞清楚“为什么是这三个”以及“用什么工具实现”至关重要。这决定了整个项目的工程化和可维护性。2.1 技术栈深度拆解从Why到How原生Agent这里的“原生”指的是不依赖LangChain等高层框架直接基于OpenAI的Function Calling或Assistant API来构建代理的核心逻辑。为什么要“原生”因为LangChain这类框架虽然开箱即用但抽象层次高在定制复杂逻辑、调试和性能优化时你可能会感觉像在隔着一层毛玻璃操作。直接使用底层的API能让你对Agent的每一次思考Reasoning、每一次工具调用Tool Call有完全的控制力理解其最本质的工作机制。这是我们项目坚实的地基。RAG检索增强生成。它的核心价值是解决LLM的“幻觉”和知识滞后问题。原理不复杂将你的私有文档PDF、Word、网页切片、向量化后存入向量数据库当用户提问时先从向量库中检索出最相关的文档片段最后将问题和这些片段一起交给LLM让它基于这些“证据”来生成答案。关键在于如何设计高效的文本切片策略、选择合适的嵌入模型、以及优化检索后的重排序Re-ranking这些直接决定了RAG系统的效果上限。LangGraph这是LangChain团队推出的新库用于构建有状态的、多环节的代理工作流。你可以把它想象成画一个流程图里面的节点Node是函数比如“调用LLM”、“执行工具”边Edge定义了流程走向。它的杀手级特性是支持“循环”Cycle这正是Agent运行的核心模式思考 - 决定调用工具 - 执行工具 - 观察结果 - 继续思考… 直到任务完成。用LangGraph来管理Agent的生命周期比用if-else或状态机代码清晰、健壮得多。2.2 工具与库的精准选择基于以上拆解我们的工具选型如下LLM与Agent基础OpenAI API。这是事实上的标准其Function Calling功能是构建Agent的基石。我们将直接使用openai官方Python包。RAG向量数据库ChromaDB。轻量、易用、纯Python、可持久化非常适合本地开发和中小型项目。相比Milvus或Pinecone它无需复杂部署学习曲线平缓。嵌入模型OpenAI的text-embedding-3-small。在效果、速度和成本间取得了很好的平衡。对于完全离线的场景可以后续替换为BAAI/bge-small-zh-v1.5等开源模型。工作流编排LangGraph。它是我们项目的“总导演”负责调度Agent和RAG。我们将重点学习其StateGraph和MessagesState的概念。Web框架与部署FastAPI。异步特性好性能高自动生成API文档。我们将用它把整个智能体封装成HTTP服务方便前端或其他系统调用。开发环境Python 3.10Poetry管理依赖比pip更清晰VS Code作为IDE。注意选择ChromaDB和OpenAI Embedding是基于“快速上手和演示”的考量。在生产环境中你需要根据数据规模、延迟要求、成本预算来重新评估比如向量数据库可能升级为Qdrant或Weaviate嵌入模型可能换成本地部署的MTEB榜单上的佼佼者。这个技术栈组合既保证了核心概念学习的纯粹性原生Agent又涵盖了从知识处理RAG到流程编排LangGraph再到服务化FastAPI的完整应用链路。3. 第1-5天构建你的第一个原生智能体前五天我们的目标是抛开所有脚手架亲手组装一个能理解指令、并调用简单工具的Agent。3.1 环境搭建与OpenAI基础配置首先用Poetry创建一个干净的项目环境。这能避免未来令人头疼的依赖冲突。# 安装Poetry (如果未安装) curl -sSL https://install.python-poetry.org | python3 - # 创建项目目录并初始化 mkdir ai-agent-project cd ai-agent-project poetry init -n # 交互式创建pyproject.toml这里用-n跳过交互 poetry add openai python-dotenv创建.env文件存放你的OpenAI API密钥永远不要把它硬编码在代码里# .env OPENAI_API_KEYsk-your-secret-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果是Azure或代理需修改接下来编写一个基础工具类。我们从一个最简单的“计算器”工具开始模拟Agent调用外部功能的能力。# core/tools.py import json import math from typing import Dict, Any class CalculatorTool: 一个简单的计算器工具演示如何定义Agent可调用的函数。 name “calculator” description “用于执行数学计算。输入应为包含‘operation’和‘numbers’的JSON字符串。” classmethod def get_schema(cls) - Dict[str, Any]: 返回OpenAI Function Calling所需的函数模式。 return { “type”: “function”, “function”: { “name”: cls.name, “description”: cls.description, “parameters”: { “type”: “object”, “properties”: { “operation”: { “type”: “string”, “enum”: [“add”, “subtract”, “multiply”, “divide”, “sqrt”], “description”: “要执行的运算类型。” }, “numbers”: { “type”: “array”, “items”: {“type”: “number”}, “description”: “参与运算的数字列表。对于‘sqrt’运算只需第一个元素。” } }, “required”: [“operation”, “numbers”] } } } classmethod def execute(cls, operation: str, numbers: list) - float: 执行具体的计算逻辑。 try: if operation “add”: return sum(numbers) elif operation “subtract”: return numbers[0] - sum(numbers[1:]) elif operation “multiply”: result 1 for num in numbers: result * num return result elif operation “divide”: if len(numbers) ! 2: raise ValueError(“除法运算需要且仅需要两个数字。”) if numbers[1] 0: raise ZeroDivisionError(“除数不能为零。”) return numbers[0] / numbers[1] elif operation “sqrt”: if numbers[0] 0: raise ValueError(“不能对负数开平方根。”) return math.sqrt(numbers[0]) else: raise ValueError(f“不支持的运算类型{operation}”) except Exception as e: return f“计算错误{str(e)}”这个CalculatorTool类做了几件关键事定义了工具名和描述LLM靠这个决定是否调用它提供了符合OpenAI规范的函数模式get_schema并实现了具体的执行逻辑execute。这是所有工具类的通用模板。3.2 实现Agent的核心推理循环有了工具接下来是Agent的大脑。我们将实现一个简单的循环让LLM根据对话历史和可用工具决定下一步是“直接回答”还是“调用工具”。# core/agent.py import os import json from typing import List, Dict, Any from openai import OpenAI from dotenv import load_dotenv from .tools import CalculatorTool load_dotenv() class NativeAgent: def __init__(self, model: str “gpt-3.5-turbo”): self.client OpenAI(api_keyos.getenv(“OPENAI_API_KEY”)) self.model model self.available_tools [CalculatorTool] # 未来可以扩展更多工具 self.conversation_history: List[Dict[str, Any]] [] # 保存对话消息 def _get_tools_schema(self): 获取所有可用工具的Function Calling模式。 return [tool.get_schema() for tool in self.available_tools] def run(self, user_input: str) - str: 运行一轮Agent推理循环。 # 1. 将用户输入加入历史 self.conversation_history.append({“role”: “user”, “content”: user_input}) # 2. 调用LLM传入历史对话和工具定义 response self.client.chat.completions.create( modelself.model, messagesself.conversation_history, toolsself._get_tools_schema(), tool_choice“auto”, # 让模型自行决定是否调用工具 ) message response.choices[0].message # 3. 将模型的响应无论是否包含工具调用加入历史 self.conversation_history.append(message.to_dict()) # 4. 检查模型是否决定调用工具 if message.tool_calls: # 5. 执行所有被调用的工具 tool_outputs [] for tool_call in message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) # 找到对应的工具类并执行 for Tool in self.available_tools: if Tool.name func_name: result Tool.execute(**func_args) tool_outputs.append({ “tool_call_id”: tool_call.id, “role”: “tool”, “name”: func_name, “content”: str(result), }) break # 6. 将工具执行结果作为消息再次加入历史 self.conversation_history.extend(tool_outputs) # 7. 携带工具结果再次调用LLM让它生成面向用户的最终回答 second_response self.client.chat.completions.create( modelself.model, messagesself.conversation_history, ) final_message second_response.choices[0].message self.conversation_history.append(final_message.to_dict()) return final_message.content else: # 模型没有调用工具直接返回其回复 return message.content # 测试一下 if __name__ “__main__”: agent NativeAgent() print(“Agent已启动输入‘quit’退出。”) while True: query input(“\n你 “) if query.lower() “quit”: break answer agent.run(query) print(f“Agent {answer}”)运行这个脚本试试问它 “123加456等于多少” 或者 “计算16的平方根”。你会看到控制台里Agent先输出一个包含tool_calls的中间响应然后执行计算器工具最后给出包含计算结果的最终答案。这就是一个最简Agent的完整心跳。实操心得在调试时强烈建议将self.conversation_history打印出来。你能清晰地看到LLM、工具、用户三者之间消息的交替这对于理解Agent的“思考过程”和排查问题至关重要。这也是“原生”开发带来的最大好处——完全的透明度和控制力。3.3 为Agent增添更多能力搜索与文件读取单一的计算器工具显然不够。接下来两天我们集成两个实用工具一个模拟的网络搜索和一个简单的文本文件读取器。这将让你掌握如何扩展Agent的能力边界。# core/tools.py (新增部分) import requests from pathlib import Path class WebSearchTool: 模拟网络搜索工具实际调用一个公共API如DuckDuckGo Instant Answer。 name “web_search” description “用于搜索网络上的最新信息。输入是一个搜索查询字符串。” classmethod def get_schema(cls): return { “type”: “function”, “function”: { “name”: cls.name, “description”: cls.description, “parameters”: { “type”: “object”, “properties”: { “query”: {“type”: “string”, “description”: “搜索关键词”} }, “required”: [“query”] } } } classmethod def execute(cls, query: str): # 注意这是一个模拟。真实场景应使用SerperAPI、SerpAPI或Bing Search API。 # 这里使用DuckDuckGo的HTML抓取作为示例仅用于演示可能不稳定。 try: url f“https://api.duckduckgo.com/?q{requests.utils.quote(query)}formatjsonpretty1” resp requests.get(url, timeout10) data resp.json() # 提取摘要信息 abstract data.get(‘AbstractText’, ‘’) if not abstract: abstract data.get(‘RelatedTopics’, [{}])[0].get(‘Text’, ‘未找到相关信息’) return f“搜索 ‘{query}’ 的结果{abstract[:300]}...” # 截断防止过长 except Exception as e: return f“搜索失败{str(e)}” class FileReadTool: 读取本地文本文件内容的工具。 name “read_file” description “读取指定路径的文本文件内容。输入是文件的绝对或相对路径。” classmethod def get_schema(cls): return { “type”: “function”, “function”: { “name”: cls.name, “description”: cls.description, “parameters”: { “type”: “object”, “properties”: { “file_path”: {“type”: “string”, “description”: “待读取文件的路径”} }, “required”: [“file_path”] } } } classmethod def execute(cls, file_path: str): path Path(file_path) if not path.exists(): return f“错误文件 ‘{file_path}’ 不存在。” if not path.is_file(): return f“错误’{file_path}’ 不是一个文件。” try: content path.read_text(encoding‘utf-8’) return f“文件 ‘{file_path}’ 的内容前1000字符\n{content[:1000]}” except Exception as e: return f“读取文件失败{str(e)}”然后在NativeAgent的__init__方法中将新工具加入列表self.available_tools [CalculatorTool, WebSearchTool, FileReadTool]现在你的Agent可以回答 “今天北京的天气怎么样”它会尝试搜索或者你让它 “读一下 ./README.md 文件的内容”。请注意文件读取工具存在安全风险在实际生产环境中必须进行严格的路径校验和权限控制。4. 第6-10天搭建一个高效的RAG知识库系统有了会思考、会使用工具的Agent我们接下来解决它的“知识短板”。RAG系统就是为Agent配备一个随时可查的、精准的私有知识库。4.1 文档加载、切分与向量化全流程RAG的第一步是处理文档。我们设计一个管道加载 - 切分 - 向量化 - 存储。# rag/processor.py from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import TextLoader, PyPDFLoader, UnstructuredFileLoader from langchain.embeddings import OpenAIEmbeddings import chromadb from chromadb.config import Settings from typing import List, Union import hashlib import os class RAGProcessor: def __init__(self, persist_directory: str “./chroma_db”): # 初始化嵌入模型 self.embeddings OpenAIEmbeddings( model“text-embedding-3-small”, openai_api_keyos.getenv(“OPENAI_API_KEY”) ) # 初始化Chroma客户端持久化存储 self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建集合类似数据库的表 self.collection self.client.get_or_create_collection( name“knowledge_base”, metadata{“hnsw:space”: “cosine”} # 使用余弦相似度进行检索 ) # 初始化文本分割器 self.text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个文本块的大小 chunk_overlap50, # 块之间的重叠部分保持上下文连贯 separators[“\n\n”, “\n”, “。”, “.”, “,”, “ “, “”] # 分割符优先级 ) def _generate_id(self, text: str) - str: 为文本块生成唯一ID。 return hashlib.md5(text.encode()).hexdigest() def load_and_split_documents(self, file_path: str) - List[str]: 加载单个文档并分割成文本块。 _, ext os.path.splitext(file_path) loader None if ext.lower() ‘.pdf’: loader PyPDFLoader(file_path) elif ext.lower() in [‘.txt’, ‘.md’, ‘.json’]: loader TextLoader(file_path, encoding‘utf-8’) else: # 尝试用UnstructuredLoader处理其他格式如Word, PPT loader UnstructuredFileLoader(file_path) documents loader.load() # 将所有页面内容合并然后分割 full_text “”.join([doc.page_content for doc in documents]) chunks self.text_splitter.split_text(full_text) return chunks def add_to_knowledge_base(self, file_path: str): 将文档处理并添加到向量数据库。 print(f“正在处理文件{file_path}”) chunks self.load_and_split_documents(file_path) if not chunks: print(“未提取到有效文本内容。”) return # 为每个块生成嵌入向量 embeddings_list self.embeddings.embed_documents(chunks) # 准备批量插入的数据 ids [self._generate_id(chunk) for chunk in chunks] metadatas [{“source”: file_path, “chunk_index”: i} for i in range(len(chunks))] # 插入到Chroma集合 self.collection.add( embeddingsembeddings_list, documentschunks, metadatasmetadatas, idsids ) print(f“成功添加 {len(chunks)} 个文本块到知识库。”) def search(self, query: str, top_k: int 3) - List[str]: 在知识库中检索与查询最相关的文本块。 # 将查询语句向量化 query_embedding self.embeddings.embed_query(query) # 执行相似性搜索 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k ) # 返回检索到的文档内容 return results[‘documents’][0] if results[‘documents’] else []这个RAGProcessor类封装了从文档到向量存储的全过程。关键点在于chunk_size和chunk_overlap的设置太小会丢失上下文太大会引入噪声。500-1000字符是通用文档的常见起点对于技术文档或法律文本可能需要调整。4.2 实现检索与生成融合的RAG问答链有了知识库下一步是构建一个问答链将用户问题、检索到的上下文和系统指令组合发送给LLM生成答案。# rag/query_engine.py from openai import OpenAI import os from .processor import RAGProcessor class RAGQueryEngine: def __init__(self, rag_processor: RAGProcessor, model: str “gpt-3.5-turbo”): self.processor rag_processor self.client OpenAI(api_keyos.getenv(“OPENAI_API_KEY”)) self.model model def query(self, question: str, top_k: int 3) - dict: 执行RAG查询检索 - 生成。 # 1. 检索相关上下文 contexts self.processor.search(question, top_ktop_k) if not contexts: return { “answer”: “知识库中未找到相关信息。”, “sources”: [] } # 2. 构建Prompt指令模型基于上下文回答 context_str “\n\n---\n\n”.join(contexts) system_prompt “””你是一个专业的助手请严格根据提供的上下文信息来回答问题。 如果上下文中的信息不足以回答问题请直接说“根据已知信息无法回答此问题”。 不要编造上下文之外的信息。 上下文信息如下 {context} “””.format(contextcontext_str) # 3. 调用LLM生成答案 response self.client.chat.completions.create( modelself.model, messages[ {“role”: “system”, “content”: system_prompt}, {“role”: “user”, “content”: question} ], temperature0.1 # 低温度让答案更确定、更贴近上下文 ) answer response.choices[0].message.content # 4. 返回答案和来源简化处理实际应返回更详细的元数据 return { “answer”: answer, “sources”: contexts # 实际项目中这里应返回包含源文件、页码等信息的列表 } # 使用示例 if __name__ “__main__”: # 初始化处理器和引擎 processor RAGProcessor() # 假设我们已经通过 processor.add_to_knowledge_base(“某文档.pdf”) 添加了文档 engine RAGQueryEngine(processor) # 进行查询 result engine.query(“LangGraph是什么”) print(“答案”, result[“answer”]) print(“\n参考来源”) for i, src in enumerate(result[“sources”], 1): print(f“[{i}] {src[:150]}...”)这个问答链的核心是系统提示词System Prompt。它明确指令LLM“严格基于上下文回答”这是抑制幻觉的关键。temperature0.1的设置也是为了减少随机性让答案更忠实于检索到的资料。注意事项RAG的效果严重依赖于检索质量。如果检索到的上下文不相关LLM再强也无力回天。常见的优化手段包括查询扩展对原始问题生成多个相关或改写的问题一起检索然后合并结果。重排序使用一个更精细的交叉编码器模型对初步检索到的Top N个结果进行重新打分排序选取最相关的几个。混合检索结合基于关键词的检索如BM25和向量检索取长补短。 在初期确保文档切分合理和嵌入模型合适就能解决80%的问题。5. 第11-15天用LangGraph编排智能体工作流最后五天我们进入高潮用LangGraph将前十天搭建的“原生Agent”和“RAG系统”优雅地组合起来构建一个能自主判断何时该查资料、何时该用工具的超级智能体。5.1 理解LangGraph的核心状态与图LangGraph的核心是两个概念状态State和图Graph。状态一个字典保存了工作流运行中的所有信息比如当前的对话消息、工具调用结果、中间变量等。我们使用MessagesState因为它专为基于消息的对话设计。图由节点Node和边Edge组成。节点是执行具体任务的函数边决定了下一个该执行哪个节点。我们的智能体工作流将包含以下节点Agent节点调用LLM决定下一步行动回答、调用工具、结束。工具执行节点根据Agent的决定执行对应的工具计算器、搜索等。RAG检索节点当Agent需要知识库支持时调用此节点进行检索。路由逻辑根据LLM的输出判断流程走向。5.2 构建智能体工作流图首先定义我们的工作流状态并创建图中需要的各个函数节点。# graph/agent_workflow.py from typing import TypedDict, Annotated, List, Literal import operator from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from openai import OpenAI import os from core.tools import CalculatorTool, WebSearchTool, FileReadTool from rag.query_engine import RAGQueryEngine # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, add_messages] # LangGraph提供的特殊注解用于自动管理消息列表 # 可以添加其他状态如 ‘knowledge’ 用于存储RAG检索结果 knowledge: str # 2. 初始化关键组件 client OpenAI(api_keyos.getenv(“OPENAI_API_KEY”)) # 假设我们已经有了一个初始化好的RAGQueryEngine实例 rag_engine RAGQueryEngine(...) # 将所有工具封装成LangGraph可识别的格式 tools [CalculatorTool, WebSearchTool, FileReadTool] tool_map {tool.name: tool for tool in tools} def _get_tools_schema(): return [tool.get_schema() for tool in tools] # 3. 定义“Agent”节点函数 def call_agent(state: AgentState): 调用LLM决定下一步行动。 messages state[‘messages’] # 检查最近的消息中是否已包含知识库信息 last_few_messages messages[-6:] # 查看最近几条消息 system_message {“role”: “system”, “content”: “你是一个强大的助手可以调用工具或查询知识库来回答问题。”} # 如果状态中包含检索到的知识将其作为系统消息的一部分 if state.get(‘knowledge’): system_message[‘content’] f“\n\n以下是相关的参考信息\n{state[‘knowledge’]}\n请基于这些信息进行回答。” # 构建发送给LLM的消息列表 llm_messages [system_message] [msg for msg in last_few_messages if msg[‘role’] ! ‘system’] response client.chat.completions.create( model“gpt-4-turbo-preview”, # 使用能力更强的模型进行推理 messagesllm_messages, tools_get_tools_schema(), tool_choice“auto”, ) # 将LLM的响应消息添加到状态中 return {“messages”: [response.choices[0].message]} # 4. 定义“工具执行”节点可以使用LangGraph预构建的ToolNode tool_node ToolNode(tools[tool.execute for tool in tools]) # 5. 定义“RAG检索”节点函数 def retrieve_knowledge(state: AgentState): 从RAG知识库中检索信息。 # 从最新的用户消息中提取问题 user_messages [m for m in state[‘messages’] if m[‘role’] ‘user’] if not user_messages: return {“knowledge”: “”} latest_query user_messages[-1][‘content’] # 调用RAG引擎进行检索 result rag_engine.query(latest_query, top_k2) # 将检索到的上下文知识存入状态供下一个Agent节点使用 knowledge_context “\n”.join(result[‘sources’]) return {“knowledge”: knowledge_context} # 6. 定义“路由逻辑”函数 def route_after_agent(state: AgentState) - Literal[“call_tool”, “retrieve”, “end”]: 根据LLM的输出来决定下一步走向。 last_message state[‘messages’][-1] # 如果LLM调用了工具则走向工具执行节点 if last_message.tool_calls: return “call_tool” # 如果LLM的回复中暗示需要更多信息这里用简单关键词判断实际可用更智能的方式 elif “根据已知信息无法回答” in last_message.content or “我需要查询” in last_message.content: return “retrieve” # 否则结束流程 else: return “end”5.3 组装图并运行工作流定义了所有节点和路由函数后现在像搭积木一样把它们组装起来。# 续 graph/agent_workflow.py # 7. 创建图并添加节点 workflow StateGraph(AgentState) workflow.add_node(“agent”, call_agent) workflow.add_node(“tools”, tool_node) workflow.add_node(“retrieve_knowledge”, retrieve_knowledge) # 8. 设置入口点 workflow.set_entry_point(“agent”) # 9. 定义边路由条件 workflow.add_conditional_edges( “agent”, # 源节点 route_after_agent, # 路由判断函数 { “call_tool”: “tools”, # 如果返回“call_tool”则前往“tools”节点 “retrieve”: “retrieve_knowledge”, # 如果返回“retrieve”则前往“retrieve_knowledge”节点 “end”: END # 如果返回“end”则结束流程 } ) # 10. 定义其他边 workflow.add_edge(“tools”, “agent”) # 工具执行完后回到Agent节点继续思考 workflow.add_edge(“retrieve_knowledge”, “agent”) # 检索完知识后回到Agent节点 # 11. 编译图 app workflow.compile() # 12. 运行工作流的函数 def run_agent_workflow(user_input: str): 运行完整的智能体工作流。 # 初始化状态 initial_state: AgentState {“messages”: [{“role”: “user”, “content”: user_input}], “knowledge”: “”} # 运行图 final_state app.invoke(initial_state) # 从最终状态中提取所有消息 all_messages final_state[“messages”] # 找到最后一条来自Assistant的、非工具调用的消息作为最终回复 for msg in reversed(all_messages): if msg[‘role’] ‘assistant’ and not msg.get(‘tool_calls’): return msg[‘content’] return “未生成有效回复。” # 测试 if __name__ “__main__”: while True: query input(“\n请输入您的问题 “) if query.lower() ‘quit’: break answer run_agent_workflow(query) print(f“\n智能体 {answer}”)现在运行这个脚本。当你问一个简单计算题时它会直接调用计算器工具当你问一个知识库里的问题时Agent节点可能先回复“根据已知信息无法回答”触发路由走向“retrieve_knowledge”节点检索到知识后流程回到Agent节点此时Agent的上下文里包含了检索结果它就能生成准确的答案了。整个流程清晰、可控、可调试。5.4 使用FastAPI将智能体服务化最后我们用FastAPI将整个系统包装成一个HTTP API方便集成。# api/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graph.agent_workflow import run_agent_workflow import uvicorn app FastAPI(title“智能体API”, description“集成RAG与工具调用的原生Agent服务”) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str app.post(“/query”, response_modelQueryResponse) async def query_agent(request: QueryRequest): try: answer run_agent_workflow(request.question) return QueryResponse(answeranswer) except Exception as e: raise HTTPException(status_code500, detailf“处理请求时出错{str(e)}”) app.get(“/health”) async def health_check(): return {“status”: “healthy”} if __name__ “__main__”: uvicorn.run(app, host“0.0.0.0”, port8000)运行python api/main.py你的智能体就拥有了一个HTTP接口。你可以用curl、Postman或任何前端应用来调用它。6. 常见问题与排查技巧实录在实际搭建和运行这套系统的过程中你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查清单。6.1 Agent相关LLM不调用工具或调用错误问题现象你问“计算一下35”但Agent直接回答了“35等于8”而没有调用计算器工具。可能原因1工具描述不清。检查CalculatorTool.description是否清晰说明了工具的用途和输入格式。LLM完全依赖这个描述来做决定。可能原因2对话历史干扰。如果历史消息很长且之前有过直接回答的例子LLM可能会模仿。尝试在系统提示词中强调“请优先使用可用工具”。可能原因3模型能力不足。gpt-3.5-turbo的工具调用能力有时不稳定。升级到gpt-4-turbo-preview或gpt-4o会有显著改善。排查技巧打印出每次发送给LLM的完整messages列表和tools参数确认信息传递无误。问题现象工具被调用了但参数解析错误比如{operation: add, numbers: 3,5}数字被传成了字符串。可能原因LLM没有严格按照JSON Schema生成参数。这比较少见但可能发生在复杂参数上。解决方案在工具执行函数execute内部对输入参数进行严格的类型校验和转换并做好异常处理返回友好的错误信息给Agent让它有机会重试。6.2 RAG相关检索结果不相关或答案质量差问题现象检索到的文本片段和问题风马牛不相及。可能原因1文本切分不合理。chunk_size可能太大或太小破坏了语义完整性。对于技术文档可以尝试按章节或标题切分而不是单纯按字符数。可能原因2嵌入模型不匹配。用于生成向量和用于查询的嵌入模型必须一致。检查OpenAIEmbeddings初始化时的model参数。可能原因3查询语句太短或模糊。对于“这是什么”这类模糊查询检索效果很差。可以实施查询重写让LLM将用户问题扩展成更利于检索的多个关键词或完整句子。排查技巧在RAGProcessor.search方法中打印出查询语句的向量和检索到的文本块计算并打印余弦相似度分数Chroma返回结果中包含distances直观感受相关性。问题现象检索到了相关上下文但LLM生成的答案还是胡言乱语。可能原因1Prompt指令不够强。确保系统提示词中有“严格根据上下文”、“不要编造”等强约束语句。可以尝试在Prompt中让模型先引用上下文中的句子再组织答案。可能原因2上下文过长或噪声多。如果检索到的Top K个片段中有不相关的会干扰LLM。减少top_k比如从5降到3或引入重排序模型对初步结果进行筛选。解决方案在RAGQueryEngine的query方法中对检索到的上下文做一个简单的过滤比如只保留与查询语句有至少一个共同关键词的片段。6.3 LangGraph相关图编译错误或状态流转异常问题现象在workflow.compile()时出现Pydantic或类型相关的错误。可能原因AgentState的类型定义与节点函数返回的字典不匹配。确保每个节点函数返回的字典键名都能在AgentState中找到对应且类型兼容。排查技巧简化状态开始时只保留messages字段确保图能跑通再逐步添加其他状态字段。问题现象工作流陷入死循环比如在agent-tools-agent之间无限循环。可能原因路由逻辑route_after_agent有缺陷。例如工具执行后LLM再次决定调用同一个工具。解决方案在路由逻辑中加入“终止条件”。比如记录工具调用次数达到一定次数后强制走向END或者在状态中设置一个max_turns字段记录对话轮数。6.4 性能与成本优化缓存嵌入向量对不变的文档其嵌入向量只需计算一次。ChromaDB在持久化模式下会自动存储但如果你更换了嵌入模型需要重建索引。异步处理FastAPI、OpenAI API客户端都支持异步。将run_agent_workflow中的client.chat.completions.create改为异步调用并使用asyncio.gather并行执行多个独立操作如同时检索多个知识库可以大幅提升API响应速度。控制Token消耗在call_agent函数中我们只取了最近几条消息 (last_few_messages)这就是一种简单的上下文窗口管理策略防止历史对话无限增长消耗大量Token。对于长对话更精细的策略是总结历史对话。备用方案OpenAI API可能不稳定或超时。在生产环境中务必为所有外部API调用OpenAI、搜索工具等添加重试机制和超时设置并考虑配置备用API端点或降级方案如使用本地轻量级LLM。走到这里你已经拥有了一个功能完整、架构清晰的AI智能体原型。它具备了思考、行动和查询知识的能力并且整个流程通过LangGraph变得可视化、可维护。接下来的路就是根据你的具体业务场景去丰富工具集、优化RAG的检索质量、以及打磨工作流的决策逻辑。这个框架的扩展性很好你可以轻松地加入新的工具节点、知识库来源甚至实现多智能体协作。