基于LangGraph构建生产级AI Agent：从客服工单处理实战到部署优化

1. 项目概述：为什么“生产级”是AI Agent的分水岭

最近和不少同行交流，发现大家聊起AI Agent，已经从“这东西挺酷”变成了“这东西怎么才能用起来”。确实，从去年开始，各种Agent框架和平台如雨后春笋般冒出来，LangChain、LangGraph、Dify、Claude Agent SDK……名字都听麻了。但当你真的想动手，把一个Demo级别的Agent变成能稳定运行、处理真实业务的生产级应用时，会发现这中间隔着一道巨大的鸿沟。Demo可以只关心“能不能跑通”，而生产级则要面对稳定性、性能、成本、可观测性等一系列现实拷问。

今天，我就以一个过来人的身份，结合我最近用LangGraph搭建的一个客服工单自动分类与路由Agent的实战经历，手把手带你走完从技术选型到最终上线的完整流程。这个项目不大，但麻雀虽小五脏俱全，它要求Agent能理解用户提交的工单内容，自动判断问题类型（如“账号问题”、“支付故障”、“产品咨询”），并分派给对应的处理队列，同时还要能处理用户的追问，维护简单的对话状态。我们不会停留在“Hello World”，而是会深入到部署、监控、优化这些真正决定项目成败的环节。无论你是刚入门想找个靠谱的起点，还是已经踩过一些坑想系统化提升，这篇指南应该都能给你一些直接的参考。

2. 核心需求解析与方案选型

在动手写第一行代码之前，我们必须把需求掰开揉碎了看。我那个客服工单Agent的需求听起来简单，但拆解后会发现不少隐藏的挑战。

2.1 需求拆解：从业务场景到技术清单

首先，业务需求是：用户通过一个Web界面提交文字工单，Agent需要自动完成分类和路由。这引申出几个核心功能点：

1. 意图识别与分类：准确理解用户工单的核心诉求，归到预设的类别。
2. 状态管理与多轮对话：用户可能会追问“我刚才提交的支付问题处理到哪一步了？”，Agent需要能关联上下文。
3. 工具调用与外部集成：分类后，需要调用内部API，将工单数据写入对应的业务系统队列。
4. 稳定性与可观测性：作为生产服务，必须保证高可用，并且能清晰地看到每一次请求的链路、耗时、LLM调用详情，方便排查问题。

基于这些，我们的技术选型清单就清晰了：需要一个能编排复杂工作流、管理状态、方便集成工具、且具备良好可观测性基础的框架。

2.2 框架横评：LangChain、LangGraph、Dify与Claude SDK

市面上主流的选项就那几个，我们来逐一分析，看哪个最适合这个“生产级”场景。

• LangChain：这无疑是生态最繁荣的“瑞士军刀”。它的链（Chain）和代理（Agent）抽象非常经典，社区工具和集成多如牛毛。但是，对于需要复杂状态循环和精细控制流的应用，原生的Agent抽象有时会显得力不从心。它的执行过程更像一个黑盒，调试和追踪每一步的中间状态比较麻烦。对于我们的多轮对话和严格的工作流，它可能不是最优雅的解决方案。

• LangGraph：你可以把它理解为LangChain的“工作流引擎”升级版。它引入了图（Graph）的概念，将Agent的执行过程明确定义为由节点（Node）和边（Edge）组成的有向图。这带来了几个巨大优势：

  • 显式状态管理：整个Agent的运行状态（State）是一个贯穿始终的、可自定义的Pydantic模型，状态变化一目了然。
  • 清晰的控制流：通过边（条件判断）来控制下一个执行节点，循环、分支、并行等逻辑变得极其直观。
  • 强大的可观测性：由于执行路径是确定的图，配合LangSmith等工具，可以非常清晰地可视化每一步的执行过程、输入输出和耗时，这对生产调试至关重要。
  • 子图（Subgraph）支持：可以将复杂流程模块化，比如把“用户身份验证”抽成一个子图，多处复用，保持代码整洁。 对于我们的工单Agent，分类、路由、状态更新、等待用户输入，正好可以建模成一个清晰的图结构。因此，LangGraph是我们的核心框架选择。

