这次我们来看一个面向企业级应用的AI Agent实战项目。如果你正在寻找一套能快速上手、具备RAG、工具调用、工作流编排等核心能力并能部署到实际业务中的AI智能体解决方案这篇文章会直接带你从零开始搭建一个可运行的AI Agent系统。我们不空谈概念重点放在环境搭建、核心模块实现、效果验证和实际部署上。AI Agent的核心价值在于它能理解复杂指令、规划任务步骤、调用工具并完成目标。对于企业而言这意味着自动化客服、智能数据分析、内部知识问答、自动化流程审批等场景的落地可能。本文将围绕“企业级”这个关键词重点拆解如何构建一个具备RAG知识库、工具调用和LangGraph工作流编排能力的AI Agent。我们会从最基础的环境准备开始一步步实现一个能够回答特定领域问题、执行简单工具调用的智能体并探讨其扩展为复杂工作流的可能性。本文适合有一定Python基础对LangChain、大模型API调用有初步了解希望将AI Agent技术应用于实际业务场景的开发者。我们将重点关注架构的清晰度、代码的可复现性以及部署的可行性。1. 核心能力速览在深入代码之前我们先通过下表快速了解本次实战项目构建的AI Agent所具备的核心能力及其定位。能力项说明项目类型企业级AI Agent原型系统集成RAG、工具调用与工作流。核心组件LangChain框架、LangGraph工作流、向量数据库Chroma、大模型API如OpenAI GPT-4o/Claude 3.5主要功能1.RAG问答基于本地知识库进行精准问答。2.工具调用智能体可调用预设工具如计算器、网络搜索API。3.工作流编排使用LangGraph定义多步骤、带状态的任务执行流程。4.记忆机制支持会话记忆实现多轮对话的上下文连贯。硬件门槛开发阶段对GPU无硬性要求。依赖大模型API云端或本地大模型如Ollama。主要消耗内存和CPU资源。部署方式可基于FastAPI等框架封装为Web API服务支持Docker容器化部署。是否支持API是最终可对外提供RESTful API接口。是否支持批量任务是可通过工作流引擎和任务队列处理批量查询或文档处理任务。适合场景企业内部知识库助手、自动化流程触发、数据查询与报告生成、智能客服原型等。2. 适用场景与使用边界这个实战项目构建的AI Agent系统旨在为企业提供一个可定制、可扩展的智能体开发基础框架。它非常适合以下场景垂直领域知识问答例如将公司产品手册、技术文档、规章制度录入向量数据库构建一个能准确回答内部员工问题的助手。自动化流程助手结合工具调用实现诸如“查询本月销售数据并生成摘要报告”、“监控系统日志并触发告警”等自动化任务。智能客服原型快速搭建一个能理解用户意图、查询知识库、并执行简单操作如创建工单、查询订单状态的客服机器人原型。数据查询与分析通过自然语言与数据库或API交互让非技术人员也能方便地获取业务数据。需要注意的使用边界非开箱即用产品本文提供的是核心框架和实现思路需要根据具体业务需求进行大量的定制化开发包括工具集成、知识库构建、工作流设计等。依赖外部大模型默认方案使用云端大模型API如OpenAI涉及网络调用、费用成本和数据出境合规风险。企业部署需考虑使用合规的国内大模型API或私有化部署的本地模型。知识库质量决定效果RAG的效果严重依赖于文档预处理、分块、向量化等环节的质量。“垃圾进垃圾出”原则在此同样适用。工具调用的安全性智能体能够调用外部工具必须建立严格的权限控制和操作确认机制防止未授权或危险操作。性能与成本频繁调用大模型API会产生费用复杂工作流可能增加响应延迟。在生产环境中需进行性能优化和成本监控。3. 环境准备与前置条件开始编码前请确保你的开发环境满足以下要求。我们将创建一个干净的Python虚拟环境来管理依赖。基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文以Ubuntu为例命令在其他系统上可能略有不同。Python版本 3.9 或 3.10。推荐使用3.10以保证库的最佳兼容性。包管理工具pip(最新版)。版本控制Git (可选但推荐)。关键依赖库我们将使用pip安装一系列核心库。请提前准备好你的大模型API密钥例如OpenAI或国内兼容API。创建并激活虚拟环境# 创建虚拟环境 python -m venv ai_agent_env # 激活虚拟环境 (Linux/macOS) source ai_agent_env/bin/activate # 激活虚拟环境 (Windows) # ai_agent_env\Scripts\activate安装核心依赖在虚拟环境激活状态下执行以下命令。这里列出了最小依赖集后续可根据需要扩展。pip install langchain langchain-community langchain-openai langgraph pip install chromadb pypdf sentence-transformers pip install fastapi uvicorn python-dotenvlangchainlanggraph: AI Agent开发的核心框架。chromadb: 轻量级向量数据库用于存储和检索文档向量。sentence-transformers: 用于生成文本向量的嵌入模型也可使用API。fastapiuvicorn: 用于后续将Agent封装为Web API。python-dotenv: 用于管理环境变量如API密钥。配置环境变量在项目根目录创建.env文件存放你的敏感配置。# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here # 如果使用其他模型例如国内平台 # DASHSCOPE_API_KEYyour-dashscope-key # 向量数据库持久化路径 PERSIST_DIRECTORY./chroma_db重要请勿将.env文件提交到Git等版本控制系统。4. 项目结构与核心模块实现接下来我们搭建项目的基本结构并逐一实现核心模块。一个清晰的结构有助于后续的维护和扩展。ai_agent_project/ ├── .env # 环境变量配置文件 ├── main.py # 主入口FastAPI应用或简单测试 ├── core/ # 核心模块 │ ├── __init__.py │ ├── knowledge_base.py # RAG知识库构建与查询模块 │ ├── tools.py # 自定义工具定义模块 │ └── agent.py # Agent与工作流定义模块 ├── data/ # 存放原始文档PDF, TXT等 │ └── your_document.pdf ├── chroma_db/ # 向量数据库持久化目录自动生成 └── requirements.txt # 依赖列表4.1 构建RAG知识库 (core/knowledge_base.py)RAG是让Agent拥有“长期记忆”和“专业知识”的关键。我们使用ChromaDB和本地嵌入模型来构建。# core/knowledge_base.py import os from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma from dotenv import load_dotenv load_dotenv() class KnowledgeBase: def __init__(self, persist_directory: str None): self.persist_directory persist_directory or os.getenv(PERSIST_DIRECTORY, ./chroma_db) # 使用开源嵌入模型避免API调用 self.embeddings HuggingFaceEmbeddings( model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 ) self.vectorstore None self.text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, length_functionlen, separators[\n\n, \n, 。, , , , , 、, ] ) def load_and_split_documents(self, data_dir: str ./data): 加载并分割文档 documents [] for filename in os.listdir(data_dir): file_path os.path.join(data_dir, filename) if filename.endswith(.pdf): loader PyPDFLoader(file_path) elif filename.endswith(.txt): loader TextLoader(file_path, encodingutf-8) else: continue loaded_docs loader.load() documents.extend(loaded_docs) if not documents: raise ValueError(fNo supported documents found in {data_dir}) print(fLoaded {len(documents)} raw documents.) split_docs self.text_splitter.split_documents(documents) print(fSplit into {len(split_docs)} chunks.) return split_docs def create_vectorstore(self, documents): 创建向量存储 self.vectorstore Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory ) print(fVectorstore created and persisted to {self.persist_directory}) return self.vectorstore def load_existing_vectorstore(self): 加载已存在的向量存储 if os.path.exists(self.persist_directory): self.vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(fLoaded existing vectorstore from {self.persist_directory}) return self.vectorstore else: print(fNo existing vectorstore found at {self.persist_directory}) return None def query(self, question: str, k: int 4): 查询知识库 if not self.vectorstore: self.vectorstore self.load_existing_vectorstore() if not self.vectorstore: raise ValueError(Vectorstore not initialized. Please create or load it first.) docs self.vectorstore.similarity_search(question, kk) context \n\n.join([doc.page_content for doc in docs]) return context, docs # 使用示例 if __name__ __main__: kb KnowledgeBase() # 首次运行构建知识库 # docs kb.load_and_split_documents() # kb.create_vectorstore(docs) # 后续运行直接加载并查询 kb.load_existing_vectorstore() context, relevant_docs kb.query(公司的年假政策是怎样的) print(Retrieved Context:\n, context[:500])4.2 定义智能体工具 (core/tools.py)工具是Agent延伸能力的“手脚”。这里定义两个示例工具一个计算器和一个模拟的网络搜索工具。# core/tools.py from langchain.tools import tool import math import requests tool def calculate(expression: str) - str: 执行数学计算。输入一个数学表达式字符串如 (3 4) * 2返回计算结果。 try: # 警告使用eval存在安全风险仅用于演示。生产环境应使用安全表达式解析器。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。输入一个搜索查询词返回模拟的搜索结果摘要。 # 此处为模拟实际应接入Serper API、Google Search API等 mock_results { 天气预报: 北京晴15-25℃。上海多云18-28℃。, 最新新闻: 模拟新闻AI Agent技术在企业中的应用持续升温。, 股票行情: 模拟数据AAPL 报价 $175.32上涨 1.2%。 } for key, value in mock_results.items(): if key in query: return f搜索 {query} 的结果{value} return f未找到关于 {query} 的模拟信息。您可以尝试‘天气预报’、‘最新新闻’等关键词。 # 工具列表方便导入 available_tools [calculate, search_web]4.3 构建智能体与工作流 (core/agent.py)这是最核心的部分我们将使用LangGraph来定义Agent的思考和工作流程。这里设计一个简单的“思考-行动-观察”循环ReAct模式。# core/agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, AIMessage from langchain.tools.render import render_text_description from dotenv import load_dotenv import os load_dotenv() # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话消息历史 knowledge_context: str # 从知识库检索到的上下文 tool_calls: List # 记录工具调用 # 2. 初始化大模型和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 导入之前定义的工具 from core.tools import available_tools tools available_tools llm_with_tools llm.bind_tools(tools) # 3. 定义各个节点函数 def retrieve_knowledge(state: AgentState): 节点从知识库检索相关信息 from core.knowledge_base import KnowledgeBase # 避免循环导入 kb KnowledgeBase() kb.load_existing_vectorstore() # 取最新的一条用户消息作为查询 last_user_msg None for msg in reversed(state[messages]): if isinstance(msg, HumanMessage): last_user_msg msg.content break if last_user_msg: context, _ kb.query(last_user_msg) state[knowledge_context] context else: state[knowledge_context] print(f[Retrieve] Context retrieved: {state[knowledge_context][:100]}...) return state def decide_action(state: AgentState): 节点决定下一步行动直接回答 or 调用工具 # 构建包含知识和历史的提示词 system_prompt f你是一个有帮助的AI助手。你可以使用工具也可以直接回答问题。 以下是可能相关的知识库信息 {state.get(knowledge_context, 无相关信息)} 请根据对话历史和知识库信息决定是直接回答还是调用工具。 如果你有足够的信息包括知识库信息来直接、准确地回答问题请直接回答。 如果问题涉及计算、实时信息或需要外部工具验证请调用合适的工具。 # 准备给模型的消息 messages_for_llm [{role: system, content: system_prompt}] state[messages] # 让模型进行思考 response llm_with_tools.invoke(messages_for_llm) state[messages].append(response) # 将模型的响应可能包含工具调用加入历史 # 检查响应中是否包含工具调用 if hasattr(response, tool_calls) and response.tool_calls: print(f[Decide] Decision: Need to use tool - {response.tool_calls[0][name]}) return use_tool else: print([Decide] Decision: Answer directly.) return answer_directly def use_tool(state: AgentState): 节点执行工具调用 last_msg state[messages][-1] tool_calls last_msg.tool_calls tool_outputs [] for tool_call in tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据名称找到对应的工具函数 tool_to_use next((t for t in tools if t.name tool_name), None) if tool_to_use: print(f[Tool] Executing {tool_name} with args: {tool_args}) result tool_to_use.invoke(tool_args) tool_outputs.append({call: tool_call, output: result}) else: error_msg fTool {tool_name} not found. print(f[Tool] Error: {error_msg}) tool_outputs.append({call: tool_call, output: error_msg}) # 将工具执行结果作为一条新消息加入历史 state[messages].append(AIMessage(contentstr(tool_outputs), additional_kwargs{tool_calls: tool_calls})) return state def answer_directly(state: AgentState): 节点直接生成最终答案终点 # 最后一个消息已经是模型的直接回答 final_answer state[messages][-1].content print(f[Answer] Final answer: {final_answer[:100]}...) # 这里可以做一些后处理比如格式化 return state # 4. 构建并编译工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_knowledge) workflow.add_node(decide, decide_action) workflow.add_node(use_tool, use_tool) workflow.add_node(answer, answer_directly) # 设置边和条件流 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, decide) # 根据decide节点的输出决定流向 workflow.add_conditional_edges( decide, lambda x: x, # decide节点的返回值就是下一个节点的名称 { use_tool: use_tool, answer_directly: answer } ) workflow.add_edge(use_tool, decide) # 使用工具后重新决策 workflow.add_edge(answer, END) # 直接回答后结束 # 编译图 agent_graph workflow.compile() # 5. 封装一个方便调用的函数 def run_agent_query(user_input: str, chat_history: List None): 运行Agent处理一次用户查询 initial_state { messages: chat_history if chat_history else [], knowledge_context: , tool_calls: [] } initial_state[messages].append(HumanMessage(contentuser_input)) print(f\n 开始处理查询: {user_input} ) final_state agent_graph.invoke(initial_state) print( 处理结束 \n) # 从最终状态中提取AI的最后一条消息作为回复 for msg in reversed(final_state[messages]): if isinstance(msg, AIMessage) and msg.content and not hasattr(msg, tool_calls): return msg.content return 抱歉未能生成有效回复。5. 功能测试与效果验证现在让我们将各个模块组合起来进行端到端的测试。创建一个main.py或使用交互式Python环境。5.1 初始化与知识库构建首先确保你的data/目录下有一些测试文档如PDF或TXT文件。# 测试脚本 test_agent.py import sys sys.path.append(.) from core.knowledge_base import KnowledgeBase from core.agent import run_agent_query # 步骤1构建或加载知识库 print(步骤1: 初始化知识库...) kb KnowledgeBase() # 如果是第一次运行取消下面两行注释来构建知识库 # docs kb.load_and_split_documents(./data) # kb.create_vectorstore(docs) # 后续运行直接加载 kb.load_existing_vectorstore() print(知识库就绪。\n)5.2 测试纯知识库问答测试Agent在不使用工具的情况下仅依靠RAG知识库回答问题。# 步骤2测试知识库检索能力 print(步骤2: 测试知识库检索...) test_question 请总结一下文档中提到的核心内容。 # 替换为你的文档中的真实问题 context, docs kb.query(test_question) print(f问题: {test_question}) print(f检索到的相关片段数: {len(docs)}) print(f首段上下文预览:\n{context[:300]}...\n)5.3 测试智能体完整流程现在测试整合了知识库和工具调用的完整Agent。# 步骤3测试完整Agent问答型 print(步骤3: 测试Agent - 知识型问题...) answer1 run_agent_query(我们公司规定的年假有多少天) # 假设知识库中有此信息 print(fAgent回复: {answer1}\n) # 步骤4测试完整Agent需计算工具 print(步骤4: 测试Agent - 需要计算的问题...) answer2 run_agent_query(如果我有15天年假已经休了7.5天还剩多少天请计算一下。) print(fAgent回复: {answer2}\n) # 步骤5测试完整Agent需搜索工具 print(步骤5: 测试Agent - 需要外部信息的问题...) answer3 run_agent_query(今天北京的天气怎么样) print(fAgent回复: {answer3}\n) # 步骤6测试多轮对话记忆 print(步骤6: 测试Agent - 多轮对话记忆...) # 模拟一个简单的对话历史 from langchain.schema import HumanMessage, AIMessage chat_history [ HumanMessage(content你好我是张三。), AIMessage(content你好张三我是AI助手有什么可以帮您), ] answer4 run_agent_query(我刚刚介绍了自己你还记得我的名字吗, chat_history) print(fAgent回复: {answer4}\n)预期结果与判断标准知识型问题Agent应能返回从你的文档中检索到的准确信息而不是凭空捏造。计算型问题Agent应识别出需要计算调用calculate工具并返回正确的计算结果。搜索型问题Agent应识别出需要外部信息调用search_web工具并返回模拟的搜索结果。多轮对话Agent应能引用对话历史中的信息用户的名字“张三”。控制台输出你应该能看到[Retrieve],[Decide],[Tool],[Answer]等日志清晰展示Agent的思考链路。如果任何一步失败请检查API密钥是否正确设置且有效。知识库是否成功构建并有内容。工具函数定义是否正确名称是否与模型调用匹配。网络连接是否通畅。6. 封装为API服务与批量任务一个企业级系统需要提供标准接口。我们使用FastAPI将Agent封装成Web服务。6.1 创建FastAPI应用# main.py (FastAPI 入口) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uvicorn from core.agent import run_agent_query from langchain.schema import HumanMessage, AIMessage import json app FastAPI(title企业级AI Agent API, description提供RAG问答和工具调用的AI智能体服务) class ChatMessage(BaseModel): role: str # user or assistant content: str class AgentRequest(BaseModel): query: str chat_history: Optional[List[ChatMessage]] None class AgentResponse(BaseModel): answer: str session_id: Optional[str] None # 可用于关联会话 debug_info: Optional[dict] None def convert_to_langchain_messages(history: List[ChatMessage]): 将API格式的历史消息转换为LangChain消息格式 lc_messages [] for msg in history: if msg.role user: lc_messages.append(HumanMessage(contentmsg.content)) elif msg.role assistant: lc_messages.append(AIMessage(contentmsg.content)) return lc_messages app.post(/chat, response_modelAgentResponse) async def chat_with_agent(request: AgentRequest): 与Agent对话的主接口 try: # 转换历史消息 lc_history convert_to_langchain_messages(request.chat_history) if request.chat_history else [] # 调用Agent核心逻辑 answer run_agent_query(request.query, lc_history) return AgentResponse( answeranswer, debug_info{status: success} # 生产环境可返回更多调试信息 ) except Exception as e: raise HTTPException(status_code500, detailfAgent处理失败: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, service: ai_agent} if __name__ __main__: # 启动服务 uvicorn main:app --reload --host 0.0.0.0 --port 8000 uvicorn.run(app, host0.0.0.0, port8000)6.2 启动与测试API服务启动服务cd /path/to/your/ai_agent_project uvicorn main:app --reload --host 0.0.0.0 --port 8000服务启动后访问http://127.0.0.1:8000/docs可以看到自动生成的交互式API文档。使用curl测试APIcurl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d { query: 计算一下(1234)*2等于多少, chat_history: [] }使用Python客户端测试# test_api_client.py import requests import json url http://127.0.0.1:8000/chat payload { query: 我们公司的年假政策是什么, chat_history: [] # 可以传入历史实现多轮对话 } headers {Content-Type: application/json} response requests.post(url, datajson.dumps(payload), headersheaders) print(response.status_code) print(response.json())6.3 实现批量任务处理对于需要处理大量文档或查询的场景可以引入任务队列。# 示例简单的基于列表的批量处理适用于轻量级任务 def batch_process_queries(queries: List[str], output_file: str results.json): 批量处理查询列表 results [] for idx, query in enumerate(queries): print(f处理任务 {idx1}/{len(queries)}: {query}) try: answer run_agent_query(query) results.append({query: query, answer: answer, status: success}) except Exception as e: results.append({query: query, error: str(e), status: failed}) # 保存结果 import json with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f批量处理完成结果已保存至 {output_file}) return results # 生产环境建议使用 Celery Redis/RabbitMQ 或 RQ 等专业任务队列。7. 资源占用与性能观察由于本项目核心依赖云端大模型API和本地轻量级向量数据库资源消耗主要在内存和网络I/O。内存占用主要来自Python进程、LangChain/ LangGraph框架、ChromaDB向量索引以及嵌入模型。在典型开发环境下内存占用可能在500MB到2GB之间具体取决于知识库文档的大小和数量。CPU占用本地句子嵌入模型 (sentence-transformers) 在进行文档编码和查询时会对CPU有一定消耗。API调用阶段CPU占用很低。网络延迟最大的性能变量在于调用大模型API的网络往返时间RTT。国内用户调用海外API可能会有显著延迟建议考虑使用国内合规的模型API或本地模型如通过Ollama部署。向量检索速度ChromaDB在内存中检索速度很快通常能在毫秒级返回结果。性能瓶颈通常不在这一环。优化建议知识库分块优化调整chunk_size和chunk_overlap找到适合你文档类型的最佳大小。使用更快的嵌入模型例如all-MiniLM-L6-v2比paraphrase-multilingual-MiniLM-L12-v2更快更小但精度略有下降。异步处理对于API服务可以使用FastAPI的异步端点 (async def) 和异步的LangChain调用如果支持来提高并发能力。缓存对常见查询的结果进行缓存可以显著减少对模型API的调用。使用本地大模型通过Ollama等工具在本地部署模型如Qwen、Llama等彻底消除网络延迟但需要足够的GPU或CPU内存。8. 常见问题与排查方法在开发和部署过程中你可能会遇到以下问题。下表列出了常见现象、原因及解决方案。问题现象可能原因排查方式解决方案启动服务时报ModuleNotFoundError依赖未安装或虚拟环境未激活。检查当前Python环境执行pip list | grep langchain。激活虚拟环境并运行pip install -r requirements.txt。知识库查询返回空结果1. 文档未成功加载/分割。2. 向量数据库路径错误或为空。3. 查询语句与文档内容语义不匹配。1. 检查data/目录下是否有文件查看加载日志。2. 检查chroma_db/目录大小。3. 尝试更简单、更贴近文档原句的查询。1. 确保文档格式受支持PDF/TXT。2. 重新运行知识库构建流程。3. 调整文本分块策略或尝试不同的嵌入模型。Agent不调用工具总是直接回答1. 工具定义未正确绑定给模型。2. 模型提示词Prompt未明确要求使用工具。3. 问题描述不够清晰模型认为可以直接回答。1. 检查llm.bind_tools(tools)是否执行。2. 查看decide_action节点中的system_prompt。3. 在问题中明确要求“请计算”或“请搜索”。1. 确保工具列表正确传入。2. 强化Prompt明确告知模型在特定场景下必须使用工具。3. 在LangGraph中可以强制在某些节点后必须调用工具。调用大模型API超时或报错1. API密钥无效或余额不足。2. 网络连接问题。3. 模型服务端不稳定。1. 在OpenAI平台检查密钥状态和用量。2. 使用curl或ping测试网络连通性。3. 查看API返回的具体错误信息。1. 更换有效API密钥。2. 配置网络代理或使用国内镜像站如果支持。3. 增加请求超时时间或实现重试机制。FastAPI服务访问127.0.0.1:8000无响应1. 服务未成功启动。2. 端口被占用。3. 防火墙限制。1. 检查终端是否有启动成功的日志。2. 使用netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口。3. 尝试用curl http://127.0.0.1:8000/health测试。1. 确保在项目根目录下启动服务。2. 更换端口如--port 8001。3. 关闭防火墙或添加规则生产环境慎用。多轮对话中Agent忘记历史1.chat_history未正确传递给run_agent_query函数。2. 工作流状态 (AgentState) 中的messages列表未被正确维护。1. 在API测试中检查请求体中的chat_history格式。2. 在run_agent_query函数开始处打印传入的chat_history长度。1. 确保在前后端之间正确序列化和传递历史消息。2. 检查LangGraph工作流图确保messages状态在节点间正确流转。9. 最佳实践与使用建议要将此原型发展为稳健的企业级应用请遵循以下建议分阶段验证不要一开始就追求复杂功能。先确保RAG检索准确再添加工具最后设计复杂工作流。提示词工程decide_action节点的系统提示词是Agent的“大脑”。精心设计它明确知识库的用途、工具的使用条件以及回答的格式要求。这是提升Agent表现最有效的手段之一。工具安全calculate工具中使用了eval这在实际生产环境中是极其危险的。必须替换为安全的表达式解析库如ast.literal_eval或第三方数学库。知识库管理建立知识库的更新、版本控制和回滚机制。可以考虑加入元数据过滤让Agent能根据用户身份或问题类型检索不同范围的文档。日志与监控为Agent的每一步检索、决策、工具调用、最终回答添加详细的结构化日志。这有助于调试和后期分析Agent的行为模式。限流与降级对API服务实施限流防止滥用。当大模型API不可用时应有降级方案例如仅返回知识库检索结果。数据合规如果处理企业敏感数据务必使用私有化部署的大模型和嵌入模型并将所有组件向量数据库、应用服务器部署在内网环境。测试用例为不同的用户查询类型知识问答、计算、搜索、多轮对话编写自动化测试用例确保每次迭代更新不会破坏核心功能。10. 总结与下一步通过本文的实战我们完成了一个具备RAG知识库、工具调用和基础工作流能力的AI Agent系统从零到一的搭建。这个系统的价值在于提供了一个清晰、可扩展的框架而不是一个固定的产品。你可以在此基础上进行深度定制。最值得尝试的扩展方向集成真实工具将模拟的search_web工具替换为真实的搜索引擎API、数据库查询接口、企业内部系统如CRM、OA的API。复杂工作流设计利用LangGraph实现更复杂的业务流程例如“接收用户需求 - 检索知识库 - 生成方案草稿 - 调用工具验证数据 - 发送邮件通知相关人员”。更换底层模型尝试接入国内大模型如通义千问、文心一言、智谱GLM的API或使用Ollama在本地部署开源模型以降低成本和延迟并满足数据合规要求。增强记忆能力实现长期记忆让Agent能够记住跨会话的用户偏好和历史交互摘要。添加前端界面使用Gradio、Streamlit或Vue/React构建一个友好的Web聊天界面让非技术人员也能方便使用。部署与运维使用Docker将整个应用容器化并通过Kubernetes或Docker Compose进行编排实现高可用和弹性伸缩。这个项目的核心代码已经为你扫清了主要的架构障碍。接下来就是将这个“大脑”和“手脚”连接到你具体的业务场景中去。建议从一个小而具体的业务痛点开始用这个框架快速实现一个原型验证价值后再逐步扩展。