十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

基于LangGraph与MCP构建可观测企业级AI Agent架构

基于LangGraph与MCP构建可观测企业级AI Agent架构 面试官问“你做过 AI Agent 项目吗讲讲你的设计。” 你心里一紧是不是只能回答“用 LangChain 调了调 API做了个聊天机器人”然后看着面试官礼貌但失望的眼神。别慌。今天要聊的不是一个玩具项目而是一个能真正体现你工程化思维、可观测性设计和解决实际业务痛点的企业级 AI Agent 项目架构。它的核心不是“让 AI 回答问题”而是让 AI 的执行过程变得透明、可控、可评估。这正是当前 AI 应用从 Demo 走向生产的关键瓶颈也是面试中能让你脱颖而出的高价值话题。本文将带你手把手构建一个基于LangGraph和MCPModel Context Protocol的 AI Agent 系统重点实现全链路追踪Tracing、多维度评估Evaluation和系统可观测Observability。你会得到一套可以直接用于面试项目阐述甚至稍作修改就能落地到真实业务中的代码框架和设计思路。我们不止讲“是什么”更会深入“为什么重要”、“解决了什么工程问题”以及“如何避开那些坑”。1. 这篇文章真正要解决的问题从玩具 Demo 到生产级 Agent 的鸿沟很多开发者学习 AI Agent 的第一个项目往往是跟着教程用 LangChain 或 LlamaIndex 快速拼凑一个能联网搜索、能总结文档的聊天应用。这很好但它只是一个“玩具”。当你想把它部署到线上服务真实用户时一系列棘手问题会瞬间涌现黑盒与失控Agent 内部究竟调用了哪些工具思考步骤Reasoning是怎样的为什么最终给出了这个答案一旦出错你几乎无从排查。效果无法量化这个 Agent 的“好坏”全凭感觉。回答得流畅就好吗有没有“一本正经地胡说八道”如何用数据证明它的效果在提升成本不可控一次复杂的任务可能调用多次大模型和外部 API成本是多少哪些步骤最耗 Token有没有优化空间技能难以复用与扩展新加一个工具如查询数据库、调用内部系统是否需要重写大量胶水代码如何优雅地管理越来越多的 Agent 技能本文要解决的正是这些工程化问题。我们将使用LangGraph来构建具备复杂状态流转和循环能力的 Agent用MCP协议来标准化、解耦地管理各种工具技能并在此基础上搭建一套完整的追踪、评估与可观测体系。最终你将拥有一个不仅“能跑”而且“看得清”、“管得住”、“测得准”的 AI Agent 系统。2. 核心概念与架构选型为什么是 LangGraph MCP在深入代码之前必须理解我们为什么选择这个技术栈。这不仅是工具介绍更是架构思想的体现。2.1 LangGraph超越链Chain的图Graph思维LangChain的核心是“链”Chain它适合线性、确定性的任务流。但对于需要根据中间结果做判断、循环、分支的复杂 Agent链就显得力不从心。LangGraph是 LangChain 团队推出的用于构建有状态、多环节 Agent 的框架。它将工作流抽象为“图”Graph节点Node代表一个执行步骤如调用 LLM、执行工具边Edge代表状态流转的条件。关键优势显式状态管理所有步骤共享并更新一个中央状态State数据流转清晰。支持循环与条件分支轻松实现“思考-行动-观察”的 ReAct 循环或根据结果选择不同路径。天然支持并发可以定义并行执行的节点。可视化与调试LangGraph 自带可视化工具能直观看到执行路径这对调试复杂 Agent 至关重要。简单类比如果说 LangChain 是组装一条流水线那么 LangGraph 就是设计一张交通路线图车辆状态可以根据路况条件选择不同的道路边到达各个站点节点。2.2 MCP (Model Context Protocol)工具管理的“统一插座”问题传统方式中工具Tool的逻辑硬编码在 Agent 代码里。每增加一个新工具如查天气、查股票、操作数据库都要修改 Agent 核心代码并处理复杂的认证、参数解析等问题。MCP是由 Anthropic 提出并逐渐成为行业事实标准的一个协议。它旨在将工具或更广义的“上下文”的提供与 AI 模型的使用进行解耦。核心思想工具以独立的MCP Server形式存在通过标准协议HTTP/SSE暴露其能力。AI 应用如我们的 LangGraph Agent作为MCP Client通过协议动态发现、描述并调用这些工具。关键优势解耦与复用工具开发独立于 Agent 开发。一个查询数据库的 MCP Server 可以被公司内所有 AI 项目复用。标准化统一的工具发现、调用和错误处理机制。安全与权限可以在 MCP Server 层面实现精细的权限控制和审计。生态丰富社区已经提供了大量开源的 MCP Server用于文件系统、数据库、Git、JIRA 等。简单类比MCP 就像电脑的 USB 协议。你不需要为了读U盘而去修改电脑主板Agent只需要找一个符合 USB 协议MCP的U盘MCP Server插上电脑就能自动识别并使用它。2.3 追踪Tracing、评估Evaluation与可观测性Observability这是本项目的“灵魂”也是区分玩具与生产系统的关键。追踪Tracing记录 Agent 执行全生命周期的详细日志。包括每次 LLM 调用的输入/输出、每次工具调用的参数/结果、每个状态节点的流转。这解决了“黑盒”问题是调试和优化的基础。我们将使用 LangSmithLangChain 官方平台来实现。评估Evaluation量化 Agent 的表现。不仅仅是最终答案的对错还包括过程评估推理步骤是否合理、工具使用评估调用工具是否必要且正确、成本评估Token 消耗、延迟评估等。我们将设计一套评估体系并利用追踪数据自动执行评估。可观测性Observability在追踪和评估的基础上构建仪表盘Dashboard和告警Alerting让研发和运维人员能够实时洞察系统健康度、性能瓶颈和异常情况。我们的架构LangGraph作为 Agent 的“大脑”和“调度中心”MCP作为标准化、可插拔的“四肢”工具集而追踪与评估体系则是覆盖整个系统的“神经系统”和“体检中心”。3. 环境准备与项目初始化我们开始动手。请确保你的开发环境满足以下要求Python 版本: 3.10 或以上推荐 3.11包管理工具: pip 或 poetry关键 API 密钥:OpenAI API Key或其他兼容 OpenAI API 的模型服务如 Azure OpenAI, Together.ai, 本地部署的 Ollama 等LangSmith API Key用于追踪和评估这是 LangChain 的官方平台提供免费额度3.1 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir enterprise-ai-agent cd enterprise-ai-agent python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate创建requirements.txt文件内容如下# 核心框架 langgraph0.0.52 langchain-core0.1.0 langchain-openai0.0.5 langchain-mcp-adapters0.0.3 # 用于连接MCP Server # MCP 协议相关 (以SQLite示例Server为例) mcp[cli]1.2.0 # 安装一个社区MCP Server示例sqlite # 通常MCP Server会单独发布这里我们演示如何运行一个Server # 追踪与评估 langsmith0.1.0 # 工具与工具包 python-dotenv1.0.0 # 管理环境变量 pydantic2.5.0 # 数据验证安装依赖pip install -r requirements.txt3.2 配置环境变量创建.env文件存放你的敏感信息。切记不要将此文件提交到版本控制系统如Git。# .env OPENAI_API_KEYsk-your-openai-api-key-here LANGCHAIN_API_KEYls-your-langsmith-api-key-here LANGCHAIN_TRACING_V2true LANGCHAIN_PROJECTEnterprise-AI-Agent-Demo # 在LangSmith中创建的项目名 LANGCHAIN_ENDPOINThttps://api.smith.langchain.com在代码中使用dotenv加载配置# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGCHAIN_API_KEY os.getenv(LANGCHAIN_API_KEY) assert OPENAI_API_KEY, OPENAI_API_KEY 未设置 assert LANGCHAIN_API_KEY, LANGCHAIN_API_KEY 未设置4. 构建核心基于 LangGraph 的 Agent我们将构建一个具备 ReActReasoning Acting能力的 Agent。它的任务是理解用户问题决定是否需要调用工具并综合所有信息给出最终答案。4.1 定义 Agent 状态State状态是 LangGraph 中贯穿整个执行流程的数据容器。我们使用TypedDict来定义。# agent/state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): Agent 的全局状态定义 # 消息历史LangGraph 内置的合并操作符 add_messages 会处理它 messages: Annotated[List, add_messages] # 用户当前的问题 question: str # Agent 的“思考”过程供调试和评估用 reasoning_steps: List[str] # 已调用工具的结果列表 tool_results: List[dict] # 最终答案 final_answer: str | NoneAnnotated和add_messages是 LangGraph 的语法糖用于自动合并消息列表非常方便。4.2 创建工具Tools并集成 MCP Client传统方式是将工具函数直接绑定到 LLM。我们将演示如何集成一个 MCP Client让它动态发现和使用工具。首先我们创建一个简单的本地工具作为备选然后连接 MCP。# agent/tools.py from langchain.tools import tool from datetime import datetime tool def get_current_time(query: str) - str: 当用户询问当前时间或日期时调用此工具。query 是用户关于时间的问题。 now datetime.now() return f当前时间是{now.strftime(%Y-%m-%d %H:%M:%S)} # 注意更复杂的工具如查询数据库、调用API我们将通过 MCP 来集成。接下来我们初始化 MCP Client。这里假设我们已经有一个在本地运行的 MCP Server例如一个提供 SQLite 查询能力的 Server。如何运行 MCP Server 将在下一节详述。# agent/mcp_client.py import subprocess import time from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def create_mcp_session(server_command: list[str]) - ClientSession: 创建并连接到一个 MCP Server server_params StdioServerParameters(commandserver_command) stdio_transport await stdio_client(server_params) session ClientSession(*stdio_transport) await session.initialize() return session async def get_mcp_tools(session: ClientSession) - list[dict[str, Any]]: 从 MCP Server 获取工具列表 # 调用 MCP 标准接口 list_tools response await session.list_tools() # 将 MCP 工具描述转换为 LangChain Tool 格式 tools [] for tool_info in response.tools: # 这里需要根据 MCP 返回的格式做适配转换 # 简化示例假设 tool_info 有 name, description, inputSchema langchain_tool { name: tool_info.name, description: tool_info.description, args_schema: ... # 根据 inputSchema 创建 Pydantic 模型 } tools.append(langchain_tool) return tools4.3 定义 LangGraph 节点Nodes和边Edges这是 Agent 的逻辑核心。我们创建几个关键节点reasoning_node: 让 LLM 分析问题决定下一步是调用工具还是直接回答。tool_node: 执行被选中的工具。final_answer_node: 生成最终答案。# agent/graph.py from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from .state import AgentState from .tools import get_current_time import json # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用成本更低的模型进行推理 # 绑定工具这里先绑定本地工具后续会替换为从MCP动态获取的工具 tools [get_current_time] llm_with_tools llm.bind_tools(tools) def reasoning_node(state: AgentState) - dict: 推理节点分析问题决定行动 messages state[messages] question state[question] # 构造给 LLM 的提示 system_prompt 你是一个有帮助的AI助手。请仔细思考用户的问题。 如果你有合适的工具可以获取更准确的信息请调用工具。 如果你已经拥有足够的信息来回答问题请直接给出答案。 prompt [{role: system, content: system_prompt}, {role: user, content: question}] # 调用 LLM response llm_with_tools.invoke(prompt) # 记录推理步骤 reasoning_steps state.get(reasoning_steps, []) reasoning_steps.append(f思考: {response.content}) # 检查 LLM 是否想调用工具 tool_calls response.tool_calls if hasattr(response, tool_calls) else [] new_state { messages: messages [response], reasoning_steps: reasoning_steps, } if tool_calls: # 下一步去执行工具 new_state[next] call_tool new_state[tool_calls] tool_calls else: # 下一步去生成最终答案 new_state[next] final_answer new_state[draft_answer] response.content return new_state def tool_node(state: AgentState) - dict: 工具执行节点 tool_calls state[tool_calls] tool_results state.get(tool_results, []) for tc in tool_calls: tool_name tc[name] tool_args tc[args] # 在实际项目中这里应该是一个工具名称到可调用对象的映射 # 我们简化处理只处理我们已知的工具 if tool_name get_current_time: result get_current_time.invoke(json.dumps(tool_args)) else: result f错误未知工具 {tool_name} # 记录结果 tool_results.append({ tool_name: tool_name, args: tool_args, result: result }) # 构造 ToolMessage 供 LLM 理解 tool_message ToolMessage(contentstr(result), tool_call_idtc.get(id, )) state[messages].append(tool_message) new_state { tool_results: tool_results, next: reasoning # 执行完工具后回到推理节点进行下一步分析 } return new_state def final_answer_node(state: AgentState) - dict: 最终答案生成节点 draft_answer state.get(draft_answer, ) question state[question] # 这里可以引入一个“总结”LLM调用整合所有工具结果和历史消息生成最终答案。 # 为了简化我们直接使用草稿答案或基于所有信息再生成一次。 final_messages state[messages] # 添加一个系统提示要求生成友好、准确的最终答案 final_prompt final_messages [HumanMessage(contentf基于以上对话和信息请直接给出问题{question}的最终答案。)] final_response llm.invoke(final_prompt) new_state { final_answer: final_response.content, next: END # 指向结束 } return new_state现在我们用这些节点构建图Graph# agent/graph.py (续) def create_agent_graph(): 创建并编译 Agent 工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(reasoning, reasoning_node) workflow.add_node(call_tool, tool_node) workflow.add_node(final_answer, final_answer_node) # 设置入口点 workflow.set_entry_point(reasoning) # 添加边根据状态中的 next 字段决定流向 workflow.add_conditional_edges( reasoning, # 这个函数根据 state 返回下一个节点的名称 lambda state: state.get(next, final_answer), { call_tool: call_tool, final_answer: final_answer } ) workflow.add_edge(call_tool, reasoning) # 工具执行完回到推理 workflow.add_edge(final_answer, END) # 编译图 app workflow.compile() return app # 创建图应用实例 agent_app create_agent_graph()5. 集成 MCP Server以 SQLite 查询为例让我们的 Agent 拥有查询数据库的能力。我们将使用一个社区提供的 SQLite MCP Server。5.1 启动 MCP Server首先确保你安装了mcpCLI。然后我们可以通过 Python 脚本或命令行启动一个 Server。这里我们使用mcp包内置的示例 Server或者使用社区的sqliteserver。假设我们有一个chinook.db示例数据库。创建一个mcp_servers/sqlite_server.py脚本# mcp_servers/sqlite_server.py import asyncio from mcp.server import Server, StdioServerParameters from mcp.server.models import Tool, TextContent import sqlite3 import json from pathlib import Path # 创建一个简单的 SQLite 工具 Server server Server(sqlite-server) server.list_tools() async def handle_list_tools() - list[Tool]: return [ Tool( namequery_sqlite, description执行一个 SQLite 查询语句。参数 sql 必须是合法的 SQL SELECT 语句。, inputSchema{ type: object, properties: { sql: {type: string, description: 要执行的 SQL SELECT 语句} }, required: [sql] } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[TextContent]: if name query_sqlite: db_path Path(./chinook.db) # 你的数据库文件路径 if not db_path.exists(): return [TextContent(typetext, textf错误数据库文件 {db_path} 不存在)] sql arguments.get(sql, ) if not sql.strip().upper().startswith(SELECT): return [TextContent(typetext, text错误只允许执行 SELECT 语句)] try: conn sqlite3.connect(str(db_path)) conn.row_factory sqlite3.Row # 返回字典样式的行 cursor conn.cursor() cursor.execute(sql) rows cursor.fetchall() conn.close() # 将结果格式化为易读的文本 if rows: columns [description[0] for description in cursor.description] result_text | .join(columns) \n -*50 \n for row in rows: result_text | .join(str(row[col]) for col in columns) \n result_text f\n共 {len(rows)} 行。 else: result_text 查询成功但未返回任何数据。 return [TextContent(typetext, textresult_text)] except sqlite3.Error as e: return [TextContent(typetext, textfSQLite 错误: {e})] else: return [TextContent(typetext, textf未知工具: {name})] async def main(): async with server.run_stdio_server() as (read_stream, write_stream): # 保持 Server 运行 await asyncio.Future() if __name__ __main__: asyncio.run(main())运行这个 Serverpython mcp_servers/sqlite_server.py5.2 修改 Agent 以动态加载 MCP 工具现在我们需要修改之前的agent/graph.py在初始化时连接 MCP Server 并获取工具然后绑定给 LLM。# agent/graph.py (修改部分) import asyncio from .mcp_client import create_mcp_session, get_mcp_tools async def initialize_agent_with_mcp(): 异步初始化 Agent连接 MCP Server # 1. 连接 MCP Server # 假设我们的 SQLite Server 通过 stdio 运行命令是 python mcp_servers/sqlite_server.py server_command [python, mcp_servers/sqlite_server.py] session await create_mcp_session(server_command) # 2. 获取 MCP 工具并转换为 LangChain Tool 格式 # 注意这是一个复杂的过程需要适配 MCP 工具描述到 LangChain Tool 对象。 # 这里我们进行大幅简化假设我们已经转换好了一个工具列表 mcp_tools_list。 # 在实际项目中你需要编写适配层代码。 from langchain.tools import StructuredTool from pydantic import BaseModel, Field # 示例手动创建一个基于 MCP 的 Tool 包装器 class QuerySQLiteInput(BaseModel): sql: str Field(description要执行的 SQL SELECT 语句) async def query_sqlite_tool(sql: str) - str: 调用 MCP Server 执行 SQL 查询 # 通过 MCP session 调用工具 result await session.call_tool(query_sqlite, arguments{sql: sql}) # 提取文本内容 text_content for content in result.content: if content.type text: text_content content.text return text_content mcp_tool StructuredTool.from_function( funcquery_sqlite_tool, namequery_sqlite, description执行一个 SQLite 查询语句。参数 sql 必须是合法的 SQL SELECT 语句。, args_schemaQuerySQLiteInput ) # 3. 组合工具本地工具 MCP 工具 all_tools [get_current_time, mcp_tool] # 4. 创建绑定工具的 LLM llm_with_tools llm.bind_tools(all_tools) # 5. 创建图需要修改 tool_node 以支持新工具 # ... (修改 tool_node 函数能根据名称调用对应的工具) return agent_app, session # 返回 app 和 session 以便后续关闭 # 注意由于涉及异步运行入口需要调整。通常我们会用一个主异步函数来运行。6. 实现追踪Tracing与评估Evaluation有了可运行的 Agent我们现在给它装上“监控系统”。6.1 集成 LangSmith 进行自动追踪LangChain/LangGraph 已经原生集成了 LangSmith。我们之前设置的环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY就是为了这个。只要使用 LangChain 的组件如ChatOpenAI,bind_tools和 LangGraph调用过程会自动被记录到 LangSmith。现在运行你的 Agent然后去 LangSmith 官网 查看你的项目Enterprise-AI-Agent-Demo你就能看到每一次运行的详细追踪信息包括每个节点的输入输出、耗时、Token 使用情况等。6.2 设计并实现评估体系Evaluation追踪记录了“发生了什么”评估则要判断“做得好不好”。我们设计几个评估维度最终答案相关性Answer Relevance最终答案是否直接回答了用户问题工具调用合理性Tool Use Appropriateness调用的工具是否必要参数是否正确推理过程连贯性Reasoning Coherence思考步骤是否逻辑清晰成本与延迟Cost Latency本次调用消耗了多少 Token总耗时多少我们将编写评估函数并利用 LangSmith 的评估功能来批量运行和打分。# evaluation/evaluators.py from langsmith.evaluation import evaluate, EvaluationResult from langsmith.schemas import Example, Run import asyncio def answer_relevance_evaluator(run: Run, example: Example) - EvaluationResult: 评估最终答案的相关性 # run.outputs 包含了 Agent 运行的最终输出 final_answer run.outputs.get(final_answer, ) # example.outputs 是期望的输出如果有的话这里我们主要看输入问题 question example.inputs.get(question, ) # 简单逻辑如果答案长度太短或不包含问题关键词则相关性低 # 在实际项目中这里应该调用一个 LLM 作为评估器LLM-as-a-judge score 0.0 feedback if len(final_answer) 5: score 0.2 feedback 答案过短可能未充分回答问题。 elif any(keyword in question.lower() for keyword in [时间, 日期]) and (时间 in final_answer or 日期 in final_answer): score 0.9 feedback 答案明确包含了时间信息相关性高。 else: # 更复杂的评估可以在这里添加 score 0.7 feedback 答案基本相关。 return EvaluationResult(keyanswer_relevance, scorescore, commentfeedback) def tool_use_evaluator(run: Run, example: Example) - EvaluationResult: 评估工具使用的合理性 # 从 run 的中间步骤中提取工具调用信息 tool_calls [] for step in run.child_runs or []: if step.name tool_node or step.run_type tool: tool_calls.append(step) if not tool_calls: # 没有调用工具评估是否应该调用 question example.inputs.get(question, ) if any(keyword in question.lower() for keyword in [查询, 数据库, select]): score 0.3 feedback 问题可能涉及数据查询但未调用工具。 else: score 0.9 feedback 无需调用工具判断合理。 else: # 调用了工具评估工具选择和参数 score 0.8 feedback f调用了 {len(tool_calls)} 次工具。 # 可以进一步分析具体调用了什么工具参数是否正确 return EvaluationResult(keytool_use_appropriateness, scorescore, commentfeedback) # 可以定义更多的评估器...然后我们可以使用 LangSmith 的evaluate函数在一个测试数据集上运行我们的 Agent 并自动评估。# evaluation/run_evaluation.py from langsmith import Client from agent.graph import agent_app # 导入我们编译好的 Agent from evaluation.evaluators import answer_relevance_evaluator, tool_use_evaluator client Client() # 1. 创建测试数据集示例 dataset_name agent-qa-test # 如果数据集不存在则创建 try: dataset client.read_dataset(dataset_namedataset_name) except: dataset client.create_dataset(dataset_namedataset_name) # 添加一些测试用例 examples [ (现在几点了, 应该调用 get_current_time 工具并返回时间。), (数据库里有多少首歌曲, 应该调用 query_sqlite 工具执行 SELECT COUNT(*) FROM tracks;), (你好请介绍一下你自己。, 无需调用工具直接回答。), ] for input_q, reference_output in examples: client.create_example( inputs{question: input_q}, outputs{expected: reference_output}, dataset_iddataset.id ) # 2. 定义一个预测函数predict function def predict_agent(inputs: dict) - dict: 包装我们的 Agent使其符合 LangSmith evaluate 的接口 question inputs[question] # 注意这里需要根据你的 Agent 输入格式调整 # 假设我们的 Agent 的输入状态是 {messages: [HumanMessage(...)], question: question} from langchain_core.messages import HumanMessage initial_state { messages: [HumanMessage(contentquestion)], question: question, reasoning_steps: [], tool_results: [], final_answer: None } # 同步运行 Agent如果 Agent 是异步的需要 asyncio.run final_state agent_app.invoke(initial_state) return {final_answer: final_state.get(final_answer, No answer generated.)} # 3. 运行评估 experiment_results evaluate( predict_agent, datadataset_name, evaluators[answer_relevance_evaluator, tool_use_evaluator], experiment_prefixmy-agent-eval-v1, # metadata{version: 1.0}, # num_repetitions3, # 可以设置重复次数以评估稳定性 ) print(f评估完成查看结果{experiment_results[project_url]})运行此脚本后你可以在 LangSmith 的对应项目中看到一个详细的评估实验包含每个测试用例的得分和反馈。7. 构建可观测仪表盘与常见问题排查7.1 利用 LangSmith 仪表盘LangSmith 本身就是一个强大的可观测平台。你可以查看追踪列表了解所有历史运行。筛选与搜索按时间、标签、状态成功/错误筛选。深入单个追踪查看完整的执行图、每个节点的输入输出、Token 使用、延迟。查看评估结果在“Evaluations”标签页查看整体评估指标和每个案例的详情。设置告警企业版功能当错误率上升或延迟超标时收到通知。7.2 常见问题与排查思路问题现象可能原因排查方式解决方案Agent 不调用工具直接回答1. LLM 提示词System Prompt未明确要求使用工具。2. 工具描述不够清晰LLM 不理解何时使用。3. 模型能力不足如使用了gpt-3.5-turbo。1. 检查 LangSmith 中reasoning_node的输入输出看 LLM 的思考过程。2. 检查工具绑定时描述是否准确传递给了 LLM。1. 优化 System Prompt加入工具使用范例。2. 精炼工具描述明确使用场景。3. 升级到更强的模型如gpt-4,gpt-4o。MCP Server 连接失败1. Server 命令路径错误。2. Server 进程崩溃。3. 端口/stdio 通信故障。1. 检查server_command是否正确。2. 单独运行 MCP Server 脚本看是否有报错。3. 查看客户端连接时的异常日志。1. 使用绝对路径。2. 在 MCP Server 代码中添加更详细的日志和错误处理。3. 考虑使用 HTTP 而非 stdio 通信如果 Server 支持。LangSmith 无追踪数据1. 环境变量LANGCHAIN_TRACING_V2未设置为true。2.LANGCHAIN_API_KEY无效或未设置。3. 网络问题。1. 确认.env文件已加载变量已设置。2. 在 LangSmith 官网检查 API Key 状态。3. 尝试在代码开头import langsmith; print(langsmith.__version__)并调用一个简单测试。1. 确保环境变量正确。2. 在代码中显式配置langsmith.Client(api_key...)。3. 检查代理或防火墙设置。工具调用参数错误1. LLM 生成的参数格式与工具期望的不符。2. Pydantic 模型验证失败。1. 在 LangSmith 中查看tool_node的输入检查tool_calls内容。2. 查看工具函数的错误日志。1. 在工具描述中更严格地定义参数格式和示例。2. 在tool_node中添加参数清洗和转换逻辑。图陷入无限循环1. 条件边Conditional Edge的逻辑有误导致在reasoning和call_tool间死循环。2. LLM 反复要求调用同一个工具。1. 在 LangSmith 追踪图中观察循环路径。2. 检查reasoning_node返回的next值。1. 在状态中添加循环计数器达到阈值后强制跳转到final_answer。2. 在提示词中要求 LLM 避免重复操作。8. 最佳实践与工程建议将上述组件组合成一个健壮的生产系统还需要考虑以下几点配置化管理将模型类型、温度、MCP Server 地址、数据库连接串等全部抽取到配置文件如config.yaml或环境变量中。错误处理与回退在每个节点尤其是工具调用和 LLM 调用添加try...except并设计优雅的回退策略例如工具调用失败后让 LLM 基于已有知识回答。状态持久化对于长对话或需要记忆的 Agent需要将AgentState持久化到数据库如 Redis并在每次请求时加载。异步优化MCP 调用、LLM 调用、数据库查询都是 I/O 密集型操作应使用异步async/await来提高并发性能。确保你的整个调用链路是异步的。安全与权限MCP Server 权限每个 MCP Server 应运行在最小权限下。数据库查询 Server 应使用只读账号。输入验证与清理对用户输入和 LLM 生成的工具参数进行严格的验证和清理防止 SQL 注入等攻击。成本限制在 Agent 层面设置最大 Token 消耗或最大循环次数防止恶意或异常查询导致巨额费用。版本控制对 Agent 的提示词、图结构、工具集进行版本控制。LangSmith 可以关联 Git Commit便于追踪变更对效果的影响。评估常态化建立自动化的评估流水线。每当 Agent 代码或提示词更新时自动在测试数据集上运行评估确保核心指标答案相关性、工具使用合理率没有下降。9. 总结与项目延伸方向通过本文我们完成了一个从零到一、具备企业级特征的 AI Agent 项目。它不再是简单的提示词拼接而是一个拥有清晰工作流LangGraph、标准化技能接入MCP、全方位可观测性LangSmith 追踪与评估的复杂系统。回顾核心价值对面试官你可以清晰地阐述 Agent 的状态机设计、工具解耦思想MCP、以及如何保障系统的可观测性与可评估性。这远超“我调过 API”的层面。对实际项目这套架构为 AI Agent 的迭代优化提供了数据基础靠追踪为效果保障建立了标准靠评估为功能扩展降低了成本靠 MCP。你可以继续深化的方向实现更复杂的图引入并行节点如同时查询多个数据源、子图将复杂功能模块化、人工审核节点。集成更多 MCP Server探索社区接入 GitHub、JIRA、Confluence、内部 CRM 等系统的 MCP Server快速赋予 Agent 新能力。构建自定义评估器使用更强大的 LLM如 GPT-4作为裁判LLM-as-a-judge对答案的事实性、安全性、有帮助性进行深度评估。开发管理界面基于追踪和评估数据构建一个内部仪表盘让产品经理和运营也能直观看到 Agent 的性能和瓶颈。探索本地模型将核心 LLM 替换为 Ollama 运行的本地模型如 Llama 3.2, Qwen2.5在保证效果的同时控制成本和数据隐私。这个项目框架为你打开了一扇门门后是构建可靠、可扩展、可评估的智能应用的全新工程范式。建议你克隆代码从连接一个真实的数据库 MCP Server 开始亲手体验从“黑盒”到“白盒”的掌控感。在面试中这份经历和思考将成为你最有力的证明。
返回列表