• Dify：这是一个优秀的低代码/无代码AI应用平台。它通过可视化工作流编排，大大降低了构建AI应用的门槛。如果你追求极致的开发速度，且业务逻辑不涉及大量自定义代码和复杂集成，Dify是很好的选择。但是，它的缺点也在于“平台化”：自定义能力受限于平台提供的节点，深度集成内部系统可能需要绕弯子；部署和运维依赖平台方；对于需要极致性能调优和复杂控制逻辑的场景，可能会遇到天花板。我们的项目需要紧密耦合内部工单API，且对可控性要求高，因此Dify更适合作为前期原型验证工具，而非最终生产框架。

• Claude Agent SDK：这是Anthropic为其Claude模型量身打造的Agent框架。如果你坚定地使用Claude系列模型，并且希望获得与模型特性深度优化的开发体验，它是一个不错的选择。但其生态和灵活性目前与LangChain/LangGraph还有差距，且将你绑定在了特定的模型提供商上。从生产级的长期维护和灵活性考虑，我们暂时不将其作为主选。

结论：我们选择LangGraph作为核心编排框架，利用其强大的状态管理和可视化工作流能力。同时，我们会利用LangChain生态中丰富的工具集成（如计算器、搜索引擎封装等）作为补充。底层LLM，为了平衡效果、成本和速度，生产环境可以选择GPT-4o或Claude 3 Haiku，开发调试阶段可以用DeepSeek或通义千问等性价比较高的模型。

3. 环境搭建与核心概念初始化

工欲善其事，必先利其器。生产级项目的第一步，就是建立一个稳定、可复现的工程环境。

3.1 工程化环境配置

别再只用pip install了，生产环境需要精确的依赖管理。

# 创建项目目录 mkdir production-ai-agent && cd production-ai-agent # 创建虚拟环境（推荐使用uv或conda，这里用uv示例，速度极快） uv venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # 使用 uv 初始化项目并安装核心依赖 uv init uv add langgraph langchain-openai langchain-community pydantic python-dotenv uv add fastapi uvicorn httpx # 用于构建API服务 uv add langsmith # 用于可观测性（强烈推荐） uv add pytest pytest-asyncio # 用于测试 # 生成 requirements.txt 以备部署 uv pip compile pyproject.toml -o requirements.txt

关键依赖说明：

• langgraph：核心工作流引擎。
• langchain-openai：OpenAI模型集成。
• langchain-community：包含大量社区工具。
• pydantic：用于定义强类型的State模型，这是LangGraph的基石。
• langsmith：LangChain官方出的可观测性平台，能记录每次链、Agent的调用轨迹，对调试和生产监控不可或缺。它有免费额度。

接下来，创建项目结构：

production-ai-agent/ ├── .env # 环境变量（API Keys等） ├── .gitignore ├── pyproject.toml # 项目依赖声明 ├── src/ │ ├── __init__.py │ ├── agent/ # Agent核心逻辑 │ │ ├── __init__.py │ │ ├── state.py # 定义State │ │ ├── nodes.py # 定义各个节点函数 │ │ ├── graph.py # 构建并编译Graph │ │ └── tools.py # 自定义工具 │ ├── api/ # FastAPI应用 │ │ ├── __init__.py │ │ └── main.py │ └── config.py # 配置管理 ├── tests/ # 测试用例 └── scripts/ # 部署或辅助脚本

3.2 理解LangGraph的核心：State与Graph

这是LangGraph最精髓的两个概念，必须吃透。

State：它定义了你的Agent在整个工作流中需要记住和传递的所有信息。它必须是一个PydanticBaseModel。对于我们的工单Agent，State可能长这样：

