最近和不少技术负责人聊发现一个有意思的现象大家普遍对 AI Agent 很感兴趣但真正动手落地的团队却不多。问起原因答案出奇地一致——“概念都懂但不知道从哪开始更不知道怎么做成企业能用的东西。”这背后其实是一个典型的“技术断层”网上充斥着各种“10分钟搭建Agent”的入门教程但当你真的想把它集成到现有业务系统处理复杂的审批流、调用内部API、保证数据安全时却发现无从下手。从玩具Demo到企业级应用中间隔着一道巨大的工程化鸿沟。这篇文章要解决的正是这个问题。我们不谈空泛的趋势而是聚焦于一个核心判断企业级AI Agent的核心竞争力不在于用了多先进的模型而在于能否将大语言模型LLM的“智能”与现有企业系统的“确定性流程”和“安全边界”可靠地结合起来。这本质是一个系统集成和工程化问题。本文将带你从零开始以实战视角构建一个具备企业级雏形的AI Agent。我们会使用当前最主流的技术栈LangGraph RAG 工具调用但重点不是复现一个Demo而是拆解其中每一个环节的工程化考量如何设计状态管理、如何集成知识库、如何安全地调用工具、如何设计人工审核节点。读完本文你将获得一套可复用的架构思路和代码模板能够真正启动你的第一个企业级Agent项目。1. 企业级AI Agent从“玩具”到“工具”的关键跨越在深入代码之前我们必须先厘清一个关键问题什么是“企业级”AI Agent它和你在Github上看到的那些炫酷Demo有什么区别简单来说企业级Agent需要解决三个核心挑战确定性与可靠性的平衡大模型本质是概率模型会“胡言乱语”。企业流程如财务报销、订单审核要求100%的确定性。企业级Agent必须能在关键节点引入规则引擎或人工审核确保最终输出的可靠性。复杂状态与流程管理一个客服Agent处理用户投诉可能需要经历“理解问题 - 查询知识库 - 生成方案 - 转交人工 - 记录结果”等多个步骤并且要记住上下文。这远非一个简单的if-else链条能解决。安全与合规集成Agent需要调用内部数据库、CRM、OA系统。如何管理认证凭据如何防止Agent越权访问如何审计它的每一步操作这些是企业安全部门的“必答题”。传统的LangChain等框架擅长构建线性的、简单的链Chain但在处理带有循环、分支、并行和状态保持的复杂工作流时就显得力不从心。这正是LangGraph这类框架的价值所在。它用“图”Graph的思维来建模Agent的工作流将每个步骤抽象为节点Node步骤间的流转抽象为边Edge并引入一个中心化的“状态”State对象来管理整个对话的上下文。这种设计天然适合企业复杂的业务流程。同时RAG检索增强生成解决了企业私有知识库的接入问题而工具调用Tool Calling则让Agent拥有了操作现实世界企业系统的“手”和“脚”。因此我们本次实战的技术选型非常明确以LangGraph作为工作流引擎和大脑以RAG作为知识库以工具调用作为执行器构建一个具备记忆、推理和行动能力的智能体。2. 核心概念与架构总览在动手之前我们需要统一几个关键概念的理解这能避免后续开发中的很多困惑。智能体Agent一个能感知环境、进行决策并执行动作以实现目标的系统。在我们的语境中它就是一个由LLM驱动、能按预定流程调用工具和知识的程序。工作流Workflow/ 图Graph在LangGraph中你将Agent的完整执行逻辑定义为一个有向图。节点是功能单元如“调用LLM”、“查询数据库”边定义了节点间的流转条件如“如果用户满意则结束否则转人工”。状态State一个贯穿整个工作流的共享数据结构。它像是一个全局白板记录了用户输入、LLM的思考过程、工具调用结果、中间决策等所有信息。这是实现复杂对话记忆和多轮交互的核心。工具ToolAgent可以调用的外部函数。它可以是一个简单的计算器也可以是一个需要认证的企业内部API封装。工具调用让Agent从“聊天机器人”升级为“自动执行器”。RAG检索增强生成一种让LLM能够基于特定知识库非其训练数据进行回答的技术。核心流程是“检索Retrieve相关文档片段 - 增强AugmentLLM的提示词 - 生成Generate答案”。它解决了LLM的“幻觉”和企业数据安全的问题。我们的目标架构如下图所示概念示意用户输入 | v [入口节点路由与意图识别] | v {状态管理LangGraph State} | v |----------------- [工具调用节点] ---- 调用内部API/DB | | | v [核心决策节点LLM] ----(工具结果)---- [状态更新] | v |----------------- [RAG检索节点] ---- 查询向量知识库 | | | v [核心决策节点LLM] ----(检索结果)---- [状态更新] | v [条件判断边] ---- 是否需人工审核 --是-- [人工干预节点] | v [输出节点回复用户]这个架构中核心决策节点LLM是大脑它根据当前状态决定下一步是调用工具、检索知识还是直接回复。状态是中枢神经传递所有信息。工具和RAG是四肢和感官。LangGraph是调度这一切的神经系统。3. 环境准备与项目初始化我们使用Python作为开发语言。请确保你的环境满足以下要求Python版本 3.10 推荐3.10或3.11避免最新版本可能存在的包兼容性问题包管理工具pip 或 condaLLM API我们将使用OpenAI的GPT-4系列模型作为“大脑”。你需要准备一个有效的OpenAI API Key。如果你希望完全本地化也可以使用Ollama部署本地模型如Llama 3.1但本文以OpenAI为例因其工具调用功能最稳定。第一步创建项目并安装核心依赖# 创建项目目录 mkdir enterprise-ai-agent cd enterprise-ai-agent # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心框架 pip install langgraph langchain langchain-openai # 安装向量数据库客户端以Chroma为例轻量易用 pip install chromadb # 安装其他工具库 pip install pydantic python-dotenv第二步组织项目结构一个清晰的结构是工程化的第一步。建议如下enterprise-ai-agent/ ├── .env # 环境变量存放API KEY等敏感信息 ├── requirements.txt # 项目依赖 ├── main.py # 应用主入口 ├── core/ # 核心逻辑 │ ├── __init__.py │ ├── state.py # 定义状态State类 │ ├── graph.py # 定义工作流图Graph │ └── nodes.py # 定义所有节点Node函数 ├── tools/ # 工具定义 │ ├── __init__.py │ ├── calculator.py # 示例计算器工具 │ └── enterprise_tools.py # 模拟的企业内部工具如查询订单 ├── knowledge/ # RAG知识库相关 │ ├── __init__.py │ ├── vector_store.py # 向量数据库初始化与操作 │ └── docs/ # 存放待处理的原始文档PDF/TXT等 └── config.py # 配置文件第三步配置环境变量在项目根目录创建.env文件并填入你的OpenAI API Key。切记将此文件加入.gitignore不要提交到代码仓库。# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here在config.py中读取配置# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) # 其他配置如模型名称、向量数据库路径等 MODEL_NAME gpt-4o # 或 gpt-4-turbo PERSIST_DIRECTORY ./chroma_db # 向量数据库持久化目录4. 定义工作流的核心状态State状态是LangGraph工作流的“记忆体”。我们使用Pydantic的BaseModel来严格定义状态的结构这能带来良好的类型提示和验证。# core/state.py from typing import TypedDict, List, Optional, Any from pydantic import BaseModel, Field class AgentState(TypedDict): LangGraph 工作流的共享状态。 使用TypedDict是为了与LangGraph的接口更好兼容。 # 用户输入 input: str # 对话历史 chat_history: List[str] # LLM的完整思考过程chain of thought reasoning: Optional[str] # 当前需要执行的动作由LLM决定 next_action: Optional[str] # 工具调用的结果 tool_output: Optional[str] # RAG检索到的文档内容 retrieved_docs: Optional[List[str]] # 最终输出给用户的结果 final_output: Optional[str] # 其他自定义上下文如用户ID、会话ID等 context: Optional[dict] class StrictAgentState(BaseModel): 使用Pydantic BaseModel定义的严格状态用于内部数据验证和传递。 在实际节点函数中我们可能使用这个类来处理数据。 input: str Field(description用户的最新输入) chat_history: List[str] Field(default_factorylist, description对话历史列表格式为[user: xxx, assistant: yyy]) reasoning: Optional[str] Field(defaultNone, descriptionLLM的推理过程) next_action: Optional[str] Field(defaultNone, description下一步动作如 call_tool:get_weather, retrieve_knowledge, respond) tool_output: Optional[str] Field(defaultNone, description工具调用的返回结果) retrieved_docs: Optional[List[str]] Field(defaultNone, description检索到的相关文档列表) final_output: Optional[str] Field(defaultNone, description最终回复内容) context: dict Field(default_factorydict, description扩展上下文信息) class Config: arbitrary_types_allowed True这个状态定义包含了Agent运行所需的所有信息。注意next_action字段它将由LLM来填充用于指导工作流的下一个节点这是实现动态路由的关键。5. 构建工具Tools与知识库RAGAgent的能力边界由它的工具和知识决定。我们先构建两个基础能力。5.1 实现一个简单的工具我们先创建一个计算器工具和一个模拟的“查询订单状态”工具。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行一个数学表达式计算。支持加减乘除-*/和乘方**。 例如calculator(3 5 * 2) 返回 13。 try: # 警告在生产环境中直接eval是危险的此处仅作演示。 # 真实场景应使用更安全的表达式解析库如 ast.literal_eval 配合限制。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # tools/enterprise_tools.py from langchain.tools import tool import random from datetime import datetime tool def get_order_status(order_id: str) - str: 根据订单ID查询订单状态。这是一个模拟的企业内部系统接口。 参数: order_id: 订单编号例如 ORD-2024-001 返回: 订单状态的描述字符串。 # 模拟数据库查询 status_options [已创建, 支付中, 已发货, 配送中, 已完成, 已取消] # 为了演示让状态看起来有点规律 status_index hash(order_id) % len(status_options) estimated_delivery (datetime.now().day 3) % 30 return f订单 {order_id} 当前状态为【{status_options[status_index]}】。预计送达日期本月{estimated_delivery}日。 tool def escalate_to_human(reason: str, user_id: str) - str: 将当前会话升级至人工客服。这是一个关键的安全与合规工具。 参数: reason: 升级原因例如 用户情绪激动 或 问题超出知识范围。 user_id: 用户标识。 返回: 告知用户已转接人工的提示并返回一个工单号。 ticket_id fTICKET-{datetime.now().strftime(%Y%m%d)}-{random.randint(1000,9999)} # 在实际系统中这里会调用工单系统API创建工单并通知客服人员。 return f您的问题已转接人工客服工单号{ticket_id}。客服人员将尽快与您联系。转接原因{reason}5.2 构建RAG知识库我们使用ChromaDB作为向量数据库并加载一些示例文档。# knowledge/vector_store.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader import os from config import OPENAI_API_KEY, PERSIST_DIRECTORY class KnowledgeBase: def __init__(self, persist_directory: str PERSIST_DIRECTORY): self.embeddings OpenAIEmbeddings(openai_api_keyOPENAI_API_KEY) self.persist_directory persist_directory 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.embeddings ) else: print(未找到已有向量库将创建新库。) # 可以在这里初始化一个空的VectorStore或者调用ingest_documents方法 self.vector_store None def ingest_documents(self, document_paths: list): 将文档切片并存入向量数据库。 参数: document_paths: 文档路径列表。 all_docs [] text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段约500字符 chunk_overlap50 # 片段间重叠50字符保证上下文连贯 ) for path in document_paths: if not os.path.exists(path): print(f警告文件 {path} 不存在跳过。) continue try: loader TextLoader(path, encodingutf-8) documents loader.load() splits text_splitter.split_documents(documents) all_docs.extend(splits) print(f已处理文档: {path}, 得到 {len(splits)} 个片段。) except Exception as e: print(f处理文档 {path} 时出错: {e}) if all_docs: print(f开始创建向量库共 {len(all_docs)} 个文本片段...) self.vector_store Chroma.from_documents( documentsall_docs, embeddingself.embeddings, persist_directoryself.persIST_DIRECTORY ) self.vector_store.persist() print(f向量库已创建并持久化到 {self.persist_directory}) else: print(没有有效的文档可供处理。) def retrieve(self, query: str, k: int 3) - list: 检索与查询最相关的k个文档片段。 参数: query: 查询文本。 k: 返回的片段数量。 返回: 相关文档片段的文本内容列表。 if self.vector_store is None: print(向量库未初始化请先调用 ingest_documents。) return [] try: docs self.vector_store.similarity_search(query, kk) return [doc.page_content for doc in docs] except Exception as e: print(f检索过程中出错: {e}) return [] # 示例创建知识库并摄入文档 if __name__ __main__: kb KnowledgeBase() # 假设在 knowledge/docs/ 目录下有一些.txt文件 doc_paths [./knowledge/docs/company_policy.txt, ./knowledge/docs/product_faq.txt] kb.ingest_documents(doc_paths) # 测试检索 results kb.retrieve(公司的年假政策是什么) for i, r in enumerate(results): print(f结果 {i1}: {r[:100]}...)你需要提前在knowledge/docs/目录下准备一些文本文件作为知识库。例如company_policy.txt内容可以是公司员工手册摘要 1. 年假政策入职满一年后享有10天年假之后每增加一年工龄年假增加1天上限15天。 2. 报销流程员工需在费用发生后的30天内通过内部OA系统提交电子发票和审批单。 3. 考勤制度标准工作时间为周一至周五9:00-18:00。弹性工作制需部门领导批准。6. 定义工作流节点Nodes节点是图的基本执行单元。每个节点是一个函数它接收当前State执行一些操作如调用LLM、运行工具然后返回更新后的State。# core/nodes.py from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List import json from core.state import AgentState, StrictAgentState from tools.calculator import calculator from tools.enterprise_tools import get_order_status, escalate_to_human from knowledge.vector_store import KnowledgeBase from config import OPENAI_API_KEY, MODEL_NAME # 初始化LLM和知识库 llm ChatOpenAI(modelMODEL_NAME, openai_api_keyOPENAI_API_KEY, temperature0) kb KnowledgeBase() # 注意这里假设知识库已初始化并有数据 def route_input(state: AgentState) - AgentState: 路由节点分析用户输入决定工作流走向。 这是第一个节点负责意图识别。 print(f[路由节点] 收到输入: {state[input]}) messages [ SystemMessage(content你是一个智能路由助手。请分析用户意图并只返回一个JSON对象包含next_action字段。), HumanMessage(contentf用户说{state[input]}。\n\n可选动作\n1. need_calculation - 如果涉及数学计算。\n2. query_order - 如果询问订单状态。\n3. general_qa - 如果是一般性问答或公司政策咨询。\n4. escalate - 如果用户表达强烈不满或问题复杂。\n\n请只返回JSON如 {{\next_action\: \general_qa\}}) ] response llm.invoke(messages) try: decision json.loads(response.content) state[next_action] decision.get(next_action, general_qa) state[reasoning] f路由分析用户意图被分类为 {state[next_action]} except json.JSONDecodeError: state[next_action] general_qa state[reasoning] f路由分析LLM返回格式错误默认路由到 general_qa。响应{response.content} return state def call_llm_for_decision(state: AgentState) - AgentState: 核心决策节点根据当前状态和next_action调用LLM决定具体执行什么。 这个节点是工作流的“大脑”。 print(f[决策节点] 当前动作指令: {state.get(next_action)}) # 构建系统提示词根据不同的next_action给予不同的指令 system_prompt 你是一个企业级AI助手。请根据对话历史、用户当前输入和已有的工具调用结果或检索到的知识决定下一步做什么并生成相应的内容。 if state.get(next_action) need_calculation: system_prompt 用户需要计算。请分析其输入提取出数学表达式并准备调用计算器工具。 elif state.get(next_action) query_order: system_prompt 用户需要查询订单。请从其输入中提取出订单号格式类似ORD-XXXX-XXX并准备调用订单查询工具。 elif state.get(next_action) general_qa: system_prompt 用户在进行一般性问答。请结合检索到的公司知识库内容给出准确、友好的回答。 else: system_prompt 请综合所有信息生成对用户的回复。 # 构建对话历史 chat_history_text \n.join(state.get(chat_history, [])[-4:]) # 取最近4轮对话 user_input state[input] # 构建检索到的知识上下文 knowledge_context if state.get(retrieved_docs): knowledge_context \n\n【相关公司知识】\n \n---\n.join(state[retrieved_docs][:2]) # 取前2个片段 # 构建工具调用结果上下文 tool_context if state.get(tool_output): tool_context f\n\n【工具执行结果】\n{state[tool_output]} full_prompt f{system_prompt}\n\n对话历史\n{chat_history_text}\n\n用户最新输入{user_input}{knowledge_context}{tool_context}\n\n请决定下一步1. 如果需要调用工具请说明调用哪个工具以及参数。2. 如果可以直接回答请生成回复内容。 messages [HumanMessage(contentfull_prompt)] response llm.invoke(messages) # 这里简化处理实际应该让LLM以结构化格式如JSON返回决策。 # 为了演示我们简单地将LLM的回复存入状态由后续节点解析。 state[reasoning] response.content return state def execute_tool(state: AgentState) - AgentState: 工具执行节点解析LLM的决策调用相应的工具。 reasoning state.get(reasoning, ) print(f[工具执行节点] 解析决策: {reasoning[:100]}...) # 这是一个简化的工具调用逻辑。在实际项目中你应该使用LangChain的Tool Calling功能 # 让LLM直接返回结构化工具调用请求。这里我们做简单的关键字匹配。 tool_output None if 计算 in reasoning or calculator in reasoning.lower(): # 尝试提取表达式这里非常简化 # 实际应用应使用更鲁棒的提取方法或直接使用LLM的tool_calling能力 import re expr_match re.search(r(\d[\\-\*/]\d), state[input]) if expr_match: expr expr_match.group(1) tool_output calculator.invoke(expr) else: tool_output 无法从输入中提取有效的数学表达式。 elif 订单 in reasoning or order in reasoning.lower(): # 尝试提取订单号 import re order_id_match re.search(rORD-\w-\d, state[input]) if order_id_match: order_id order_id_match.group(0) tool_output get_order_status.invoke(order_id) else: tool_output 未在输入中找到有效的订单号格式如ORD-2024-001。 elif 人工 in reasoning or escalate in reasoning.lower(): tool_output escalate_to_human.invoke(用户请求或问题复杂, user_123) if tool_output: state[tool_output] tool_output print(f[工具执行节点] 工具调用结果: {tool_output}) else: state[tool_output] 未触发任何工具调用。 return state def retrieve_knowledge(state: AgentState) - AgentState: 知识检索节点从向量数据库中检索与用户问题相关的文档。 if state.get(next_action) general_qa: query state[input] print(f[知识检索节点] 检索查询: {query}) docs kb.retrieve(query, k2) state[retrieved_docs] docs print(f[知识检索节点] 检索到 {len(docs)} 条相关文档。) else: state[retrieved_docs] [] return state def generate_final_response(state: AgentState) - AgentState: 最终响应生成节点综合所有信息生成给用户的最终回复。 print([响应生成节点] 生成最终回复...) # 这里可以再次调用LLM利用所有上下文生成更连贯的回复。 # 为了简化我们直接拼接信息。 final_parts [] if state.get(tool_output) and 未触发 not in state[tool_output]: final_parts.append(f根据系统查询{state[tool_output]}) if state.get(retrieved_docs): # 简单取第一条检索结果作为回答依据实际应更智能地整合 if state[retrieved_docs]: final_parts.append(f根据公司规定{state[retrieved_docs][0][:150]}...) if not final_parts: final_parts.append(我已收到您的请求。) state[final_output] .join(final_parts) # 更新对话历史 state[chat_history].append(fuser: {state[input]}) state[chat_history].append(fassistant: {state[final_output]}) return state def human_intervention(state: AgentState) - AgentState: 人工干预节点模拟人工客服接入。 在实际系统中这里会触发通知、创建工单等。 print([人工干预节点] 触发人工客服流程...) state[final_output] 您的问题已转接给人工客服请稍候。 # 这里可以添加调用工单系统的逻辑 return state7. 组装工作流图Graph并运行现在我们将所有节点连接起来形成一个完整的工作流。# core/graph.py from langgraph.graph import StateGraph, END from core.state import AgentState from core.nodes import route_input, call_llm_for_decision, execute_tool, retrieve_knowledge, generate_final_response, human_intervention def create_agent_workflow(): 创建并返回配置好的工作流图 workflow StateGraph(AgentState) # 1. 添加节点 workflow.add_node(router, route_input) # 路由 workflow.add_node(decision, call_llm_for_decision) # 决策 workflow.add_node(retrieve, retrieve_knowledge) # 检索 workflow.add_node(execute_tool, execute_tool) # 执行工具 workflow.add_node(generate_response, generate_final_response) # 生成回复 workflow.add_node(human_intervention, human_intervention) # 人工干预 # 2. 设置入口点 workflow.set_entry_point(router) # 3. 定义边条件流转 # 从路由节点到决策节点 workflow.add_edge(router, decision) # 从决策节点出来根据情况分流 def decide_after_decision(state: AgentState) - str: 决策节点后的条件判断 reasoning state.get(reasoning, ).lower() if 人工 in reasoning or escalate in reasoning: return human_intervention elif 计算 in reasoning or 订单 in reasoning: return execute_tool elif 知识 in reasoning or general_qa in state.get(next_action, ): return retrieve else: return generate_response workflow.add_conditional_edges( decision, decide_after_decision, { human_intervention: human_intervention, execute_tool: execute_tool, retrieve: retrieve, generate_response: generate_response, } ) # 4. 定义其他节点的流转 # 工具执行后去生成回复 workflow.add_edge(execute_tool, generate_response) # 知识检索后去生成回复 workflow.add_edge(retrieve, generate_response) # 人工干预后直接结束 workflow.add_edge(human_intervention, END) # 生成回复后结束本轮 workflow.add_edge(generate_response, END) # 5. 编译图 app workflow.compile() return app # 主程序入口 if __name__ __main__: from core.graph import create_agent_workflow app create_agent_workflow() # 测试1数学计算 print(\n 测试1数学计算 ) initial_state {input: 帮我算一下 125 37 等于多少, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试2订单查询 print(\n 测试2订单查询 ) initial_state {input: 我的订单 ORD-2024-1001 到哪里了, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试3知识库问答 print(\n 测试3知识库问答 ) initial_state {input: 我们公司的年假有多少天, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试4触发人工 print(\n 测试4触发人工 ) initial_state {input: 我要投诉你们的产品有问题, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]})8. 运行结果与效果验证运行python core/graph.py你应该能看到类似以下的输出 测试1数学计算 [路由节点] 收到输入: 帮我算一下 125 37 等于多少 [决策节点] 当前动作指令: need_calculation [工具执行节点] 解析决策: 用户需要计算。请分析其输入提取出数学表达式并准备调用计算器工具。... [工具执行节点] 工具调用结果: 计算结果: 162 [响应生成节点] 生成最终回复... 最终回复: 根据系统查询计算结果: 162 测试2订单查询 [路由节点] 收到输入: 我的订单 ORD-2024-1001 到哪里了 [决策节点] 当前动作指令: query_order [工具执行节点] 解析决策: 用户需要查询订单。请从其输入中提取出订单号格式类似ORD-XXXX-XXX并准备调用订单查询工具。... [工具执行节点] 工具调用结果: 订单 ORD-2024-1001 当前状态为【已发货】。预计送达日期本月15日。 [响应生成节点] 生成最终回复... 最终回复: 根据系统查询订单 ORD-2024-1001 当前状态为【已发货】。预计送达日期本月15日。 测试3知识库问答 [路由节点] 收到输入: 我们公司的年假有多少天 [决策节点] 当前动作指令: general_qa [知识检索节点] 检索查询: 我们公司的年假有多少天 [知识检索节点] 检索到 2 条相关文档。 [响应生成节点] 生成最终回复... 最终回复: 根据公司规定公司员工手册摘要1. 年假政策入职满一年后享有10天年假之后每增加一年工龄年假增加1天上限15天。... 测试4触发人工 [路由节点] 收到输入: 我要投诉你们的产品有问题 [决策节点] 当前动作指令: escalate [人工干预节点] 触发人工客服流程... 最终回复: 您的问题已转接给人工客服请稍候。如何验证成功流程正确针对不同类型的输入工作流正确地路由到了不同的分支计算、查询、检索、人工。工具调用成功计算器和订单查询工具被正确调用并返回了结果。RAG生效对于政策类问题成功从向量库中检索到了相关片段。状态流转State对象在各个节点间正确传递和更新最终生成了包含上下文的回复。9. 常见问题与排查思路在实际部署中你可能会遇到以下问题问题现象可能原因排查方式解决方案启动时报ModuleNotFoundError依赖未安装或虚拟环境未激活1. 运行pip list | grep langchain检查。2. 确认终端路径在项目目录下。1. 激活虚拟环境。2. 运行pip install -r requirements.txt。OpenAI API 调用失败API Key 错误或网络问题1. 检查.env文件格式和内容。2. 在Python中直接调用openai.ChatCompletion.create测试。1. 确保.env文件在根目录且变量名正确。2. 检查网络连接和API配额。向量数据库检索无结果1. 文档未成功摄入。2. 查询语句与文档语义不匹配。1. 检查ingest_documents时的日志和chroma_db目录大小。2. 直接运行kb.retrieve(“test”)看是否有基础结果。1. 确认文档路径正确且内容为纯文本。2. 尝试调整chunk_size或使用不同的嵌入模型。工作流卡在某个节点无输出节点函数逻辑错误或返回状态格式不对。1. 在每个节点函数开始添加print语句。2. 检查节点函数返回值是否为字典且包含必需的State键。1. 使用pdb或ipdb进行调试。2. 确保所有节点函数都接收并返回AgentState类型。LLM 的决策不符合预期提示词Prompt设计不佳。1. 打印出call_llm_for_decision节点中的full_prompt。2. 在OpenAI Playground中手动测试提示词。1. 优化系统提示词给出更明确的指令和示例。2. 考虑使用LangChain的StructuredOutputParser来约束LLM输出格式。工具调用参数提取错误在execute_tool节点中使用了简单的正则匹配不鲁棒。观察reasoning字段的内容看LLM是否给出了清晰的工具调用指令。最佳实践使用LangChain的bind_tools和with_structured_output功能让LLM直接返回结构化的工具调用请求。10. 企业级最佳实践与工程建议将上述Demo推向企业生产环境你需要考虑以下关键点1. 状态管理的持久化问题当前State存在于内存中服务重启后对话历史丢失。方案将AgentState与一个唯一的session_id绑定并存入Redis或数据库。每个请求根据session_id加载历史状态。2. 工具调用的安全与审计问题工具可能执行高危操作如删除数据、发送消息。方案权限控制为每个工具定义权限等级并在调用前校验用户/会话权限。参数校验与净化对所有输入参数进行严格的类型、范围和内容校验。操作审计记录每一次工具调用的时间、用户、参数和结果便于追溯。模拟模式在生产环境外运行一个“模拟环境”Agent可以调用工具但不会产生真实副作用。3. RAG的优化问题简单的向量检索可能返回不相关或碎片化信息。方案混合检索结合向量检索语义相似和关键词检索BM25提升召回率。重排序Re-ranking使用更精细的模型对检索结果进行重排序提升精度。元数据过滤为文档片段添加元数据如部门、有效期检索时进行过滤。查询改写在检索前先用LLM对用户原始查询进行扩展或改写。4. 工作流的可观测性与调试问题Agent的决策过程是黑盒出错难以排查。方案全链路日志在每个节点的入口和出口记录详细的日志包括输入、输出、耗时。可视化利用LangGraph Studio如果可用或自定义图表来可视化工作流的执行路径。追踪Tracing集成像LangSmith这样的平台记录每一次LLM调用、工具调用和节点流转。5. 版本控制与灰度发布问题直接更新Agent的工作流或提示词可能影响线上业务。方案配置化将提示词、工具列表、路由规则等抽取为配置文件。A/B测试对新旧版本的Agent进行流量切分对比关键指标如解决率、用户满意度。回滚机制确保能快速回退到上一个稳定版本。6. 与现有系统集成问题Agent需要调用大量的内部系统API。方案API网关通过统一的API网关来管理和调用内部服务Agent工具层只需调用网关。认证传递安全地处理用户身份将其传递给下游系统如使用JWT Token。异步处理对于耗时的工具调用如生成报告采用异步任务队列如Celery避免阻塞Agent的响应。构建企业级AI Agent是一个系统工程它考验的不仅是你对LLM和框架的掌握更是你对软件工程、系统架构和安全合规的综合理解。本文提供的代码是一个坚实的起点但真正的挑战和价值在于你如何根据自己企业的独特需求在这个骨架上填充血肉——设计更精细的状态、开发更强大的工具、构建更准确的知识库并最终将它无缝、安全、可靠地融入到你现有的业务血脉之中。 企业级AI Agent实战：基于LangGraph与RAG的工程化架构与实现
最近和不少技术负责人聊发现一个有意思的现象大家普遍对 AI Agent 很感兴趣但真正动手落地的团队却不多。问起原因答案出奇地一致——“概念都懂但不知道从哪开始更不知道怎么做成企业能用的东西。”这背后其实是一个典型的“技术断层”网上充斥着各种“10分钟搭建Agent”的入门教程但当你真的想把它集成到现有业务系统处理复杂的审批流、调用内部API、保证数据安全时却发现无从下手。从玩具Demo到企业级应用中间隔着一道巨大的工程化鸿沟。这篇文章要解决的正是这个问题。我们不谈空泛的趋势而是聚焦于一个核心判断企业级AI Agent的核心竞争力不在于用了多先进的模型而在于能否将大语言模型LLM的“智能”与现有企业系统的“确定性流程”和“安全边界”可靠地结合起来。这本质是一个系统集成和工程化问题。本文将带你从零开始以实战视角构建一个具备企业级雏形的AI Agent。我们会使用当前最主流的技术栈LangGraph RAG 工具调用但重点不是复现一个Demo而是拆解其中每一个环节的工程化考量如何设计状态管理、如何集成知识库、如何安全地调用工具、如何设计人工审核节点。读完本文你将获得一套可复用的架构思路和代码模板能够真正启动你的第一个企业级Agent项目。1. 企业级AI Agent从“玩具”到“工具”的关键跨越在深入代码之前我们必须先厘清一个关键问题什么是“企业级”AI Agent它和你在Github上看到的那些炫酷Demo有什么区别简单来说企业级Agent需要解决三个核心挑战确定性与可靠性的平衡大模型本质是概率模型会“胡言乱语”。企业流程如财务报销、订单审核要求100%的确定性。企业级Agent必须能在关键节点引入规则引擎或人工审核确保最终输出的可靠性。复杂状态与流程管理一个客服Agent处理用户投诉可能需要经历“理解问题 - 查询知识库 - 生成方案 - 转交人工 - 记录结果”等多个步骤并且要记住上下文。这远非一个简单的if-else链条能解决。安全与合规集成Agent需要调用内部数据库、CRM、OA系统。如何管理认证凭据如何防止Agent越权访问如何审计它的每一步操作这些是企业安全部门的“必答题”。传统的LangChain等框架擅长构建线性的、简单的链Chain但在处理带有循环、分支、并行和状态保持的复杂工作流时就显得力不从心。这正是LangGraph这类框架的价值所在。它用“图”Graph的思维来建模Agent的工作流将每个步骤抽象为节点Node步骤间的流转抽象为边Edge并引入一个中心化的“状态”State对象来管理整个对话的上下文。这种设计天然适合企业复杂的业务流程。同时RAG检索增强生成解决了企业私有知识库的接入问题而工具调用Tool Calling则让Agent拥有了操作现实世界企业系统的“手”和“脚”。因此我们本次实战的技术选型非常明确以LangGraph作为工作流引擎和大脑以RAG作为知识库以工具调用作为执行器构建一个具备记忆、推理和行动能力的智能体。2. 核心概念与架构总览在动手之前我们需要统一几个关键概念的理解这能避免后续开发中的很多困惑。智能体Agent一个能感知环境、进行决策并执行动作以实现目标的系统。在我们的语境中它就是一个由LLM驱动、能按预定流程调用工具和知识的程序。工作流Workflow/ 图Graph在LangGraph中你将Agent的完整执行逻辑定义为一个有向图。节点是功能单元如“调用LLM”、“查询数据库”边定义了节点间的流转条件如“如果用户满意则结束否则转人工”。状态State一个贯穿整个工作流的共享数据结构。它像是一个全局白板记录了用户输入、LLM的思考过程、工具调用结果、中间决策等所有信息。这是实现复杂对话记忆和多轮交互的核心。工具ToolAgent可以调用的外部函数。它可以是一个简单的计算器也可以是一个需要认证的企业内部API封装。工具调用让Agent从“聊天机器人”升级为“自动执行器”。RAG检索增强生成一种让LLM能够基于特定知识库非其训练数据进行回答的技术。核心流程是“检索Retrieve相关文档片段 - 增强AugmentLLM的提示词 - 生成Generate答案”。它解决了LLM的“幻觉”和企业数据安全的问题。我们的目标架构如下图所示概念示意用户输入 | v [入口节点路由与意图识别] | v {状态管理LangGraph State} | v |----------------- [工具调用节点] ---- 调用内部API/DB | | | v [核心决策节点LLM] ----(工具结果)---- [状态更新] | v |----------------- [RAG检索节点] ---- 查询向量知识库 | | | v [核心决策节点LLM] ----(检索结果)---- [状态更新] | v [条件判断边] ---- 是否需人工审核 --是-- [人工干预节点] | v [输出节点回复用户]这个架构中核心决策节点LLM是大脑它根据当前状态决定下一步是调用工具、检索知识还是直接回复。状态是中枢神经传递所有信息。工具和RAG是四肢和感官。LangGraph是调度这一切的神经系统。3. 环境准备与项目初始化我们使用Python作为开发语言。请确保你的环境满足以下要求Python版本 3.10 推荐3.10或3.11避免最新版本可能存在的包兼容性问题包管理工具pip 或 condaLLM API我们将使用OpenAI的GPT-4系列模型作为“大脑”。你需要准备一个有效的OpenAI API Key。如果你希望完全本地化也可以使用Ollama部署本地模型如Llama 3.1但本文以OpenAI为例因其工具调用功能最稳定。第一步创建项目并安装核心依赖# 创建项目目录 mkdir enterprise-ai-agent cd enterprise-ai-agent # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心框架 pip install langgraph langchain langchain-openai # 安装向量数据库客户端以Chroma为例轻量易用 pip install chromadb # 安装其他工具库 pip install pydantic python-dotenv第二步组织项目结构一个清晰的结构是工程化的第一步。建议如下enterprise-ai-agent/ ├── .env # 环境变量存放API KEY等敏感信息 ├── requirements.txt # 项目依赖 ├── main.py # 应用主入口 ├── core/ # 核心逻辑 │ ├── __init__.py │ ├── state.py # 定义状态State类 │ ├── graph.py # 定义工作流图Graph │ └── nodes.py # 定义所有节点Node函数 ├── tools/ # 工具定义 │ ├── __init__.py │ ├── calculator.py # 示例计算器工具 │ └── enterprise_tools.py # 模拟的企业内部工具如查询订单 ├── knowledge/ # RAG知识库相关 │ ├── __init__.py │ ├── vector_store.py # 向量数据库初始化与操作 │ └── docs/ # 存放待处理的原始文档PDF/TXT等 └── config.py # 配置文件第三步配置环境变量在项目根目录创建.env文件并填入你的OpenAI API Key。切记将此文件加入.gitignore不要提交到代码仓库。# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here在config.py中读取配置# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) # 其他配置如模型名称、向量数据库路径等 MODEL_NAME gpt-4o # 或 gpt-4-turbo PERSIST_DIRECTORY ./chroma_db # 向量数据库持久化目录4. 定义工作流的核心状态State状态是LangGraph工作流的“记忆体”。我们使用Pydantic的BaseModel来严格定义状态的结构这能带来良好的类型提示和验证。# core/state.py from typing import TypedDict, List, Optional, Any from pydantic import BaseModel, Field class AgentState(TypedDict): LangGraph 工作流的共享状态。 使用TypedDict是为了与LangGraph的接口更好兼容。 # 用户输入 input: str # 对话历史 chat_history: List[str] # LLM的完整思考过程chain of thought reasoning: Optional[str] # 当前需要执行的动作由LLM决定 next_action: Optional[str] # 工具调用的结果 tool_output: Optional[str] # RAG检索到的文档内容 retrieved_docs: Optional[List[str]] # 最终输出给用户的结果 final_output: Optional[str] # 其他自定义上下文如用户ID、会话ID等 context: Optional[dict] class StrictAgentState(BaseModel): 使用Pydantic BaseModel定义的严格状态用于内部数据验证和传递。 在实际节点函数中我们可能使用这个类来处理数据。 input: str Field(description用户的最新输入) chat_history: List[str] Field(default_factorylist, description对话历史列表格式为[user: xxx, assistant: yyy]) reasoning: Optional[str] Field(defaultNone, descriptionLLM的推理过程) next_action: Optional[str] Field(defaultNone, description下一步动作如 call_tool:get_weather, retrieve_knowledge, respond) tool_output: Optional[str] Field(defaultNone, description工具调用的返回结果) retrieved_docs: Optional[List[str]] Field(defaultNone, description检索到的相关文档列表) final_output: Optional[str] Field(defaultNone, description最终回复内容) context: dict Field(default_factorydict, description扩展上下文信息) class Config: arbitrary_types_allowed True这个状态定义包含了Agent运行所需的所有信息。注意next_action字段它将由LLM来填充用于指导工作流的下一个节点这是实现动态路由的关键。5. 构建工具Tools与知识库RAGAgent的能力边界由它的工具和知识决定。我们先构建两个基础能力。5.1 实现一个简单的工具我们先创建一个计算器工具和一个模拟的“查询订单状态”工具。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行一个数学表达式计算。支持加减乘除-*/和乘方**。 例如calculator(3 5 * 2) 返回 13。 try: # 警告在生产环境中直接eval是危险的此处仅作演示。 # 真实场景应使用更安全的表达式解析库如 ast.literal_eval 配合限制。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # tools/enterprise_tools.py from langchain.tools import tool import random from datetime import datetime tool def get_order_status(order_id: str) - str: 根据订单ID查询订单状态。这是一个模拟的企业内部系统接口。 参数: order_id: 订单编号例如 ORD-2024-001 返回: 订单状态的描述字符串。 # 模拟数据库查询 status_options [已创建, 支付中, 已发货, 配送中, 已完成, 已取消] # 为了演示让状态看起来有点规律 status_index hash(order_id) % len(status_options) estimated_delivery (datetime.now().day 3) % 30 return f订单 {order_id} 当前状态为【{status_options[status_index]}】。预计送达日期本月{estimated_delivery}日。 tool def escalate_to_human(reason: str, user_id: str) - str: 将当前会话升级至人工客服。这是一个关键的安全与合规工具。 参数: reason: 升级原因例如 用户情绪激动 或 问题超出知识范围。 user_id: 用户标识。 返回: 告知用户已转接人工的提示并返回一个工单号。 ticket_id fTICKET-{datetime.now().strftime(%Y%m%d)}-{random.randint(1000,9999)} # 在实际系统中这里会调用工单系统API创建工单并通知客服人员。 return f您的问题已转接人工客服工单号{ticket_id}。客服人员将尽快与您联系。转接原因{reason}5.2 构建RAG知识库我们使用ChromaDB作为向量数据库并加载一些示例文档。# knowledge/vector_store.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader import os from config import OPENAI_API_KEY, PERSIST_DIRECTORY class KnowledgeBase: def __init__(self, persist_directory: str PERSIST_DIRECTORY): self.embeddings OpenAIEmbeddings(openai_api_keyOPENAI_API_KEY) self.persist_directory persist_directory 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.embeddings ) else: print(未找到已有向量库将创建新库。) # 可以在这里初始化一个空的VectorStore或者调用ingest_documents方法 self.vector_store None def ingest_documents(self, document_paths: list): 将文档切片并存入向量数据库。 参数: document_paths: 文档路径列表。 all_docs [] text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段约500字符 chunk_overlap50 # 片段间重叠50字符保证上下文连贯 ) for path in document_paths: if not os.path.exists(path): print(f警告文件 {path} 不存在跳过。) continue try: loader TextLoader(path, encodingutf-8) documents loader.load() splits text_splitter.split_documents(documents) all_docs.extend(splits) print(f已处理文档: {path}, 得到 {len(splits)} 个片段。) except Exception as e: print(f处理文档 {path} 时出错: {e}) if all_docs: print(f开始创建向量库共 {len(all_docs)} 个文本片段...) self.vector_store Chroma.from_documents( documentsall_docs, embeddingself.embeddings, persist_directoryself.persIST_DIRECTORY ) self.vector_store.persist() print(f向量库已创建并持久化到 {self.persist_directory}) else: print(没有有效的文档可供处理。) def retrieve(self, query: str, k: int 3) - list: 检索与查询最相关的k个文档片段。 参数: query: 查询文本。 k: 返回的片段数量。 返回: 相关文档片段的文本内容列表。 if self.vector_store is None: print(向量库未初始化请先调用 ingest_documents。) return [] try: docs self.vector_store.similarity_search(query, kk) return [doc.page_content for doc in docs] except Exception as e: print(f检索过程中出错: {e}) return [] # 示例创建知识库并摄入文档 if __name__ __main__: kb KnowledgeBase() # 假设在 knowledge/docs/ 目录下有一些.txt文件 doc_paths [./knowledge/docs/company_policy.txt, ./knowledge/docs/product_faq.txt] kb.ingest_documents(doc_paths) # 测试检索 results kb.retrieve(公司的年假政策是什么) for i, r in enumerate(results): print(f结果 {i1}: {r[:100]}...)你需要提前在knowledge/docs/目录下准备一些文本文件作为知识库。例如company_policy.txt内容可以是公司员工手册摘要 1. 年假政策入职满一年后享有10天年假之后每增加一年工龄年假增加1天上限15天。 2. 报销流程员工需在费用发生后的30天内通过内部OA系统提交电子发票和审批单。 3. 考勤制度标准工作时间为周一至周五9:00-18:00。弹性工作制需部门领导批准。6. 定义工作流节点Nodes节点是图的基本执行单元。每个节点是一个函数它接收当前State执行一些操作如调用LLM、运行工具然后返回更新后的State。# core/nodes.py from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List import json from core.state import AgentState, StrictAgentState from tools.calculator import calculator from tools.enterprise_tools import get_order_status, escalate_to_human from knowledge.vector_store import KnowledgeBase from config import OPENAI_API_KEY, MODEL_NAME # 初始化LLM和知识库 llm ChatOpenAI(modelMODEL_NAME, openai_api_keyOPENAI_API_KEY, temperature0) kb KnowledgeBase() # 注意这里假设知识库已初始化并有数据 def route_input(state: AgentState) - AgentState: 路由节点分析用户输入决定工作流走向。 这是第一个节点负责意图识别。 print(f[路由节点] 收到输入: {state[input]}) messages [ SystemMessage(content你是一个智能路由助手。请分析用户意图并只返回一个JSON对象包含next_action字段。), HumanMessage(contentf用户说{state[input]}。\n\n可选动作\n1. need_calculation - 如果涉及数学计算。\n2. query_order - 如果询问订单状态。\n3. general_qa - 如果是一般性问答或公司政策咨询。\n4. escalate - 如果用户表达强烈不满或问题复杂。\n\n请只返回JSON如 {{\next_action\: \general_qa\}}) ] response llm.invoke(messages) try: decision json.loads(response.content) state[next_action] decision.get(next_action, general_qa) state[reasoning] f路由分析用户意图被分类为 {state[next_action]} except json.JSONDecodeError: state[next_action] general_qa state[reasoning] f路由分析LLM返回格式错误默认路由到 general_qa。响应{response.content} return state def call_llm_for_decision(state: AgentState) - AgentState: 核心决策节点根据当前状态和next_action调用LLM决定具体执行什么。 这个节点是工作流的“大脑”。 print(f[决策节点] 当前动作指令: {state.get(next_action)}) # 构建系统提示词根据不同的next_action给予不同的指令 system_prompt 你是一个企业级AI助手。请根据对话历史、用户当前输入和已有的工具调用结果或检索到的知识决定下一步做什么并生成相应的内容。 if state.get(next_action) need_calculation: system_prompt 用户需要计算。请分析其输入提取出数学表达式并准备调用计算器工具。 elif state.get(next_action) query_order: system_prompt 用户需要查询订单。请从其输入中提取出订单号格式类似ORD-XXXX-XXX并准备调用订单查询工具。 elif state.get(next_action) general_qa: system_prompt 用户在进行一般性问答。请结合检索到的公司知识库内容给出准确、友好的回答。 else: system_prompt 请综合所有信息生成对用户的回复。 # 构建对话历史 chat_history_text \n.join(state.get(chat_history, [])[-4:]) # 取最近4轮对话 user_input state[input] # 构建检索到的知识上下文 knowledge_context if state.get(retrieved_docs): knowledge_context \n\n【相关公司知识】\n \n---\n.join(state[retrieved_docs][:2]) # 取前2个片段 # 构建工具调用结果上下文 tool_context if state.get(tool_output): tool_context f\n\n【工具执行结果】\n{state[tool_output]} full_prompt f{system_prompt}\n\n对话历史\n{chat_history_text}\n\n用户最新输入{user_input}{knowledge_context}{tool_context}\n\n请决定下一步1. 如果需要调用工具请说明调用哪个工具以及参数。2. 如果可以直接回答请生成回复内容。 messages [HumanMessage(contentfull_prompt)] response llm.invoke(messages) # 这里简化处理实际应该让LLM以结构化格式如JSON返回决策。 # 为了演示我们简单地将LLM的回复存入状态由后续节点解析。 state[reasoning] response.content return state def execute_tool(state: AgentState) - AgentState: 工具执行节点解析LLM的决策调用相应的工具。 reasoning state.get(reasoning, ) print(f[工具执行节点] 解析决策: {reasoning[:100]}...) # 这是一个简化的工具调用逻辑。在实际项目中你应该使用LangChain的Tool Calling功能 # 让LLM直接返回结构化工具调用请求。这里我们做简单的关键字匹配。 tool_output None if 计算 in reasoning or calculator in reasoning.lower(): # 尝试提取表达式这里非常简化 # 实际应用应使用更鲁棒的提取方法或直接使用LLM的tool_calling能力 import re expr_match re.search(r(\d[\\-\*/]\d), state[input]) if expr_match: expr expr_match.group(1) tool_output calculator.invoke(expr) else: tool_output 无法从输入中提取有效的数学表达式。 elif 订单 in reasoning or order in reasoning.lower(): # 尝试提取订单号 import re order_id_match re.search(rORD-\w-\d, state[input]) if order_id_match: order_id order_id_match.group(0) tool_output get_order_status.invoke(order_id) else: tool_output 未在输入中找到有效的订单号格式如ORD-2024-001。 elif 人工 in reasoning or escalate in reasoning.lower(): tool_output escalate_to_human.invoke(用户请求或问题复杂, user_123) if tool_output: state[tool_output] tool_output print(f[工具执行节点] 工具调用结果: {tool_output}) else: state[tool_output] 未触发任何工具调用。 return state def retrieve_knowledge(state: AgentState) - AgentState: 知识检索节点从向量数据库中检索与用户问题相关的文档。 if state.get(next_action) general_qa: query state[input] print(f[知识检索节点] 检索查询: {query}) docs kb.retrieve(query, k2) state[retrieved_docs] docs print(f[知识检索节点] 检索到 {len(docs)} 条相关文档。) else: state[retrieved_docs] [] return state def generate_final_response(state: AgentState) - AgentState: 最终响应生成节点综合所有信息生成给用户的最终回复。 print([响应生成节点] 生成最终回复...) # 这里可以再次调用LLM利用所有上下文生成更连贯的回复。 # 为了简化我们直接拼接信息。 final_parts [] if state.get(tool_output) and 未触发 not in state[tool_output]: final_parts.append(f根据系统查询{state[tool_output]}) if state.get(retrieved_docs): # 简单取第一条检索结果作为回答依据实际应更智能地整合 if state[retrieved_docs]: final_parts.append(f根据公司规定{state[retrieved_docs][0][:150]}...) if not final_parts: final_parts.append(我已收到您的请求。) state[final_output] .join(final_parts) # 更新对话历史 state[chat_history].append(fuser: {state[input]}) state[chat_history].append(fassistant: {state[final_output]}) return state def human_intervention(state: AgentState) - AgentState: 人工干预节点模拟人工客服接入。 在实际系统中这里会触发通知、创建工单等。 print([人工干预节点] 触发人工客服流程...) state[final_output] 您的问题已转接给人工客服请稍候。 # 这里可以添加调用工单系统的逻辑 return state7. 组装工作流图Graph并运行现在我们将所有节点连接起来形成一个完整的工作流。# core/graph.py from langgraph.graph import StateGraph, END from core.state import AgentState from core.nodes import route_input, call_llm_for_decision, execute_tool, retrieve_knowledge, generate_final_response, human_intervention def create_agent_workflow(): 创建并返回配置好的工作流图 workflow StateGraph(AgentState) # 1. 添加节点 workflow.add_node(router, route_input) # 路由 workflow.add_node(decision, call_llm_for_decision) # 决策 workflow.add_node(retrieve, retrieve_knowledge) # 检索 workflow.add_node(execute_tool, execute_tool) # 执行工具 workflow.add_node(generate_response, generate_final_response) # 生成回复 workflow.add_node(human_intervention, human_intervention) # 人工干预 # 2. 设置入口点 workflow.set_entry_point(router) # 3. 定义边条件流转 # 从路由节点到决策节点 workflow.add_edge(router, decision) # 从决策节点出来根据情况分流 def decide_after_decision(state: AgentState) - str: 决策节点后的条件判断 reasoning state.get(reasoning, ).lower() if 人工 in reasoning or escalate in reasoning: return human_intervention elif 计算 in reasoning or 订单 in reasoning: return execute_tool elif 知识 in reasoning or general_qa in state.get(next_action, ): return retrieve else: return generate_response workflow.add_conditional_edges( decision, decide_after_decision, { human_intervention: human_intervention, execute_tool: execute_tool, retrieve: retrieve, generate_response: generate_response, } ) # 4. 定义其他节点的流转 # 工具执行后去生成回复 workflow.add_edge(execute_tool, generate_response) # 知识检索后去生成回复 workflow.add_edge(retrieve, generate_response) # 人工干预后直接结束 workflow.add_edge(human_intervention, END) # 生成回复后结束本轮 workflow.add_edge(generate_response, END) # 5. 编译图 app workflow.compile() return app # 主程序入口 if __name__ __main__: from core.graph import create_agent_workflow app create_agent_workflow() # 测试1数学计算 print(\n 测试1数学计算 ) initial_state {input: 帮我算一下 125 37 等于多少, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试2订单查询 print(\n 测试2订单查询 ) initial_state {input: 我的订单 ORD-2024-1001 到哪里了, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试3知识库问答 print(\n 测试3知识库问答 ) initial_state {input: 我们公司的年假有多少天, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]}) # 测试4触发人工 print(\n 测试4触发人工 ) initial_state {input: 我要投诉你们的产品有问题, chat_history: []} result app.invoke(initial_state) print(f最终回复: {result[final_output]})8. 运行结果与效果验证运行python core/graph.py你应该能看到类似以下的输出 测试1数学计算 [路由节点] 收到输入: 帮我算一下 125 37 等于多少 [决策节点] 当前动作指令: need_calculation [工具执行节点] 解析决策: 用户需要计算。请分析其输入提取出数学表达式并准备调用计算器工具。... [工具执行节点] 工具调用结果: 计算结果: 162 [响应生成节点] 生成最终回复... 最终回复: 根据系统查询计算结果: 162 测试2订单查询 [路由节点] 收到输入: 我的订单 ORD-2024-1001 到哪里了 [决策节点] 当前动作指令: query_order [工具执行节点] 解析决策: 用户需要查询订单。请从其输入中提取出订单号格式类似ORD-XXXX-XXX并准备调用订单查询工具。... [工具执行节点] 工具调用结果: 订单 ORD-2024-1001 当前状态为【已发货】。预计送达日期本月15日。 [响应生成节点] 生成最终回复... 最终回复: 根据系统查询订单 ORD-2024-1001 当前状态为【已发货】。预计送达日期本月15日。 测试3知识库问答 [路由节点] 收到输入: 我们公司的年假有多少天 [决策节点] 当前动作指令: general_qa [知识检索节点] 检索查询: 我们公司的年假有多少天 [知识检索节点] 检索到 2 条相关文档。 [响应生成节点] 生成最终回复... 最终回复: 根据公司规定公司员工手册摘要1. 年假政策入职满一年后享有10天年假之后每增加一年工龄年假增加1天上限15天。... 测试4触发人工 [路由节点] 收到输入: 我要投诉你们的产品有问题 [决策节点] 当前动作指令: escalate [人工干预节点] 触发人工客服流程... 最终回复: 您的问题已转接给人工客服请稍候。如何验证成功流程正确针对不同类型的输入工作流正确地路由到了不同的分支计算、查询、检索、人工。工具调用成功计算器和订单查询工具被正确调用并返回了结果。RAG生效对于政策类问题成功从向量库中检索到了相关片段。状态流转State对象在各个节点间正确传递和更新最终生成了包含上下文的回复。9. 常见问题与排查思路在实际部署中你可能会遇到以下问题问题现象可能原因排查方式解决方案启动时报ModuleNotFoundError依赖未安装或虚拟环境未激活1. 运行pip list | grep langchain检查。2. 确认终端路径在项目目录下。1. 激活虚拟环境。2. 运行pip install -r requirements.txt。OpenAI API 调用失败API Key 错误或网络问题1. 检查.env文件格式和内容。2. 在Python中直接调用openai.ChatCompletion.create测试。1. 确保.env文件在根目录且变量名正确。2. 检查网络连接和API配额。向量数据库检索无结果1. 文档未成功摄入。2. 查询语句与文档语义不匹配。1. 检查ingest_documents时的日志和chroma_db目录大小。2. 直接运行kb.retrieve(“test”)看是否有基础结果。1. 确认文档路径正确且内容为纯文本。2. 尝试调整chunk_size或使用不同的嵌入模型。工作流卡在某个节点无输出节点函数逻辑错误或返回状态格式不对。1. 在每个节点函数开始添加print语句。2. 检查节点函数返回值是否为字典且包含必需的State键。1. 使用pdb或ipdb进行调试。2. 确保所有节点函数都接收并返回AgentState类型。LLM 的决策不符合预期提示词Prompt设计不佳。1. 打印出call_llm_for_decision节点中的full_prompt。2. 在OpenAI Playground中手动测试提示词。1. 优化系统提示词给出更明确的指令和示例。2. 考虑使用LangChain的StructuredOutputParser来约束LLM输出格式。工具调用参数提取错误在execute_tool节点中使用了简单的正则匹配不鲁棒。观察reasoning字段的内容看LLM是否给出了清晰的工具调用指令。最佳实践使用LangChain的bind_tools和with_structured_output功能让LLM直接返回结构化的工具调用请求。10. 企业级最佳实践与工程建议将上述Demo推向企业生产环境你需要考虑以下关键点1. 状态管理的持久化问题当前State存在于内存中服务重启后对话历史丢失。方案将AgentState与一个唯一的session_id绑定并存入Redis或数据库。每个请求根据session_id加载历史状态。2. 工具调用的安全与审计问题工具可能执行高危操作如删除数据、发送消息。方案权限控制为每个工具定义权限等级并在调用前校验用户/会话权限。参数校验与净化对所有输入参数进行严格的类型、范围和内容校验。操作审计记录每一次工具调用的时间、用户、参数和结果便于追溯。模拟模式在生产环境外运行一个“模拟环境”Agent可以调用工具但不会产生真实副作用。3. RAG的优化问题简单的向量检索可能返回不相关或碎片化信息。方案混合检索结合向量检索语义相似和关键词检索BM25提升召回率。重排序Re-ranking使用更精细的模型对检索结果进行重排序提升精度。元数据过滤为文档片段添加元数据如部门、有效期检索时进行过滤。查询改写在检索前先用LLM对用户原始查询进行扩展或改写。4. 工作流的可观测性与调试问题Agent的决策过程是黑盒出错难以排查。方案全链路日志在每个节点的入口和出口记录详细的日志包括输入、输出、耗时。可视化利用LangGraph Studio如果可用或自定义图表来可视化工作流的执行路径。追踪Tracing集成像LangSmith这样的平台记录每一次LLM调用、工具调用和节点流转。5. 版本控制与灰度发布问题直接更新Agent的工作流或提示词可能影响线上业务。方案配置化将提示词、工具列表、路由规则等抽取为配置文件。A/B测试对新旧版本的Agent进行流量切分对比关键指标如解决率、用户满意度。回滚机制确保能快速回退到上一个稳定版本。6. 与现有系统集成问题Agent需要调用大量的内部系统API。方案API网关通过统一的API网关来管理和调用内部服务Agent工具层只需调用网关。认证传递安全地处理用户身份将其传递给下游系统如使用JWT Token。异步处理对于耗时的工具调用如生成报告采用异步任务队列如Celery避免阻塞Agent的响应。构建企业级AI Agent是一个系统工程它考验的不仅是你对LLM和框架的掌握更是你对软件工程、系统架构和安全合规的综合理解。本文提供的代码是一个坚实的起点但真正的挑战和价值在于你如何根据自己企业的独特需求在这个骨架上填充血肉——设计更精细的状态、开发更强大的工具、构建更准确的知识库并最终将它无缝、安全、可靠地融入到你现有的业务血脉之中。