在实际企业级应用开发中AI Agent 正从一个前沿概念迅速转变为解决复杂业务流程自动化的核心工具。它不再是简单的聊天机器人而是能够理解上下文、调用工具、处理长期任务、并具备一定自主决策能力的智能工作流引擎。对于开发者而言理解如何从零开始构建一个稳定、可扩展、符合企业安全合规要求的 AI Agent是抓住这一技术红利的关键。本文将围绕一个典型的企业级 AI Agent 核心架构展开涵盖从基础概念到实战落地的全流程重点解析 RAG检索增强生成、工具调用、工作流编排以 LangGraph 为例等核心模块的实现与集成最终构建一个具备长期记忆和自动化决策能力的智能体原型。1. 理解企业级 AI Agent 的核心组件与架构企业级 AI Agent 与简单的对话接口或单次问答模型有本质区别。它需要像一个虚拟员工能够处理多步骤、长周期、依赖外部数据和工具的任务。其核心能力建立在几个关键组件的协同工作之上。1.1 大脑大语言模型与提示工程大语言模型是 Agent 的“大脑”负责理解用户意图、规划任务步骤、生成自然语言回复和决策。在企业级场景中选择模型时不仅要考虑其通用能力更要关注其可控性、成本和对私有化部署的支持。提示工程则是引导这个“大脑”按我们期望方式工作的关键一个结构化的提示词Prompt通常包含角色定义、任务描述、可用工具列表、输出格式约束以及历史对话上下文。1.2 记忆短期、长期与会话记忆记忆系统决定了 Agent 的“经验”和“连续性”。短期记忆通常指单次对话的上下文窗口受模型 Token 长度限制。长期记忆则需要外部存储来维护例如将重要的对话摘要、用户偏好、任务状态持久化到数据库或向量库中供后续会话检索使用。会话记忆则关联特定对话线程确保在同一个任务流程中Agent 能记住之前的步骤和结果。1.3 感知与行动工具调用与 RAGAgent 需要通过“工具”来感知和影响外部世界。工具可以是查询数据库的 API、执行计算的函数、调用第三方服务的接口或是检索内部知识库的 RAG 系统。RAG 通过将用户查询与向量化的企业知识库文档、手册、代码库进行相似性检索并将检索到的相关片段作为上下文提供给大模型从而生成更准确、更具事实依据的答案有效缓解大模型的“幻觉”问题。1.4 协调中枢工作流与状态管理当任务涉及多个步骤、条件分支和循环时就需要一个协调中枢来管理整个流程的状态和流转。这就是工作流引擎或 Agent 框架如 LangGraph、n8n 的工作流模块的作用。它们将 Agent 的思考、工具调用、记忆更新等步骤编排成一个有向图确保复杂任务能够被可靠、可观测地执行。一个典型的企业级 AI Agent 架构可以概括为用户请求首先被工作流引擎接收引擎根据当前状态调用大模型进行“思考”模型决定下一步是调用工具如 RAG 检索、更新记忆还是直接回复。工具执行的结果返回给模型模型据此进行下一步决策直至任务完成。整个过程中的关键状态和决策路径都被记录便于审计和调试。2. 环境准备与核心工具选型在开始编码之前需要搭建一个隔离、可复现的开发环境并选择适合企业级场景的技术栈。这里我们以一个基于 Python 的本地开发环境为例兼顾学习成本和生产可迁移性。2.1 基础开发环境配置首先确保你的开发机具备以下基础环境Python 3.10: 这是当前多数 AI 框架稳定支持的版本。包管理工具: 使用pip或更推荐的poetry、conda来管理依赖避免环境冲突。代码编辑器: VS Code 或 PyCharm并安装 Python 和相关的 AI 扩展。创建一个干净的虚拟环境并初始化项目# 使用 conda (推荐用于管理复杂的科学计算依赖) conda create -n ai-agent-env python3.10 conda activate ai-agent-env # 或使用 venv python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 创建项目目录并初始化 mkdir enterprise-ai-agent cd enterprise-ai-agent touch requirements.txt main.py2.2 核心框架与库选型根据热搜词和当前生态我们选择以下工具链构建原型组件选型说明企业级考量大模型接入OpenAI API / 本地 Ollama初期快速验证用 OpenAI GPT-4/3.5后期考虑成本与安全可切换为本地部署的 Ollama (Llama 3, Qwen 等)。必须支持 API 密钥管理、请求限流、日志审计。本地化部署是最终方向。Agent 框架LangChain LangGraphLangChain 提供了组件化基础LangGraph 专为多步骤、有状态的 Agent 工作流设计是构建复杂 Agent 的理想选择。LangGraph 的状态图StateGraph模型非常适合描述企业审批、客服工单等流程。向量数据库与 RAGChroma (本地) / Qdrant (生产)Chroma 轻量易用适合本地开发和测试。Qdrant、Weaviate 或 Pinecone 更适合生产环境支持分布式和高级过滤。生产环境需考虑向量库的持久化、高可用、备份策略以及与现有数据系统的集成。工具调用LangChain ToolsLangChain 内置了大量工具如搜索、计算也支持轻松自定义工具将内部 API 封装成 Agent 可调用的函数。自定义工具需做好输入验证、错误处理和权限控制避免 Agent 执行危险操作。记忆存储SQLite / PostgreSQL短期会话状态可存于内存或 Redis长期记忆如用户画像、对话摘要建议存入关系型数据库便于查询和管理。需设计合理的数据表结构并考虑 GDPR 等合规要求下的数据留存与删除。工作流可视化/部署LangGraph Studio / n8nLangGraph Studio 可用于本地调试和可视化工作流。n8n 可作为企业级自动化平台集成并部署 Agent 工作流。n8n 等工具提供了用户界面、权限管理、任务队列和监控更适合非研发人员维护。将选定的依赖写入requirements.txtlangchain0.1.0 langchain-openai0.0.5 langgraph0.0.26 chromadb0.4.22 sentence-transformers2.2.2 pydantic2.5.0 sqlite3 # 通常为 Python 内置 # 如需使用 Ollama # ollama0.1.0 # 如需使用 n8n 集成通常通过其 UI 或 REST API 操作使用pip install -r requirements.txt安装依赖。3. 构建核心模块从 RAG 知识库到工具封装在搭建完整 Agent 之前我们需要先实现其核心能力模块一个可靠的 RAG 知识库查询系统和几个可被调用的工具。3.1 创建并加载企业知识库RAG假设我们有一些企业内部的 Markdown 格式文档。首先我们需要将它们处理并存入向量数据库。文档加载与分割将长文档分割成有重叠的小块以保持语义完整性。文本向量化使用嵌入模型将文本块转换为向量。向量存储将向量和原文存入向量数据库。创建一个rag_system.py文件# rag_system.py import os from typing import List from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings # 如果使用本地模型例如 # from langchain_community.embeddings import OllamaEmbeddings class RAGKnowledgeBase: def __init__(self, persist_directory: str ./chroma_db, embedding_modelNone): 初始化 RAG 知识库。 :param persist_directory: 向量数据库持久化目录 :param embedding_model: 嵌入模型默认使用 OpenAI self.persist_directory persist_directory if embedding_model is None: # 使用 OpenAI 嵌入模型需设置环境变量 OPENAI_API_KEY self.embedding_model OpenAIEmbeddings(modeltext-embedding-3-small) # 本地部署可选self.embedding_model OllamaEmbeddings(modelnomic-embed-text) else: self.embedding_model embedding_model self.vector_store None self._init_vector_store() def _init_vector_store(self): 初始化或加载已有的向量存储。 if os.path.exists(self.persist_directory): print(f加载已有向量库: {self.persist_directory}) self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embedding_model ) else: print(未找到已有向量库将创建新库。) # 初始化为空等待添加文档 self.vector_store None def build_from_documents(self, data_dir: str, glob_pattern**/*.md): 从指定目录加载文档并构建向量库。 :param data_dir: 存放文档的目录 :param glob_pattern: 文件匹配模式 # 1. 加载文档 loader DirectoryLoader(data_dir, globglob_pattern, loader_clsTextLoader) documents loader.load() if not documents: print(未加载到任何文档。) return # 2. 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块约1000字符 chunk_overlap200, # 块间重叠200字符以保持上下文 separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f已将 {len(documents)} 个文档分割为 {len(splits)} 个文本块。) # 3. 创建向量存储 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embedding_model, persist_directoryself.persist_directory ) print(f向量库已构建并保存至: {self.persist_directory}) def query(self, question: str, k: int 4) - List[str]: 查询知识库返回最相关的 k 个文本片段。 :param question: 用户问题 :param k: 返回的相似文本数量 :return: 相关文本列表 if self.vector_store is None: raise ValueError(向量库未初始化请先构建或加载知识库。) # 执行相似性搜索 docs self.vector_store.similarity_search(question, kk) # 提取文本内容 contexts [doc.page_content for doc in docs] return contexts # 使用示例 if __name__ __main__: # 假设你的企业文档放在 ./data 目录下 rag RAGKnowledgeBase() # 首次运行构建知识库 # rag.build_from_documents(./data) # 后续运行直接加载并查询 contexts rag.query(我们公司的年假政策是怎样的) for i, ctx in enumerate(contexts): print(f[片段 {i1}]: {ctx[:200]}...) # 打印前200字符关键点解释RecursiveCharacterTextSplitter能根据字符递归分割尽量保持段落和句子的完整性。chunk_overlap设置重叠可以避免一个完整的句子或概念被割裂到两个块中。Chroma持久化后下次启动可直接加载无需重复处理文档。生产环境中文档更新需要设计增量更新或重建索引的策略。3.2 封装自定义工具工具是 Agent 的手和脚。下面封装两个工具一个用于查询上述 RAG 知识库另一个用于执行简单的计算模拟一个业务 API。创建一个custom_tools.py文件# custom_tools.py from typing import Type, Optional from langchain.tools import BaseTool, StructuredTool, tool from pydantic import BaseModel, Field import requests import json # --- 工具1: RAG 查询工具 --- class RAGQueryInput(BaseModel): RAG 查询工具的输入模型。 question: str Field(description需要查询企业知识库的问题) class RAGQueryTool(BaseTool): name query_company_knowledge_base description 当用户询问关于公司政策、产品手册、规章制度等内部知识时使用此工具进行查询。 args_schema: Type[BaseModel] RAGQueryInput def __init__(self, rag_system): super().__init__() self.rag_system rag_system def _run(self, question: str) - str: 执行工具调用。 try: contexts self.rag_system.query(question, k3) if not contexts: return 在知识库中未找到相关信息。 # 将检索到的上下文拼接成一个字符串供大模型参考 combined_context \n\n--- 参考知识 ---\n \n\n.join(contexts) return combined_context except Exception as e: return f查询知识库时发生错误: {str(e)} # --- 工具2: 模拟业务 API 调用工具 --- class BusinessAPIInput(BaseModel): 业务API调用工具的输入模型。 employee_id: str Field(description员工工号) action: str Field(description要执行的操作例如get_annual_leave_balance查询年假余额) tool(args_schemaBusinessAPIInput) def call_business_api(employee_id: str, action: str) - str: 调用内部业务系统API查询或处理与员工相关的数据。 这是一个模拟工具实际项目中应替换为真实的 API 调用。 # 模拟一个内部 API 的响应 mock_responses { get_annual_leave_balance: {balance: 10, unit: 天}, get_salary_info: {basic_salary: 15000, currency: CNY}, } if action not in mock_responses: return json.dumps({error: f不支持的操作: {action}}) # 模拟根据员工ID进行逻辑处理 result mock_responses[action] result[employee_id] employee_id result[status] success return json.dumps(result, ensure_asciiFalse) # --- 工具3: 通用计算器工具使用 LangChain 内置工具示例--- # 可以直接从 langchain 导入这里展示如何声明 from langchain.tools import Tool from langchain_community.utilities import ArxivAPIWrapper arxiv ArxivAPIWrapper() arxiv_tool Tool( namearxiv_search, description用于搜索 Arxiv 学术论文。当用户需要查询最新的研究论文时使用。, funcarxiv.run ) # 工具集合 def get_tools(rag_system): 获取所有可用工具的列表。 return [ RAGQueryTool(rag_systemrag_system), call_business_api, arxiv_tool, # 这是一个外部知识工具仅作示例 ]关键点解释每个工具都需要清晰的name和description这直接决定了大模型是否会以及如何调用它。描述要具体说明何时使用。使用 Pydantic 模型 (BaseModel) 定义输入参数可以强制类型检查并为大模型提供清晰的参数结构。在_run方法或函数内部必须做好异常处理并返回字符串结果。大模型将接收这个结果作为下一步推理的依据。工具可以串联使用例如先通过 RAG 工具查到政策再通过业务 API 工具查询个人数据。4. 使用 LangGraph 编排企业级 AI Agent 工作流有了核心模块现在我们可以用 LangGraph 来编排一个具备状态管理和长期记忆的复杂 Agent。我们将构建一个“员工自助服务 Agent”它可以处理混合了知识问答和业务操作的复杂请求。4.1 定义 Agent 状态与节点LangGraph 的核心是状态图。我们首先定义整个工作流需要维护的状态然后创建不同的节点函数来操作这个状态。创建一个agent_workflow.py文件# agent_workflow.py from typing import TypedDict, Annotated, List, Union from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage import operator from rag_system import RAGKnowledgeBase from custom_tools import get_tools from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.runnables import RunnablePassthrough # --- 1. 定义状态结构 --- class AgentState(TypedDict): 定义 Agent 工作流的状态。 这是一个有类型的字典描述了图中每个节点可以读写的数据。 # 消息列表存储整个对话历史用于模型上下文 messages: Annotated[List[Union[HumanMessage, AIMessage, SystemMessage]], add_messages] # 用户的最新输入 user_input: str # 从知识库检索到的上下文 knowledge_context: str # 工具调用的结果 tool_outputs: List[str] # 最终给用户的回复 final_answer: str # --- 2. 初始化核心组件 --- # 初始化大模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 使用 gpt-4 以获得更好的工具调用能力 # 初始化 RAG rag RAGKnowledgeBase(persist_directory./chroma_db) # 假设已构建好知识库 # 获取工具集 tools get_tools(rag) # 创建 LangChain Agent (用于工具调用决策) agent_runnable create_tool_calling_agent(llm, tools) agent_executor AgentExecutor(agentagent_runnable, toolstools, verboseTrue) # --- 3. 定义各个节点Node--- def receive_input(state: AgentState) - AgentState: 节点接收用户输入并初始化状态。 # 在实际应用中这里可能从 API 请求中获取 user_input # 我们假设 state[user_input] 已被传入 new_message HumanMessage(contentstate[user_input]) # 将用户消息添加到历史中 state[messages].append(new_message) # 清空临时字段 state[knowledge_context] state[tool_outputs] [] state[final_answer] print(f[节点: receive_input] 收到用户输入: {state[user_input][:50]}...) return state def decide_action(state: AgentState) - AgentState: 节点分析用户意图决定下一步是直接回答、查询知识库还是调用业务工具。 这是一个简化的路由决策。 last_message state[messages][-1].content.lower() # 简单的关键词路由逻辑。在实际项目中可以使用一个更复杂的分类模型或提示词。 if any(keyword in last_message for keyword in [政策, 手册, 如何, 是什么, 为什么]): state[next_action] query_knowledge elif any(keyword in last_message for keyword in [余额, 工资, 申请, 提交, 工号]): state[next_action] call_tool else: state[next_action] direct_chat print(f[节点: decide_action] 决策下一步动作为: {state[next_action]}) return state def query_knowledge_node(state: AgentState) - AgentState: 节点调用 RAG 工具查询知识库。 question state[user_input] try: # 这里直接复用之前封装的 RAG 查询功能 contexts rag.query(question, k3) knowledge_context \n\n--- 参考知识 ---\n \n\n.join(contexts) if contexts else 未找到相关信息。 state[knowledge_context] knowledge_context # 将检索到的知识也作为一条“系统”消息加入历史帮助模型理解 state[messages].append(SystemMessage(contentf相关背景知识{knowledge_context})) except Exception as e: state[knowledge_context] f知识库查询失败{str(e)} print(f[节点: query_knowledge] 已检索知识上下文。) return state def call_tool_node(state: AgentState) - AgentState: 节点执行工具调用。使用 LangChain AgentExecutor 来处理。 # 将最新的对话历史包含可能的系统消息传给 Agent inputs {input: state[messages], chat_history: state[messages][:-1]} # 最后一条是当前用户输入 try: result agent_executor.invoke(inputs) # result 是一个字典其中 output 键包含模型的最终回复 tool_output result.get(output, 工具执行完成但无明确输出。) # 记录工具调用过程中的中间输出如果 executor 的 verboseTrue会在控制台看到 state[tool_outputs].append(tool_output) # 将工具执行的结果也作为一条 AIMessage 加入历史 state[messages].append(AIMessage(contenttool_output)) except Exception as e: error_msg f工具调用过程中出错{str(e)} state[tool_outputs].append(error_msg) state[messages].append(AIMessage(contenterror_msg)) print(f[节点: call_tool] 工具调用完成。) return state def generate_final_answer(state: AgentState) - AgentState: 节点综合所有信息生成最终回复给用户。 # 准备给模型的提示词整合历史、知识、工具结果 prompt_messages state[messages] # 历史消息已包含所有上下文 try: response llm.invoke(prompt_messages) final_answer response.content state[final_answer] final_answer # 将最终回复也加入消息历史 state[messages].append(AIMessage(contentfinal_answer)) except Exception as e: state[final_answer] f生成回复时出错{str(e)} print(f[节点: generate_final_answer] 已生成最终回复。) return state # --- 4. 构建工作流图 --- def create_agent_graph(): 创建并编译 LangGraph 工作流。 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(receive_input, receive_input) workflow.add_node(decide_action, decide_action) workflow.add_node(query_knowledge, query_knowledge_node) workflow.add_node(call_tool, call_tool_node) workflow.add_node(generate_answer, generate_final_answer) # 设置入口点 workflow.set_entry_point(receive_input) # 定义边路由逻辑 workflow.add_edge(receive_input, decide_action) # 根据决策结果路由到不同分支 def route_after_decision(state): next_action state.get(next_action, direct_chat) if next_action query_knowledge: return query_knowledge elif next_action call_tool: return call_tool else: # 直接聊天跳过工具和知识库节点 return generate_answer workflow.add_conditional_edges( decide_action, route_after_decision, { query_knowledge: query_knowledge, call_tool: call_tool, generate_answer: generate_answer, } ) # 定义节点执行后的流向 workflow.add_edge(query_knowledge, generate_answer) workflow.add_edge(call_tool, generate_answer) workflow.add_edge(generate_answer, END) # 编译图 graph workflow.compile() return graph # 创建图实例 agent_graph create_agent_graph()关键点解释AgentState是一个强类型的状态容器它清晰地定义了工作流中流动的数据。Annotated和add_messages用于自动管理消息列表的合并。每个节点都是一个纯函数接收状态修改状态返回新状态。这使逻辑清晰且易于测试。decide_action节点实现了简单的路由逻辑。更复杂的场景可以使用一个小型分类模型或让大模型本身来决定下一步。add_conditional_edges是 LangGraph 的核心特性它允许根据状态动态决定下一步走向实现了复杂的业务流程如审批流、循环检查。最终编译的graph是一个可执行对象我们可以传入初始状态来运行整个工作流。4.2 运行与测试 Agent创建一个main.py文件来运行和测试我们构建的 Agent# main.py from agent_workflow import agent_graph, AgentState from langchain_core.messages import HumanMessage def run_agent_interactive(): 以交互方式运行 Agent。 print( 企业级 AI Agent 测试 ) print(输入 quit 或 exit 退出。) # 初始化一个空的对话历史 initial_state: AgentState { messages: [], # 初始为空 user_input: , knowledge_context: , tool_outputs: [], final_answer: } while True: try: user_input input(\n 用户: ).strip() if user_input.lower() in [quit, exit]: print(再见) break if not user_input: continue # 准备本次调用的输入状态 current_state initial_state.copy() current_state[user_input] user_input # 保留上一次的对话历史实现多轮对话 current_state[messages] initial_state[messages].copy() print(Agent 正在思考...) # 执行工作流图 final_state agent_graph.invoke(current_state) # 输出最终回复 print(f\n Agent: {final_state[final_answer]}) # 更新全局历史状态为下一轮对话准备 initial_state[messages] final_state[messages] # 可选打印一些调试信息 # if final_state.get(knowledge_context): # print(f[调试] 检索到的知识: {final_state[knowledge_context][:200]}...) # if final_state.get(tool_outputs): # print(f[调试] 工具输出: {final_state[tool_outputs]}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f运行过程中出现错误: {e}) if __name__ __main__: # 确保你的 OpenAI API Key 已设置环境变量 OPENAI_API_KEY # 确保你的知识库已构建运行过一次 rag_system.py 中的 build_from_documents run_agent_interactive()运行python main.py你可以测试不同类型的查询知识类“公司的年假政策是什么” - 触发query_knowledge节点。业务操作类“查询工号 E001 的年假余额。” - 触发call_tool节点模拟调用业务 API。普通聊天“你好介绍一下你自己。” - 触发direct_chat路径直接由大模型回答。5. 企业级考量安全、合规、监控与部署一个能在企业内部落地的 AI Agent除了核心功能还必须解决安全、合规和运维问题。5.1 安全与权限控制工具调用沙箱化确保工具尤其是写操作、系统命令、网络请求在受控环境中运行限制其资源访问权限如文件系统、网络。输入输出过滤与审查对用户的输入和模型的输出进行内容安全过滤防止注入攻击、敏感信息泄露或生成不当内容。基于角色的访问控制将 Agent 能力与公司现有的权限系统集成。例如只有 HR 部门的 Agent 才能调用查询全员薪资的工具。这可以在decide_action节点或工具内部实现。API 密钥与凭证管理切勿在代码中硬编码密钥。使用环境变量、密钥管理服务如 AWS Secrets Manager, HashiCorp Vault或云厂商提供的托管服务。5.2 合规与审计数据隐私确保 RAG 知识库中的文档不包含个人隐私信息PII。必要时使用数据脱敏工具。在欧盟 GDPR 或中国个人信息保护法框架下需明确告知用户数据使用方式。可解释性与审计日志记录 Agent 的完整决策链路包括用户原始输入。模型每一步的思考如果使用 ReAct 等模式。调用了哪些工具输入输出是什么。最终回复。 这些日志应存入安全的、不可篡改的存储中供事后审计。人工审核介入对于高风险操作如审批、支付、重要数据修改设计“人在环路”机制。LangGraph 的HumanInTheLoop功能可以暂停工作流等待人工确认后再继续。5.3 监控、性能与可观测性关键指标监控延迟用户查询到收到回复的总时间及各节点耗时。成本大模型 API 调用 Token 消耗、工具调用次数。成功率任务完成率、工具调用失败率。用户满意度可通过简单的“是否解决”反馈按钮收集。链路追踪使用 OpenTelemetry 等标准对工作流的每个节点进行埋点在 Jaeger 或 Grafana Tempo 中可视化整个调用链便于排查性能瓶颈和错误。优雅降级与熔断当大模型 API 或关键工具服务不可用时Agent 应有备用方案例如返回缓存答案、提示用户稍后再试或转接人工客服。5.4 部署与扩展容器化使用 Docker 将 Agent 及其依赖Python 环境、模型文件等打包确保环境一致性。编排与伸缩使用 Kubernetes 或 Docker Swarm 管理容器化后的 Agent 服务根据负载自动伸缩。API 网关通过 API 网关如 Kong, APISIX对外暴露 Agent 服务统一处理认证、限流、日志和监控。与现有系统集成身份认证集成公司的 SSO单点登录系统。消息通道将 Agent 部署为企业微信、钉钉、Slack 或 Teams 的机器人。业务流程通过 n8n、Airflow 或 Camunda 等平台将 Agent 作为自动化流程中的一个智能节点调用。6. 常见问题排查与优化实践在开发和运行 AI Agent 过程中你会遇到一些典型问题。以下是排查清单和优化建议。6.1 问题排查清单问题现象可能原因检查步骤解决方案Agent 不调用工具总是直接回答。1. 工具描述不清晰。2. 模型温度temperature过高创造性太强。3. 提示词未明确要求使用工具。1. 检查工具的name和description是否准确描述了其功能和适用场景。2. 将模型temperature设为 0 或接近 0 的值使其更确定性。3. 在给模型的系统提示词中明确指令“你必须使用提供的工具来回答问题”。优化工具描述使其更具体。在系统提示词中强化工具使用规则。使用gpt-4-turbo等工具调用能力更强的模型。RAG 检索结果不相关。1. 文档分割策略不佳。2. 嵌入模型不匹配。3. 检索 top-k 值不合适。1. 检查分割后的文本块是否语义完整。2. 尝试不同的嵌入模型如text-embedding-3-large。3. 调整similarity_search的k参数或尝试max_marginal_relevance_search提升多样性。调整文本分割器的chunk_size和chunk_overlap。对查询进行重写或扩展。引入元数据过滤如按文档类型检索。工作流状态混乱或丢失。1. 状态对象在节点间被意外覆盖。2. 多线程/异步调用导致状态竞争。1. 检查每个节点函数是否都正确返回了完整的state字典。2. 确保AgentState中Annotated字段的归约函数如add_messages使用正确。遵循 LangGraph 最佳实践将状态修改限制在节点函数内。对于持久化状态使用数据库支持的状态后端。多轮对话中Agent 忘记之前的内容。1. 对话历史未正确传递给模型。2. 上下文长度超限历史被截断。1. 检查state[‘messages’]列表是否包含了所有历史消息。2. 计算已使用的 Token 数确认是否超限。确保每轮对话都将更新后的messages传回图中。对于长对话实现摘要机制将过长的历史压缩成摘要存入长期记忆。工具调用超时或失败。1. 工具依赖的外部 API 不可用。2. 网络或权限问题。3. 工具函数内部有 bug。1. 直接调用工具函数检查其独立运行是否正常。2. 查看工具函数的错误日志和异常捕获。3. 检查网络连接和 API 密钥。为工具调用添加超时和重试机制。实现完善的错误处理返回清晰的错误信息供模型和用户理解。6.2 性能与效果优化实践提示词工程优化系统提示词明确 Agent 的角色、职责、约束和输出格式。例如“你是一个严谨的员工自助助手必须基于公司知识库和可用工具来回答问题不得编造信息。”思维链对于复杂问题在提示词中要求模型“逐步思考”这能提升其推理和工具调用的准确性。RAG 优化查询重写在检索前先用一个小模型将用户查询重写为更利于检索的形式如关键词提取、问题分解、陈述句转换。混合检索结合向量检索语义和关键词检索BM25取长补短。重排序初步检索出较多结果如 k20后用一个更精细的交叉编码器模型对结果进行重排序选取最相关的几个。记忆优化对话摘要当对话轮数增多时将早期对话总结成一段摘要替换掉原始的长篇历史以节省 Token 并保留核心信息。向量记忆将重要的对话片段或事实存入另一个向量库当后续对话涉及相关话题时主动检索出来作为上下文。成本优化模型分级简单的意图识别或路由使用小模型如 GPT-3.5-turbo复杂的推理和生成再用大模型如 GPT-4。缓存对常见的、结果不变的查询如“公司地址”将 RAG 结果或模型回复缓存起来。Token 管理监控并设置 Token 使用上限防止异常输入导致巨额费用。从零构建企业级 AI Agent 是一个系统工程涉及模型、框架、数据、工具和运维多个层面。本文提供了一个以 LangGraph 和 RAG 为核心的可运行原型和完整的设计思路。真正的挑战在于将这套原型与具体的企业环境、业务流程和安全规范深度融合。建议从一个小而具体的业务场景开始试点例如 IT 内部问答或新员工入职引导在迭代中不断完善其可靠性、安全性和用户体验逐步扩展其能力边界。 从零构建企业级AI Agent：基于LangGraph与RAG的智能工作流实战
在实际企业级应用开发中AI Agent 正从一个前沿概念迅速转变为解决复杂业务流程自动化的核心工具。它不再是简单的聊天机器人而是能够理解上下文、调用工具、处理长期任务、并具备一定自主决策能力的智能工作流引擎。对于开发者而言理解如何从零开始构建一个稳定、可扩展、符合企业安全合规要求的 AI Agent是抓住这一技术红利的关键。本文将围绕一个典型的企业级 AI Agent 核心架构展开涵盖从基础概念到实战落地的全流程重点解析 RAG检索增强生成、工具调用、工作流编排以 LangGraph 为例等核心模块的实现与集成最终构建一个具备长期记忆和自动化决策能力的智能体原型。1. 理解企业级 AI Agent 的核心组件与架构企业级 AI Agent 与简单的对话接口或单次问答模型有本质区别。它需要像一个虚拟员工能够处理多步骤、长周期、依赖外部数据和工具的任务。其核心能力建立在几个关键组件的协同工作之上。1.1 大脑大语言模型与提示工程大语言模型是 Agent 的“大脑”负责理解用户意图、规划任务步骤、生成自然语言回复和决策。在企业级场景中选择模型时不仅要考虑其通用能力更要关注其可控性、成本和对私有化部署的支持。提示工程则是引导这个“大脑”按我们期望方式工作的关键一个结构化的提示词Prompt通常包含角色定义、任务描述、可用工具列表、输出格式约束以及历史对话上下文。1.2 记忆短期、长期与会话记忆记忆系统决定了 Agent 的“经验”和“连续性”。短期记忆通常指单次对话的上下文窗口受模型 Token 长度限制。长期记忆则需要外部存储来维护例如将重要的对话摘要、用户偏好、任务状态持久化到数据库或向量库中供后续会话检索使用。会话记忆则关联特定对话线程确保在同一个任务流程中Agent 能记住之前的步骤和结果。1.3 感知与行动工具调用与 RAGAgent 需要通过“工具”来感知和影响外部世界。工具可以是查询数据库的 API、执行计算的函数、调用第三方服务的接口或是检索内部知识库的 RAG 系统。RAG 通过将用户查询与向量化的企业知识库文档、手册、代码库进行相似性检索并将检索到的相关片段作为上下文提供给大模型从而生成更准确、更具事实依据的答案有效缓解大模型的“幻觉”问题。1.4 协调中枢工作流与状态管理当任务涉及多个步骤、条件分支和循环时就需要一个协调中枢来管理整个流程的状态和流转。这就是工作流引擎或 Agent 框架如 LangGraph、n8n 的工作流模块的作用。它们将 Agent 的思考、工具调用、记忆更新等步骤编排成一个有向图确保复杂任务能够被可靠、可观测地执行。一个典型的企业级 AI Agent 架构可以概括为用户请求首先被工作流引擎接收引擎根据当前状态调用大模型进行“思考”模型决定下一步是调用工具如 RAG 检索、更新记忆还是直接回复。工具执行的结果返回给模型模型据此进行下一步决策直至任务完成。整个过程中的关键状态和决策路径都被记录便于审计和调试。2. 环境准备与核心工具选型在开始编码之前需要搭建一个隔离、可复现的开发环境并选择适合企业级场景的技术栈。这里我们以一个基于 Python 的本地开发环境为例兼顾学习成本和生产可迁移性。2.1 基础开发环境配置首先确保你的开发机具备以下基础环境Python 3.10: 这是当前多数 AI 框架稳定支持的版本。包管理工具: 使用pip或更推荐的poetry、conda来管理依赖避免环境冲突。代码编辑器: VS Code 或 PyCharm并安装 Python 和相关的 AI 扩展。创建一个干净的虚拟环境并初始化项目# 使用 conda (推荐用于管理复杂的科学计算依赖) conda create -n ai-agent-env python3.10 conda activate ai-agent-env # 或使用 venv python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 创建项目目录并初始化 mkdir enterprise-ai-agent cd enterprise-ai-agent touch requirements.txt main.py2.2 核心框架与库选型根据热搜词和当前生态我们选择以下工具链构建原型组件选型说明企业级考量大模型接入OpenAI API / 本地 Ollama初期快速验证用 OpenAI GPT-4/3.5后期考虑成本与安全可切换为本地部署的 Ollama (Llama 3, Qwen 等)。必须支持 API 密钥管理、请求限流、日志审计。本地化部署是最终方向。Agent 框架LangChain LangGraphLangChain 提供了组件化基础LangGraph 专为多步骤、有状态的 Agent 工作流设计是构建复杂 Agent 的理想选择。LangGraph 的状态图StateGraph模型非常适合描述企业审批、客服工单等流程。向量数据库与 RAGChroma (本地) / Qdrant (生产)Chroma 轻量易用适合本地开发和测试。Qdrant、Weaviate 或 Pinecone 更适合生产环境支持分布式和高级过滤。生产环境需考虑向量库的持久化、高可用、备份策略以及与现有数据系统的集成。工具调用LangChain ToolsLangChain 内置了大量工具如搜索、计算也支持轻松自定义工具将内部 API 封装成 Agent 可调用的函数。自定义工具需做好输入验证、错误处理和权限控制避免 Agent 执行危险操作。记忆存储SQLite / PostgreSQL短期会话状态可存于内存或 Redis长期记忆如用户画像、对话摘要建议存入关系型数据库便于查询和管理。需设计合理的数据表结构并考虑 GDPR 等合规要求下的数据留存与删除。工作流可视化/部署LangGraph Studio / n8nLangGraph Studio 可用于本地调试和可视化工作流。n8n 可作为企业级自动化平台集成并部署 Agent 工作流。n8n 等工具提供了用户界面、权限管理、任务队列和监控更适合非研发人员维护。将选定的依赖写入requirements.txtlangchain0.1.0 langchain-openai0.0.5 langgraph0.0.26 chromadb0.4.22 sentence-transformers2.2.2 pydantic2.5.0 sqlite3 # 通常为 Python 内置 # 如需使用 Ollama # ollama0.1.0 # 如需使用 n8n 集成通常通过其 UI 或 REST API 操作使用pip install -r requirements.txt安装依赖。3. 构建核心模块从 RAG 知识库到工具封装在搭建完整 Agent 之前我们需要先实现其核心能力模块一个可靠的 RAG 知识库查询系统和几个可被调用的工具。3.1 创建并加载企业知识库RAG假设我们有一些企业内部的 Markdown 格式文档。首先我们需要将它们处理并存入向量数据库。文档加载与分割将长文档分割成有重叠的小块以保持语义完整性。文本向量化使用嵌入模型将文本块转换为向量。向量存储将向量和原文存入向量数据库。创建一个rag_system.py文件# rag_system.py import os from typing import List from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings # 如果使用本地模型例如 # from langchain_community.embeddings import OllamaEmbeddings class RAGKnowledgeBase: def __init__(self, persist_directory: str ./chroma_db, embedding_modelNone): 初始化 RAG 知识库。 :param persist_directory: 向量数据库持久化目录 :param embedding_model: 嵌入模型默认使用 OpenAI self.persist_directory persist_directory if embedding_model is None: # 使用 OpenAI 嵌入模型需设置环境变量 OPENAI_API_KEY self.embedding_model OpenAIEmbeddings(modeltext-embedding-3-small) # 本地部署可选self.embedding_model OllamaEmbeddings(modelnomic-embed-text) else: self.embedding_model embedding_model self.vector_store None self._init_vector_store() def _init_vector_store(self): 初始化或加载已有的向量存储。 if os.path.exists(self.persist_directory): print(f加载已有向量库: {self.persist_directory}) self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embedding_model ) else: print(未找到已有向量库将创建新库。) # 初始化为空等待添加文档 self.vector_store None def build_from_documents(self, data_dir: str, glob_pattern**/*.md): 从指定目录加载文档并构建向量库。 :param data_dir: 存放文档的目录 :param glob_pattern: 文件匹配模式 # 1. 加载文档 loader DirectoryLoader(data_dir, globglob_pattern, loader_clsTextLoader) documents loader.load() if not documents: print(未加载到任何文档。) return # 2. 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块约1000字符 chunk_overlap200, # 块间重叠200字符以保持上下文 separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f已将 {len(documents)} 个文档分割为 {len(splits)} 个文本块。) # 3. 创建向量存储 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embedding_model, persist_directoryself.persist_directory ) print(f向量库已构建并保存至: {self.persist_directory}) def query(self, question: str, k: int 4) - List[str]: 查询知识库返回最相关的 k 个文本片段。 :param question: 用户问题 :param k: 返回的相似文本数量 :return: 相关文本列表 if self.vector_store is None: raise ValueError(向量库未初始化请先构建或加载知识库。) # 执行相似性搜索 docs self.vector_store.similarity_search(question, kk) # 提取文本内容 contexts [doc.page_content for doc in docs] return contexts # 使用示例 if __name__ __main__: # 假设你的企业文档放在 ./data 目录下 rag RAGKnowledgeBase() # 首次运行构建知识库 # rag.build_from_documents(./data) # 后续运行直接加载并查询 contexts rag.query(我们公司的年假政策是怎样的) for i, ctx in enumerate(contexts): print(f[片段 {i1}]: {ctx[:200]}...) # 打印前200字符关键点解释RecursiveCharacterTextSplitter能根据字符递归分割尽量保持段落和句子的完整性。chunk_overlap设置重叠可以避免一个完整的句子或概念被割裂到两个块中。Chroma持久化后下次启动可直接加载无需重复处理文档。生产环境中文档更新需要设计增量更新或重建索引的策略。3.2 封装自定义工具工具是 Agent 的手和脚。下面封装两个工具一个用于查询上述 RAG 知识库另一个用于执行简单的计算模拟一个业务 API。创建一个custom_tools.py文件# custom_tools.py from typing import Type, Optional from langchain.tools import BaseTool, StructuredTool, tool from pydantic import BaseModel, Field import requests import json # --- 工具1: RAG 查询工具 --- class RAGQueryInput(BaseModel): RAG 查询工具的输入模型。 question: str Field(description需要查询企业知识库的问题) class RAGQueryTool(BaseTool): name query_company_knowledge_base description 当用户询问关于公司政策、产品手册、规章制度等内部知识时使用此工具进行查询。 args_schema: Type[BaseModel] RAGQueryInput def __init__(self, rag_system): super().__init__() self.rag_system rag_system def _run(self, question: str) - str: 执行工具调用。 try: contexts self.rag_system.query(question, k3) if not contexts: return 在知识库中未找到相关信息。 # 将检索到的上下文拼接成一个字符串供大模型参考 combined_context \n\n--- 参考知识 ---\n \n\n.join(contexts) return combined_context except Exception as e: return f查询知识库时发生错误: {str(e)} # --- 工具2: 模拟业务 API 调用工具 --- class BusinessAPIInput(BaseModel): 业务API调用工具的输入模型。 employee_id: str Field(description员工工号) action: str Field(description要执行的操作例如get_annual_leave_balance查询年假余额) tool(args_schemaBusinessAPIInput) def call_business_api(employee_id: str, action: str) - str: 调用内部业务系统API查询或处理与员工相关的数据。 这是一个模拟工具实际项目中应替换为真实的 API 调用。 # 模拟一个内部 API 的响应 mock_responses { get_annual_leave_balance: {balance: 10, unit: 天}, get_salary_info: {basic_salary: 15000, currency: CNY}, } if action not in mock_responses: return json.dumps({error: f不支持的操作: {action}}) # 模拟根据员工ID进行逻辑处理 result mock_responses[action] result[employee_id] employee_id result[status] success return json.dumps(result, ensure_asciiFalse) # --- 工具3: 通用计算器工具使用 LangChain 内置工具示例--- # 可以直接从 langchain 导入这里展示如何声明 from langchain.tools import Tool from langchain_community.utilities import ArxivAPIWrapper arxiv ArxivAPIWrapper() arxiv_tool Tool( namearxiv_search, description用于搜索 Arxiv 学术论文。当用户需要查询最新的研究论文时使用。, funcarxiv.run ) # 工具集合 def get_tools(rag_system): 获取所有可用工具的列表。 return [ RAGQueryTool(rag_systemrag_system), call_business_api, arxiv_tool, # 这是一个外部知识工具仅作示例 ]关键点解释每个工具都需要清晰的name和description这直接决定了大模型是否会以及如何调用它。描述要具体说明何时使用。使用 Pydantic 模型 (BaseModel) 定义输入参数可以强制类型检查并为大模型提供清晰的参数结构。在_run方法或函数内部必须做好异常处理并返回字符串结果。大模型将接收这个结果作为下一步推理的依据。工具可以串联使用例如先通过 RAG 工具查到政策再通过业务 API 工具查询个人数据。4. 使用 LangGraph 编排企业级 AI Agent 工作流有了核心模块现在我们可以用 LangGraph 来编排一个具备状态管理和长期记忆的复杂 Agent。我们将构建一个“员工自助服务 Agent”它可以处理混合了知识问答和业务操作的复杂请求。4.1 定义 Agent 状态与节点LangGraph 的核心是状态图。我们首先定义整个工作流需要维护的状态然后创建不同的节点函数来操作这个状态。创建一个agent_workflow.py文件# agent_workflow.py from typing import TypedDict, Annotated, List, Union from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage import operator from rag_system import RAGKnowledgeBase from custom_tools import get_tools from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.runnables import RunnablePassthrough # --- 1. 定义状态结构 --- class AgentState(TypedDict): 定义 Agent 工作流的状态。 这是一个有类型的字典描述了图中每个节点可以读写的数据。 # 消息列表存储整个对话历史用于模型上下文 messages: Annotated[List[Union[HumanMessage, AIMessage, SystemMessage]], add_messages] # 用户的最新输入 user_input: str # 从知识库检索到的上下文 knowledge_context: str # 工具调用的结果 tool_outputs: List[str] # 最终给用户的回复 final_answer: str # --- 2. 初始化核心组件 --- # 初始化大模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 使用 gpt-4 以获得更好的工具调用能力 # 初始化 RAG rag RAGKnowledgeBase(persist_directory./chroma_db) # 假设已构建好知识库 # 获取工具集 tools get_tools(rag) # 创建 LangChain Agent (用于工具调用决策) agent_runnable create_tool_calling_agent(llm, tools) agent_executor AgentExecutor(agentagent_runnable, toolstools, verboseTrue) # --- 3. 定义各个节点Node--- def receive_input(state: AgentState) - AgentState: 节点接收用户输入并初始化状态。 # 在实际应用中这里可能从 API 请求中获取 user_input # 我们假设 state[user_input] 已被传入 new_message HumanMessage(contentstate[user_input]) # 将用户消息添加到历史中 state[messages].append(new_message) # 清空临时字段 state[knowledge_context] state[tool_outputs] [] state[final_answer] print(f[节点: receive_input] 收到用户输入: {state[user_input][:50]}...) return state def decide_action(state: AgentState) - AgentState: 节点分析用户意图决定下一步是直接回答、查询知识库还是调用业务工具。 这是一个简化的路由决策。 last_message state[messages][-1].content.lower() # 简单的关键词路由逻辑。在实际项目中可以使用一个更复杂的分类模型或提示词。 if any(keyword in last_message for keyword in [政策, 手册, 如何, 是什么, 为什么]): state[next_action] query_knowledge elif any(keyword in last_message for keyword in [余额, 工资, 申请, 提交, 工号]): state[next_action] call_tool else: state[next_action] direct_chat print(f[节点: decide_action] 决策下一步动作为: {state[next_action]}) return state def query_knowledge_node(state: AgentState) - AgentState: 节点调用 RAG 工具查询知识库。 question state[user_input] try: # 这里直接复用之前封装的 RAG 查询功能 contexts rag.query(question, k3) knowledge_context \n\n--- 参考知识 ---\n \n\n.join(contexts) if contexts else 未找到相关信息。 state[knowledge_context] knowledge_context # 将检索到的知识也作为一条“系统”消息加入历史帮助模型理解 state[messages].append(SystemMessage(contentf相关背景知识{knowledge_context})) except Exception as e: state[knowledge_context] f知识库查询失败{str(e)} print(f[节点: query_knowledge] 已检索知识上下文。) return state def call_tool_node(state: AgentState) - AgentState: 节点执行工具调用。使用 LangChain AgentExecutor 来处理。 # 将最新的对话历史包含可能的系统消息传给 Agent inputs {input: state[messages], chat_history: state[messages][:-1]} # 最后一条是当前用户输入 try: result agent_executor.invoke(inputs) # result 是一个字典其中 output 键包含模型的最终回复 tool_output result.get(output, 工具执行完成但无明确输出。) # 记录工具调用过程中的中间输出如果 executor 的 verboseTrue会在控制台看到 state[tool_outputs].append(tool_output) # 将工具执行的结果也作为一条 AIMessage 加入历史 state[messages].append(AIMessage(contenttool_output)) except Exception as e: error_msg f工具调用过程中出错{str(e)} state[tool_outputs].append(error_msg) state[messages].append(AIMessage(contenterror_msg)) print(f[节点: call_tool] 工具调用完成。) return state def generate_final_answer(state: AgentState) - AgentState: 节点综合所有信息生成最终回复给用户。 # 准备给模型的提示词整合历史、知识、工具结果 prompt_messages state[messages] # 历史消息已包含所有上下文 try: response llm.invoke(prompt_messages) final_answer response.content state[final_answer] final_answer # 将最终回复也加入消息历史 state[messages].append(AIMessage(contentfinal_answer)) except Exception as e: state[final_answer] f生成回复时出错{str(e)} print(f[节点: generate_final_answer] 已生成最终回复。) return state # --- 4. 构建工作流图 --- def create_agent_graph(): 创建并编译 LangGraph 工作流。 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(receive_input, receive_input) workflow.add_node(decide_action, decide_action) workflow.add_node(query_knowledge, query_knowledge_node) workflow.add_node(call_tool, call_tool_node) workflow.add_node(generate_answer, generate_final_answer) # 设置入口点 workflow.set_entry_point(receive_input) # 定义边路由逻辑 workflow.add_edge(receive_input, decide_action) # 根据决策结果路由到不同分支 def route_after_decision(state): next_action state.get(next_action, direct_chat) if next_action query_knowledge: return query_knowledge elif next_action call_tool: return call_tool else: # 直接聊天跳过工具和知识库节点 return generate_answer workflow.add_conditional_edges( decide_action, route_after_decision, { query_knowledge: query_knowledge, call_tool: call_tool, generate_answer: generate_answer, } ) # 定义节点执行后的流向 workflow.add_edge(query_knowledge, generate_answer) workflow.add_edge(call_tool, generate_answer) workflow.add_edge(generate_answer, END) # 编译图 graph workflow.compile() return graph # 创建图实例 agent_graph create_agent_graph()关键点解释AgentState是一个强类型的状态容器它清晰地定义了工作流中流动的数据。Annotated和add_messages用于自动管理消息列表的合并。每个节点都是一个纯函数接收状态修改状态返回新状态。这使逻辑清晰且易于测试。decide_action节点实现了简单的路由逻辑。更复杂的场景可以使用一个小型分类模型或让大模型本身来决定下一步。add_conditional_edges是 LangGraph 的核心特性它允许根据状态动态决定下一步走向实现了复杂的业务流程如审批流、循环检查。最终编译的graph是一个可执行对象我们可以传入初始状态来运行整个工作流。4.2 运行与测试 Agent创建一个main.py文件来运行和测试我们构建的 Agent# main.py from agent_workflow import agent_graph, AgentState from langchain_core.messages import HumanMessage def run_agent_interactive(): 以交互方式运行 Agent。 print( 企业级 AI Agent 测试 ) print(输入 quit 或 exit 退出。) # 初始化一个空的对话历史 initial_state: AgentState { messages: [], # 初始为空 user_input: , knowledge_context: , tool_outputs: [], final_answer: } while True: try: user_input input(\n 用户: ).strip() if user_input.lower() in [quit, exit]: print(再见) break if not user_input: continue # 准备本次调用的输入状态 current_state initial_state.copy() current_state[user_input] user_input # 保留上一次的对话历史实现多轮对话 current_state[messages] initial_state[messages].copy() print(Agent 正在思考...) # 执行工作流图 final_state agent_graph.invoke(current_state) # 输出最终回复 print(f\n Agent: {final_state[final_answer]}) # 更新全局历史状态为下一轮对话准备 initial_state[messages] final_state[messages] # 可选打印一些调试信息 # if final_state.get(knowledge_context): # print(f[调试] 检索到的知识: {final_state[knowledge_context][:200]}...) # if final_state.get(tool_outputs): # print(f[调试] 工具输出: {final_state[tool_outputs]}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f运行过程中出现错误: {e}) if __name__ __main__: # 确保你的 OpenAI API Key 已设置环境变量 OPENAI_API_KEY # 确保你的知识库已构建运行过一次 rag_system.py 中的 build_from_documents run_agent_interactive()运行python main.py你可以测试不同类型的查询知识类“公司的年假政策是什么” - 触发query_knowledge节点。业务操作类“查询工号 E001 的年假余额。” - 触发call_tool节点模拟调用业务 API。普通聊天“你好介绍一下你自己。” - 触发direct_chat路径直接由大模型回答。5. 企业级考量安全、合规、监控与部署一个能在企业内部落地的 AI Agent除了核心功能还必须解决安全、合规和运维问题。5.1 安全与权限控制工具调用沙箱化确保工具尤其是写操作、系统命令、网络请求在受控环境中运行限制其资源访问权限如文件系统、网络。输入输出过滤与审查对用户的输入和模型的输出进行内容安全过滤防止注入攻击、敏感信息泄露或生成不当内容。基于角色的访问控制将 Agent 能力与公司现有的权限系统集成。例如只有 HR 部门的 Agent 才能调用查询全员薪资的工具。这可以在decide_action节点或工具内部实现。API 密钥与凭证管理切勿在代码中硬编码密钥。使用环境变量、密钥管理服务如 AWS Secrets Manager, HashiCorp Vault或云厂商提供的托管服务。5.2 合规与审计数据隐私确保 RAG 知识库中的文档不包含个人隐私信息PII。必要时使用数据脱敏工具。在欧盟 GDPR 或中国个人信息保护法框架下需明确告知用户数据使用方式。可解释性与审计日志记录 Agent 的完整决策链路包括用户原始输入。模型每一步的思考如果使用 ReAct 等模式。调用了哪些工具输入输出是什么。最终回复。 这些日志应存入安全的、不可篡改的存储中供事后审计。人工审核介入对于高风险操作如审批、支付、重要数据修改设计“人在环路”机制。LangGraph 的HumanInTheLoop功能可以暂停工作流等待人工确认后再继续。5.3 监控、性能与可观测性关键指标监控延迟用户查询到收到回复的总时间及各节点耗时。成本大模型 API 调用 Token 消耗、工具调用次数。成功率任务完成率、工具调用失败率。用户满意度可通过简单的“是否解决”反馈按钮收集。链路追踪使用 OpenTelemetry 等标准对工作流的每个节点进行埋点在 Jaeger 或 Grafana Tempo 中可视化整个调用链便于排查性能瓶颈和错误。优雅降级与熔断当大模型 API 或关键工具服务不可用时Agent 应有备用方案例如返回缓存答案、提示用户稍后再试或转接人工客服。5.4 部署与扩展容器化使用 Docker 将 Agent 及其依赖Python 环境、模型文件等打包确保环境一致性。编排与伸缩使用 Kubernetes 或 Docker Swarm 管理容器化后的 Agent 服务根据负载自动伸缩。API 网关通过 API 网关如 Kong, APISIX对外暴露 Agent 服务统一处理认证、限流、日志和监控。与现有系统集成身份认证集成公司的 SSO单点登录系统。消息通道将 Agent 部署为企业微信、钉钉、Slack 或 Teams 的机器人。业务流程通过 n8n、Airflow 或 Camunda 等平台将 Agent 作为自动化流程中的一个智能节点调用。6. 常见问题排查与优化实践在开发和运行 AI Agent 过程中你会遇到一些典型问题。以下是排查清单和优化建议。6.1 问题排查清单问题现象可能原因检查步骤解决方案Agent 不调用工具总是直接回答。1. 工具描述不清晰。2. 模型温度temperature过高创造性太强。3. 提示词未明确要求使用工具。1. 检查工具的name和description是否准确描述了其功能和适用场景。2. 将模型temperature设为 0 或接近 0 的值使其更确定性。3. 在给模型的系统提示词中明确指令“你必须使用提供的工具来回答问题”。优化工具描述使其更具体。在系统提示词中强化工具使用规则。使用gpt-4-turbo等工具调用能力更强的模型。RAG 检索结果不相关。1. 文档分割策略不佳。2. 嵌入模型不匹配。3. 检索 top-k 值不合适。1. 检查分割后的文本块是否语义完整。2. 尝试不同的嵌入模型如text-embedding-3-large。3. 调整similarity_search的k参数或尝试max_marginal_relevance_search提升多样性。调整文本分割器的chunk_size和chunk_overlap。对查询进行重写或扩展。引入元数据过滤如按文档类型检索。工作流状态混乱或丢失。1. 状态对象在节点间被意外覆盖。2. 多线程/异步调用导致状态竞争。1. 检查每个节点函数是否都正确返回了完整的state字典。2. 确保AgentState中Annotated字段的归约函数如add_messages使用正确。遵循 LangGraph 最佳实践将状态修改限制在节点函数内。对于持久化状态使用数据库支持的状态后端。多轮对话中Agent 忘记之前的内容。1. 对话历史未正确传递给模型。2. 上下文长度超限历史被截断。1. 检查state[‘messages’]列表是否包含了所有历史消息。2. 计算已使用的 Token 数确认是否超限。确保每轮对话都将更新后的messages传回图中。对于长对话实现摘要机制将过长的历史压缩成摘要存入长期记忆。工具调用超时或失败。1. 工具依赖的外部 API 不可用。2. 网络或权限问题。3. 工具函数内部有 bug。1. 直接调用工具函数检查其独立运行是否正常。2. 查看工具函数的错误日志和异常捕获。3. 检查网络连接和 API 密钥。为工具调用添加超时和重试机制。实现完善的错误处理返回清晰的错误信息供模型和用户理解。6.2 性能与效果优化实践提示词工程优化系统提示词明确 Agent 的角色、职责、约束和输出格式。例如“你是一个严谨的员工自助助手必须基于公司知识库和可用工具来回答问题不得编造信息。”思维链对于复杂问题在提示词中要求模型“逐步思考”这能提升其推理和工具调用的准确性。RAG 优化查询重写在检索前先用一个小模型将用户查询重写为更利于检索的形式如关键词提取、问题分解、陈述句转换。混合检索结合向量检索语义和关键词检索BM25取长补短。重排序初步检索出较多结果如 k20后用一个更精细的交叉编码器模型对结果进行重排序选取最相关的几个。记忆优化对话摘要当对话轮数增多时将早期对话总结成一段摘要替换掉原始的长篇历史以节省 Token 并保留核心信息。向量记忆将重要的对话片段或事实存入另一个向量库当后续对话涉及相关话题时主动检索出来作为上下文。成本优化模型分级简单的意图识别或路由使用小模型如 GPT-3.5-turbo复杂的推理和生成再用大模型如 GPT-4。缓存对常见的、结果不变的查询如“公司地址”将 RAG 结果或模型回复缓存起来。Token 管理监控并设置 Token 使用上限防止异常输入导致巨额费用。从零构建企业级 AI Agent 是一个系统工程涉及模型、框架、数据、工具和运维多个层面。本文提供了一个以 LangGraph 和 RAG 为核心的可运行原型和完整的设计思路。真正的挑战在于将这套原型与具体的企业环境、业务流程和安全规范深度融合。建议从一个小而具体的业务场景开始试点例如 IT 内部问答或新员工入职引导在迭代中不断完善其可靠性、安全性和用户体验逐步扩展其能力边界。