# src/agent/state.py from typing import Annotated, Literal, Optional from typing_extensions import TypedDict import operator from pydantic import BaseModel, Field from datetime import datetime class AgentState(BaseModel): """Agent的完整状态，贯穿工作流始终。""" # 用户输入 user_input: str = Field(description="用户当前轮次的输入内容") # 对话历史（简化版，生产环境可能需要更复杂的结构） conversation_history: Annotated[list, operator.add] = Field( default_factory=list, description="对话历史记录，每轮为一个字典，包含角色和内容" ) # Agent的分析结果 intent: Optional[str] = Field(default=None, description="识别的用户意图类别") confidence: Optional[float] = Field(default=None, description="意图识别的置信度") # 系统执行结果 ticket_id: Optional[str] = Field(default=None, description="创建的工单ID") assigned_queue: Optional[str] = Field(default=None, description="分配的工单队列") # 控制流标志 requires_human: bool = Field(default=False, description="是否需要人工介入") error_message: Optional[str] = Field(default=None, description="错误信息") # 元数据 created_at: datetime = Field(default_factory=datetime.now)

注意conversation_history字段的Annotated[list, operator.add]，这是LangGraph的归约器（Reducer），它告诉框架在每次循环中，如何更新这个列表（这里是追加）。这是实现多轮对话记忆的关键。

Graph与Node：Graph是由Node（节点）和Edge（边）组成的。Node是一个普通的Python函数（或异步函数），它接收当前的State，执行一些操作（如调用LLM、调用工具），然后返回一个State的更新（字典）。Edge决定了基于更新后的State，下一个该执行哪个Node。

4. 构建生产级工单处理Agent

现在，我们开始用代码将设计落地。

4.1 定义工具（Tools）：Agent的手和脚

工具是Agent与外部世界交互的方式。我们先定义两个核心工具：一个用于调用内部工单系统API，一个用于查询知识库（模拟）。

# src/agent/tools.py import httpx from typing import Dict, Any from langchain.tools import tool from langchain_core.tools import ToolException import logging logger = logging.getLogger(__name__) class InternalTicketSystem: """模拟内部工单系统客户端。生产环境需替换为真实SDK或API调用。""" def __init__(self, base_url: str, api_key: str): self.client = httpx.AsyncClient(base_url=base_url, headers={"X-API-Key": api_key}) async def create_ticket(self, title: str, description: str, category: str, user_info: Dict) -> Dict[str, Any]: """在内部系统创建工单。""" payload = { "title": title, "description": description, "category": category, "submitter": user_info } try: resp = await self.client.post("/api/v1/tickets", json=payload, timeout=30.0) resp.raise_for_status() data = resp.json() logger.info(f"工单创建成功: ID={data.get('id')}") return {"success": True, "ticket_id": data.get("id"), "queue": data.get("assignedQueue")} except httpx.HTTPStatusError as e: logger.error(f"创建工单HTTP错误: {e}") raise ToolException(f"工单系统创建失败: {e.response.text}") except Exception as e: logger.error(f"创建工单未知错误: {e}") raise ToolException(f"工单系统请求异常: {str(e)}") async def close(self): await self.client.aclose() # 使用LangChain的@tool装饰器将其暴露给Agent @tool async def create_service_ticket(ticket_title: str, ticket_description: str, problem_category: str) -> str: """ 在内部工单系统中创建一个新的服务请求工单。 请确保problem_category是以下之一：['billing', 'technical', 'account', 'general_inquiry']。 Args: ticket_title: 工单的简短标题。 ticket_description: 问题的详细描述。 problem_category: 问题分类。 """ # 这里简化处理，实际应从State或上下文中获取用户信息 user_info = {"user_id": "extracted_from_context"} # 初始化客户端（生产环境应从依赖注入或全局配置获取） # 此处为示例，直接模拟返回 # async with InternalTicketSystem("https://internal.example.com", "key") as its: # result = await its.create_ticket(ticket_title, ticket_description, problem_category, user_info) # return f"工单创建成功！工单号：{result['ticket_id']}，已分配至{result['queue']}队列。" return f"[模拟]工单创建成功！标题：{ticket_title}，分类：{problem_category}，工单号：T-20240527-001。" @tool async def search_knowledge_base(query: str) -> str: """在公司内部知识库中搜索相关解决方案或文档。""" # 模拟搜索过程 simulated_results = [ f"找到关于'{query}'的文档：如何重置账户密码。", f"相关文章：'{query}'常见问题解答。" ] return "\n".join(simulated_results[:2]) # 返回前两条

