企业官网建设流程全解析

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 工具与库的精准选择

基于以上拆解，我们的工具选型如下：

1. LLM与Agent基础：OpenAI API。这是事实上的标准，其Function Calling功能是构建Agent的基石。我们将直接使用openai官方Python包。
2. RAG向量数据库：ChromaDB。轻量、易用、纯Python、可持久化，非常适合本地开发和中小型项目。相比Milvus或Pinecone，它无需复杂部署，学习曲线平缓。
3. 嵌入模型：OpenAI的text-embedding-3-small。在效果、速度和成本间取得了很好的平衡。对于完全离线的场景，可以后续替换为BAAI/bge-small-zh-v1.5等开源模型。
4. 工作流编排：LangGraph。它是我们项目的“总导演”，负责调度Agent和RAG。我们将重点学习其StateGraph和MessagesState的概念。
5. Web框架与部署：FastAPI。异步特性好，性能高，自动生成API文档。我们将用它把整个智能体封装成HTTP服务，方便前端或其他系统调用。
6. 开发环境：Python 3.10+，Poetry管理依赖（比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_KEY=sk-your-secret-key-here OPENAI_BASE_URL=https://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_key=os.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( model=self.model, messages=self.conversation_history, tools=self._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( model=self.model, messages=self.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)}&format=json&pretty=1” resp = requests.get(url, timeout=10) 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_key=os.getenv(“OPENAI_API_KEY”) ) # 初始化Chroma客户端，持久化存储 self.client = chromadb.PersistentClient( path=persist_directory, settings=Settings(anonymized_telemetry=False) ) # 获取或创建集合（类似数据库的表） self.collection = self.client.get_or_create_collection( name=“knowledge_base”, metadata={“hnsw:space”: “cosine”} # 使用余弦相似度进行检索 ) # 初始化文本分割器 self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的大小 chunk_overlap=50, # 块之间的重叠部分，保持上下文连贯 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( embeddings=embeddings_list, documents=chunks, metadatas=metadatas, ids=ids ) 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_results=top_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_key=os.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_k=top_k) if not contexts: return { “answer”: “知识库中未找到相关信息。”, “sources”: [] } # 2. 构建Prompt，指令模型基于上下文回答 context_str = “\n\n---\n\n”.join(contexts) system_prompt = “””你是一个专业的助手，请严格根据提供的上下文信息来回答问题。 如果上下文中的信息不足以回答问题，请直接说“根据已知信息无法回答此问题”。 不要编造上下文之外的信息。 上下文信息如下： {context} “””.format(context=context_str) # 3. 调用LLM生成答案 response = self.client.chat.completions.create( model=self.model, messages=[ {“role”: “system”, “content”: system_prompt}, {“role”: “user”, “content”: question} ], temperature=0.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“严格基于上下文回答”，这是抑制幻觉的关键。temperature=0.1的设置也是为了减少随机性，让答案更忠实于检索到的资料。

注意事项：RAG的效果严重依赖于检索质量。如果检索到的上下文不相关，LLM再强也无力回天。常见的优化手段包括：

1. 查询扩展：对原始问题生成多个相关或改写的问题，一起检索，然后合并结果。
2. 重排序：使用一个更精细的交叉编码器模型对初步检索到的Top N个结果进行重新打分排序，选取最相关的几个。
3. 混合检索：结合基于关键词的检索（如BM25）和向量检索，取长补短。 在初期，确保文档切分合理和嵌入模型合适，就能解决80%的问题。

5. 第11-15天：用LangGraph编排智能体工作流

最后五天，我们进入高潮：用LangGraph将前十天搭建的“原生Agent”和“RAG系统”优雅地组合起来，构建一个能自主判断何时该查资料、何时该用工具的超级智能体。

5.1 理解LangGraph的核心：状态与图

LangGraph的核心是两个概念：状态（State）和图（Graph）。

• 状态：一个字典，保存了工作流运行中的所有信息，比如当前的对话消息、工具调用结果、中间变量等。我们使用MessagesState，因为它专为基于消息的对话设计。
• 图：由节点（Node）和边（Edge）组成。节点是执行具体任务的函数，边决定了下一个该执行哪个节点。

我们的智能体工作流将包含以下节点：

1. Agent节点：调用LLM，决定下一步行动（回答、调用工具、结束）。
2. 工具执行节点：根据Agent的决定，执行对应的工具（计算器、搜索等）。
3. RAG检索节点：当Agent需要知识库支持时，调用此节点进行检索。
4. 路由逻辑：根据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_key=os.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”, # 使用能力更强的模型进行推理 messages=llm_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_k=2) # 将检索到的上下文知识存入状态，供下一个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_model=QueryResponse) async def query_agent(request: QueryRequest): try: answer = run_agent_workflow(request.question) return QueryResponse(answer=answer) except Exception as e: raise HTTPException(status_code=500, detail=f“处理请求时出错：{str(e)}”) @app.get(“/health”) async def health_check(): return {“status”: “healthy”} if __name__ == “__main__”: uvicorn.run(app, host=“0.0.0.0”, port=8000)

运行python api/main.py，你的智能体就拥有了一个HTTP接口。你可以用curl、Postman或任何前端应用来调用它。

6. 常见问题与排查技巧实录

在实际搭建和运行这套系统的过程中，你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查清单。

6.1 Agent相关：LLM不调用工具或调用错误

问题现象：你问“计算一下3+5”，但Agent直接回答了“3+5等于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生成的答案还是胡言乱语。

• 可能原因1：Prompt指令不够强。确保系统提示词中有“严格根据上下文”、“不要编造”等强约束语句。可以尝试在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 性能与成本优化

1. 缓存嵌入向量：对不变的文档，其嵌入向量只需计算一次。ChromaDB在持久化模式下会自动存储，但如果你更换了嵌入模型，需要重建索引。
2. 异步处理：FastAPI、OpenAI API客户端都支持异步。将run_agent_workflow中的client.chat.completions.create改为异步调用，并使用asyncio.gather并行执行多个独立操作（如同时检索多个知识库），可以大幅提升API响应速度。
3. 控制Token消耗：在call_agent函数中，我们只取了最近几条消息 (last_few_messages)，这就是一种简单的上下文窗口管理策略，防止历史对话无限增长消耗大量Token。对于长对话，更精细的策略是总结历史对话。
4. 备用方案：OpenAI API可能不稳定或超时。在生产环境中，务必为所有外部API调用（OpenAI、搜索工具等）添加重试机制和超时设置，并考虑配置备用API端点或降级方案（如使用本地轻量级LLM）。

走到这里，你已经拥有了一个功能完整、架构清晰的AI智能体原型。它具备了思考、行动和查询知识的能力，并且整个流程通过LangGraph变得可视化、可维护。接下来的路，就是根据你的具体业务场景，去丰富工具集、优化RAG的检索质量、以及打磨工作流的决策逻辑。这个框架的扩展性很好，你可以轻松地加入新的工具节点、知识库来源，甚至实现多智能体协作。