注意：工具函数的文档字符串（Docstring）至关重要！LLM（尤其是GPT-4）依赖这些描述来决定何时以及如何调用工具。描述要清晰、准确，并说明参数格式。

4.2 实现节点（Nodes）：工作流中的每一步

节点是工作流中的具体执行单元。我们将工单处理流程分解为几个节点。

# src/agent/nodes.py from typing import Dict, Any from langgraph.graph import StateGraph, START, END from .state import AgentState from .tools import create_service_ticket, search_knowledge_base from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage import logging import json logger = logging.getLogger(__name__) # 初始化LLM。生产环境应从配置读取模型名和API Key。 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, streaming=False) async def classify_intent_node(state: AgentState) -> Dict[str, Any]: """节点1：分析用户输入，识别意图和分类。""" logger.info(f"开始意图分类，用户输入: {state.user_input[:100]}...") # 构建系统提示词，这是影响效果的关键！ system_prompt = SystemMessage(content="""你是一个专业的客服工单分类助手。请严格根据用户输入，判断其意图属于以下哪一类： 1. `report_issue` - 用户报告一个具体的问题或故障（如无法登录、支付失败、页面错误）。 2. `ask_question` - 用户提出一个咨询类问题（如如何操作、资费说明、功能咨询）。 3. `request_service` - 用户明确请求一项服务（如开通权限、重置密码、注销账户）。 4. `other` - 不属于以上任何一类。 同时，如果属于`report_issue`或`request_service`，请进一步判断问题类别： - `billing` (账单/支付) - `technical` (技术/故障) - `account` (账户/登录) - `general_inquiry` (普通咨询) 请以纯JSON格式回复，包含两个字段：`intent` (意图) 和 `category` (类别，仅当意图为1或3时提供，否则为null)。不要有任何额外解释。""") user_message = HumanMessage(content=state.user_input) try: response = await llm.ainvoke([system_prompt, user_message]) result_text = response.content.strip() # 安全地解析JSON result = json.loads(result_text) intent = result.get("intent", "other") category = result.get("category") logger.info(f"意图分类结果: intent={intent}, category={category}") # 更新状态 update = { "intent": intent, "assigned_queue": category # 这里简单映射，实际可能更复杂 } # 将本轮交互加入历史 new_history_entry = {"role": "user", "content": state.user_input} new_ai_entry = {"role": "assistant", "content": f"分析意图: {intent}, 类别: {category}"} return {"conversation_history": [new_history_entry, new_ai_entry], **update} except json.JSONDecodeError as e: logger.error(f"LLM返回的JSON解析失败: {result_text}, 错误: {e}") return {"intent": "other", "error_message": f"意图分析失败: {e}"} except Exception as e: logger.error(f"意图分类节点执行异常: {e}") return {"intent": "other", "error_message": str(e), "requires_human": True} async def handle_question_node(state: AgentState) -> Dict[str, Any]: """节点2：处理咨询类问题，尝试从知识库寻找答案。""" if state.intent != "ask_question": # 如果不是咨询类，直接传递状态，不执行操作 return {} logger.info(f"处理咨询问题: {state.user_input}") # 调用知识库搜索工具 kb_result = await search_knowledge_base(state.user_input) # 让LLM基于知识库结果生成友好回复 prompt = f"""用户问题：{state.user_input} 从知识库中找到的相关信息： {kb_result} 请根据以上信息，用友好、专业的口吻直接回答用户的问题。如果知识库信息不足，请如实告知并建议其提交工单。""" ai_response = await llm.ainvoke([HumanMessage(content=prompt)]) reply_content = ai_response.content update = { "conversation_history": [{"role": "assistant", "content": reply_content}] } return update async def create_ticket_node(state: AgentState) -> Dict[str, Any]: """节点3：对于需要创建工单的意图，调用工具创建工单。""" if state.intent not in ["report_issue", "request_service"]: return {} if not state.assigned_queue: return {"error_message": "无法确定工单类别", "requires_human": True} logger.info(f"开始创建工单，类别: {state.assigned_queue}") # 这里可以构建更详细的工单描述，例如结合对话历史 ticket_title = f"{state.intent}: {state.user_input[:50]}..." ticket_description = f"用户描述：{state.user_input}\n\n分析意图：{state.intent}\n分配队列：{state.assigned_queue}" try: # 调用工具 tool_result = await create_service_ticket.ainvoke({ "ticket_title": ticket_title, "ticket_description": ticket_description, "problem_category": state.assigned_queue }) update = { "ticket_id": "T-SIMULATED-001", # 应从工具返回结果中提取 "conversation_history": [{"role": "assistant", "content": tool_result}] } return update except Exception as e: logger.error(f"创建工单失败: {e}") return {"error_message": f"创建工单时出错: {e}", "requires_human": True} async def finalize_response_node(state: AgentState) -> Dict[str, Any]: """节点4：整理最终回复给用户。""" # 这里可以根据state中的ticket_id, assigned_queue等信息，生成最终总结性消息 if state.ticket_id: final_msg = f"您的问题已处理完毕。工单号：{state.ticket_id}，已转至{state.assigned_queue}团队跟进。您可以通过工单号查询进度。" elif state.error_message: final_msg = f"处理过程中遇到问题：{state.error_message}。已为您转接人工客服。" # 更新状态，标志需要人工介入 return {"conversation_history": [{"role": "assistant", "content": final_msg}], "requires_human": True} else: # 对于咨询类，回复已在handle_question_node生成，这里可能不需要额外动作 # 或者可以生成一个结束语 final_msg = "请问还有其他可以帮您的吗？" return {"conversation_history": [{"role": "assistant", "content": final_msg}]}

4.3 组装工作流图（Graph）：编排节点与边

这是LangGraph最直观的部分，像画流程图一样把节点连接起来。

# src/agent/graph.py from langgraph.graph import StateGraph, START, END from .state import AgentState from .nodes import classify_intent_node, handle_question_node, create_ticket_node, finalize_response_node import logging logger = logging.getLogger(__name__) def route_after_classify(state: AgentState) -> str: """路由函数：根据意图分类结果，决定下一个节点。""" intent = state.intent if intent == "ask_question": return "handle_question" elif intent in ["report_issue", "request_service"]: return "create_ticket" else: return "finalize_response" # 其他意图或分类失败，直接结束 def check_for_human_intervention(state: AgentState) -> str: """路由函数：检查是否需要人工介入。""" if state.requires_human: logger.warning("流程标记为需要人工介入。") return END # 直接结束，或跳转到人工交接节点 return "__end__" # 继续默认流程 def create_agent_graph() -> StateGraph: """创建并编译工单处理Agent的工作流图。""" # 1. 创建图构建器，并指定状态模式 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node("classify_intent", classify_intent_node) workflow.add_node("handle_question", handle_question_node) workflow.add_node("create_ticket", create_ticket_node) workflow.add_node("finalize_response", finalize_response_node) # 3. 添加边，定义控制流 workflow.add_edge(START, "classify_intent") # 从 classify_intent 出来，根据路由函数决定去向 workflow.add_conditional_edges( "classify_intent", route_after_classify, { "handle_question": "handle_question", "create_ticket": "create_ticket", "finalize_response": "finalize_response", } ) # handle_question 和 create_ticket 节点执行后，都走向 finalize_response workflow.add_edge("handle_question", "finalize_response") workflow.add_edge("create_ticket", "finalize_response") # 在最终响应前，检查是否需要人工介入 workflow.add_conditional_edges( "finalize_response", check_for_human_intervention, {END: END, "__end__": END} # 这里简化，实际可能有人工节点 ) # 设置最终节点 workflow.add_edge("finalize_response", END) # 4. 编译图 compiled_graph = workflow.compile() logger.info("工单处理Agent图编译完成。") return compiled_graph # 全局可用的图实例 agent_graph = create_agent_graph()

现在，一个具备完整工作流的Agent就构建好了。你可以通过agent_graph.invoke({"user_input": "我的账户登录不上了，提示密码错误"})来测试它。它会自动执行分类->创建工单->生成回复的流程。

5. 部署、监控与性能优化

一个能在本地运行的Agent离“生产级”还差得远。接下来是关键环节：让它成为一个可靠的服务。

5.1 使用FastAPI构建RESTful API

我们需要一个标准接口供前端或其他服务调用。

# src/api/main.py from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from contextlib import asynccontextmanager import logging from src.agent.graph import agent_graph from src.agent.state import AgentState import os from langsmith import Client from dotenv import load_dotenv load_dotenv() # 配置LangSmith追踪（非必须但强烈推荐） os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com" os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGCHAIN_API_KEY") os.environ["LANGCHAIN_PROJECT"] = "production-ticket-agent" # 你的项目名 logger = logging.getLogger(__name__) class ChatRequest(BaseModel): message: str session_id: str | None = None # 用于支持多轮对话会话 class ChatResponse(BaseModel): reply: str session_id: str requires_human: bool ticket_id: str | None = None # 简单的内存会话存储（生产环境请使用Redis、数据库等） session_store = {} @asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑，如初始化数据库连接池 logger.info("AI Agent API 启动中...") yield # 关闭逻辑 logger.info("AI Agent API 关闭中...") # 如有需要，可关闭工具中的HTTP客户端等 app = FastAPI(title="工单处理AI Agent API", lifespan=lifespan) @app.post("/chat", response_model=ChatResponse) async def chat_with_agent(request: ChatRequest): """与工单处理Agent对话的主端点。""" try: # 1. 获取或初始化会话状态 session_id = request.session_id or f"sess_{os.urandom(4).hex()}" current_state = session_store.get(session_id, AgentState(user_input="")) # 2. 更新状态中的用户输入 inputs = {"user_input": request.message} # 可以在这里合并历史状态，LangGraph的invoke会处理Reducer # 为了简单演示，我们每次重新初始化，生产环境需持久化完整State # 更佳实践是将整个State序列化后存入session_store # 3. 调用编译好的图执行工作流 config = {"configurable": {"thread_id": session_id}} # 用于LangSmith追踪 final_state = await agent_graph.ainvoke(inputs, config=config) # 4. 从最终状态提取回复 # 取对话历史中最后一条AI回复 history = final_state.get("conversation_history", []) last_ai_msg = next( (msg["content"] for msg in reversed(history) if msg["role"] == "assistant"), "抱歉，我未能生成回复。" ) # 5. 更新会话存储（简化处理，实际应存储整个State） session_store[session_id] = final_state # 6. 返回响应 return ChatResponse( reply=last_ai_msg, session_id=session_id, requires_human=final_state.get("requires_human", False), ticket_id=final_state.get("ticket_id") ) except Exception as e: logger.exception(f"处理聊天请求时发生未预期错误: {e}") raise HTTPException(status_code=500, detail="内部服务器错误") @app.get("/health") async def health_check(): """健康检查端点。""" return {"status": "healthy"}

使用Uvicorn运行：uvicorn src.api.main:app --host 0.0.0.0 --port 8000 --reload

5.2 集成可观测性：LangSmith实战

没有可观测性的AI应用就是“盲人摸象”。LangSmith能记录每一次LLM调用、工具调用和图的执行路径。

1. 注册并获取API Key：前往 LangSmith官网 注册。
2. 配置环境变量：如上文代码所示，设置LANGCHAIN_TRACING_V2、LANGCHAIN_API_KEY等。
3. 查看追踪：启动你的Agent并发送请求后，在LangSmith控制台即可看到详细的追踪记录。你可以看到：
   • 每个节点的输入输出。
   • LLM调用的具体提示词（Prompt）和补全（Completion）。
   • 工具调用的参数和结果。
   • 整个工作流的耗时瀑布图。
   • 这对于调试“为什么Agent做出了这个决策”和性能优化至关重要。

5.3 性能优化与成本控制要点

生产环境必须关注这两点。

• 缓存：对LLM结果进行缓存可以大幅减少重复调用和成本。使用langchain.cache（如SQLiteCache,RedisCache）。

  from langchain.globals import set_llm_cache from langchain.cache import SQLiteCache set_llm_cache(SQLiteCache(database_path=".langchain.db"))

• 超时与重试：为LLM和工具调用设置合理的超时和重试策略，避免单个慢请求拖垮整个服务。langchain-openai和httpx都支持配置。
• 模型降级：非关键路径或简单任务，使用更小、更快的模型（如gpt-4o-minivsgpt-4o）。
• 异步化：确保你的节点函数、工具调用都是异步的（async/await），并使用异步HTTP客户端（如httpx.AsyncClient），以支持高并发。
• 批量处理：如果场景允许，将多个用户请求聚合后批量调用LLM（需要模型支持），能显著降低成本。

6. 测试、迭代与常见问题排查

6.1 编写自动化测试

测试AI应用有其特殊性，因为输出是非确定性的。我们的策略是：

1. 单元测试节点函数：Mock掉LLM和工具，测试业务逻辑。
2. 集成测试工作流：使用固定的测试用例和Mock的LLM响应，验证整个图的执行路径是否符合预期。
3. 评估测试（Evaluation）：使用LangSmith的评估功能，针对一批真实或构造的测试用例，评估Agent输出的相关性、准确性和安全性。

# tests/test_agent.py import pytest from unittest.mock import AsyncMock, patch from src.agent.nodes import classify_intent_node from src.agent.state import AgentState @pytest.mark.asyncio async def test_classify_intent_node_report_issue(): """测试分类节点对报障类输入的处理。""" # 模拟LLM返回固定的JSON mock_response = AsyncMock() mock_response.content = '{"intent": "report_issue", "category": "technical"}' with patch('src.agent.nodes.llm.ainvoke', return_value=mock_response): initial_state = AgentState(user_input="网站突然打不开了，显示500错误。") result = await classify_intent_node(initial_state) assert result["intent"] == "report_issue" assert result["assigned_queue"] == "technical" assert len(result["conversation_history"]) == 2

6.2 常见问题与排查清单

在实际开发和运维中，你肯定会遇到下面这些问题：

• 问题1：Agent总是调用错误的工具，或参数不对。

  • 排查：首先去LangSmith查看LLM接收到工具定义（tool.description）的提示词部分是否清晰。检查工具的描述文档是否准确描述了功能和参数。尝试在系统提示词中更明确地指导LLM何时使用工具。
  • 技巧：给工具起一个清晰、动词开头的名字（如search_knowledge_base而非kb_search）。在工具描述中明确写出调用示例。

• 问题2：图执行陷入循环，或停在某个节点不动。

  • 排查：检查add_conditional_edges中的路由函数。确保所有可能的状态分支都有对应的目标节点。在路由函数中增加详细的日志，打印出判断条件。
  • 技巧：使用workflow.get_graph().draw_mermaid()输出图的Mermaid代码，在线渲染成流程图，直观检查逻辑。

• 问题3：多轮对话中，状态（如对话历史）混乱或丢失。

  • 排查：确认State中用于存储历史的字段是否正确使用了Annotated[list, operator.add]。检查每次invoke时是否正确地传递和恢复了完整的上一轮State。会话存储（如session_store）的实现是否正确。
  • 技巧：在开发初期，可以将完整的State在API响应中也返回给前端（调试用），便于观察状态变化。

• 问题4：性能瓶颈，API响应慢。

  • 排查：利用LangSmith的追踪时间线，找出耗时最长的节点。通常是LLM调用或外部工具调用（如网络请求）。
  • 优化：为LLM调用启用缓存。优化工具调用的网络连接（使用连接池、设置超时）。考虑将非必要的同步操作改为异步。

• 问题5：LLM输出格式不符合要求，无法解析。

  • 排查：这是提示词工程问题。在系统提示词中严格要求输出格式（如“请以纯JSON格式回复，包含xx字段”）。使用response_format参数（如果模型支持，如OpenAI的JSON模式）。在代码中添加更健壮的解析逻辑和fallback机制。

构建生产级AI Agent是一个系统工程，它远不止是调用API。从清晰的架构设计（LangGraph），到工程化的环境与部署，再到不可或缺的可观测性（LangSmith）和测试评估，每一步都关乎最终应用的稳定性和可用性。这套以LangGraph为核心的方案，提供了所需的控制力、透明度和扩展性。希望这个从零到一的实战指南，能帮你避开我当初踩过的那些坑，更顺畅地将你的AI想法落地为真正可用的服务。