构建企业级AI Agent：LangGraph与MCP协议下的可观测性实践

1. 先搞清楚面试官到底想听什么：从“玩具”到“企业级”的鸿沟

面试时被问到“写一个 AI Agent 项目”，如果你只讲通了 LangChain 的链条、调了 OpenAI 的 API，大概率会被认为项目深度不够。现在的面试官，尤其是中高级岗位，想听的远不止功能实现。他们真正关心的是：你的 Agent 在复杂、真实、长期运行的环境下，是否可控、可观测、可评估、可维护。这恰恰是“玩具 Demo”和“企业级项目”的核心分水岭。

这个主题之所以“被问爆”，是因为它直击了 AI 应用工程化的痛点。一个能跑起来的 Agent 只是起点，如何确保它在处理成千上万次用户交互时不“失忆”、不“跑偏”、不“崩溃”，并且出了问题能快速定位，才是体现你工程化思维和项目深度的关键。

所以，一个高分的回答不应该从“我用了 LangGraph 画了个图”开始，而应该从“我如何解决 Agent 在长周期、多任务场景下的状态追踪、效果评估和系统可观测性问题”切入。LangGraph 是你的编排引擎，而 MCP（Model Context Protocol）和配套的追踪评估体系，才是你项目里的“压舱石”和“黑匣子”。本文将围绕这个核心，拆解一个可以直接用于面试或企业级参考的项目架构与实现要点。

2. 项目核心架构：LangGraph 负责流程，MCP 与可观测层负责“稳住”

在开始写代码之前，必须把架构想清楚。一个具备追踪、评估、可观测能力的 AI Agent 系统，通常分为三层：

1. 编排执行层（LangGraph）：负责定义 Agent 的工作流（StateGraph），管理状态（State）的流转，决定下一步调用哪个工具（Tool）或哪个 LLM。
2. 上下文与工具层（MCP）：这是项目的关键升级点。不再把数据库、API、内部系统等工具硬编码或简单封装，而是通过MCP 服务器来统一暴露。Agent 通过标准的 MCP 协议与这些资源对话，这使得工具可以独立开发、热插拔，并且所有的工具调用（输入、输出、耗时）天生就是可追踪的。
3. 可观测与评估层：这是体现企业级思维的灵魂。它贯穿整个系统，负责收集、记录、分析和评估。

下面这张图概括了核心的数据流与关注点：

flowchart TD subgraph A [可观测与评估层] direction TB A1[“链路追踪<br>(Trace Collection)”] --> A2[“评估体系<br>(Evaluation)”] A2 --> A3[“监控与告警<br>(Monitoring & Alerting)”] end subgraph B [编排执行层] B1[LangGraph StateGraph] --> B2[“执行引擎<br>(运行工作流)”] end subgraph C [上下文与工具层] C1[“MCP Server<br>(数据库/API/文件等)”] --> C2[“MCP Client<br>(在Agent中调用)”] end B2 -- “执行过程产生<br>状态、决策、调用” --> A1 C2 -- “所有工具调用<br>皆被记录” --> A1 A3 -- “发现问题<br>触发干预” --> B1

为什么是 LangGraph + MCP？

• LangGraph解决了“有状态、多步骤”工作流的编排问题。面试时你可以对比 LangChain 的简单链（Chain），强调 LangGraph 的State对象如何优雅地承载对话历史、中间结果和自定义变量，以及Graph如何清晰定义循环、分支、并行等复杂逻辑。
• MCP解决了“工具集成标准化”和“上下文管理精细化”的问题。传统方式下，工具逻辑和 Agent 代码耦合深，难以管理和追踪。MCP 将每个数据源（如数据库、CRM、知识库）抽象为一个独立的 Server，通过标准协议提供“资源列表”和“操作指令”。Agent 只需知道协议，无需关心底层实现，这使得系统更模块化，并且所有通过 MCP 的交互都自动具备了上下文和审计线索。

企业级关注点直接映射：

• 追踪：通过 LangGraph 的天然执行路径和 MCP 的标准化调用日志，实现全链路追踪。你知道一个用户问题，Agent 走了哪几个节点，调了哪几个工具，每个步骤的输入输出是什么。
• 评估：基于追踪数据，构建评估体系。不仅是最终答案的对错（难以量化），更是过程指标：工具调用是否必要？返回结果是否相关？执行路径是否高效？
• 可观测：将追踪日志和评估指标，通过 OpenTelemetry 等标准输出到监控系统（如 Prometheus + Grafana），实现实时监控、历史回溯和智能告警。

3. 环境准备与核心组件落地

理论说完，我们落到实地。假设我们要构建一个“智能客服工单处理 Agent”，它需要理解用户问题、查询知识库、检索相似工单、最终生成解决方案或转人工。

3.1 基础环境与依赖

首先，明确你的技术栈。这里是一个 Python 环境的示例：

# 核心框架 pip install langgraph langchain langchain-openai # MCP 相关 - 这是关键 pip install mcp[cli] mcp-client # 可观测性 - 用于数据收集和导出 pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp # 向量数据库（用于知识库和工单检索） pip install chromadb # 其他工具 pip install pydantic python-dotenv

你需要准备：

1. LLM API Key：如 OpenAI、DeepSeek 等。建议在环境变量中配置。
2. 一个简单的数据库或模拟数据源：用于模拟工单系统。
3. 一个知识库文件：如 Markdown 或文本文件，作为内部知识。

3.2 构建 MCP Server（以数据库查询为例）

这是体现你项目深度的第一步。我们不直接在 Agent 里写 SQL，而是创建一个 MCP Server。

创建一个文件database_mcp_server.py：

import json from typing import Any, List from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import sqlite3 import pydantic # 1. 定义工具参数模型 class QueryArgs(pydantic.BaseModel): query_sql: str # 2. 创建 Server 实例 app = Server("ticket-database-server") # 3. 声明 Server 提供的“资源”（这里可以理解为可查询的表或视图） @app.list_resources() async def handle_list_resources() -> List[Any]: return [ { "uri": "db://tickets/table", "name": "工单表", "description": "包含所有历史工单记录", "mimeType": "application/x-sqlite3", # 非标准，示例用 } ] # 4. 声明 Server 提供的“工具”（即可以执行的操作） @app.list_tools() async def handle_list_tools() -> List[Any]: return [ { "name": "query_tickets", "description": "执行SQL查询工单数据。请提供合法的SQL SELECT语句。", "inputSchema": { "type": "object", "properties": { "query_sql": {"type": "string", "description": "SQL查询语句"} }, "required": ["query_sql"] } } ] # 5. 实现工具的执行逻辑 @app.call_tool() async def handle_call_tool(name: str, arguments: Any) -> List[Any]: if name == "query_tickets": args = QueryArgs(**arguments) # 连接模拟数据库 conn = sqlite3.connect('tickets.db') cursor = conn.cursor() try: cursor.execute(args.query_sql) results = cursor.fetchall() columns = [description[0] for description in cursor.description] formatted_results = [dict(zip(columns, row)) for row in results] return [{ "type": "text", "text": json.dumps(formatted_results, ensure_ascii=False, indent=2) }] except sqlite3.Error as e: return [{"type": "text", "text": f"数据库查询错误: {e}"}] finally: conn.close() else: raise ValueError(f"未知工具: {name}") # 6. 运行 Server (通常通过 mcp cli 或 stdio 调用) if __name__ == "__main__": # 开发时可以直接运行测试，生产环境通过 stdio 与 MCP Client 通信 import asyncio from mcp.server.stdio import stdio_server async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, InitializationOptions()) asyncio.run(main())

关键点：

• 这个 Server 独立运行，通过标准输入输出（stdio）与 LangGraph Agent 通信。
• Agent 端只需要知道这个 Server 提供了query_tickets工具，而无需知道背后是 SQLite、MySQL 还是 REST API。
• 所有对query_tickets的调用，其请求参数（SQL）和返回结果都会被 MCP 框架自动记录，这是实现追踪的基础。

3.3 定义 LangGraph 工作流与状态

创建agent_workflow.py，定义 Agent 的核心大脑。

from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import Tool from mcp import ClientSession from mcp.client.stdio import stdio_client import asyncio # 1. 定义状态 State，这是 LangGraph 的核心 class AgentState(TypedDict): user_input: str conversation_history: Annotated[List[str], "append"] # 自动追加历史 knowledge_context: str ticket_context: str current_step: str final_answer: str # 可以在这里添加追踪ID，用于关联日志 trace_id: str # 2. 初始化 LLM 和工具 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 3. 定义“节点”函数 - 每个节点是工作流的一步 async def understand_user_input(state: AgentState): """节点1：理解用户意图，并规划步骤""" prompt = f""" 用户的问题是：{state['user_input']} 请分析用户意图，并决定后续步骤。可能的步骤有： 1. 查询知识库 (query_knowledge_base) 2. 检索相似工单 (query_tickets) 3. 直接生成答案 (generate_answer) 4. 请求人工 (transfer_to_human) 输出格式为：下一步：<步骤名称> """ response = await llm.ainvoke(prompt) next_step = response.content.strip() return {"current_step": next_step} async def query_knowledge_tool(state: AgentState): """节点2：调用知识库工具（这里简化为模拟，实际应连接另一个MCP Server或向量库）""" # 模拟知识库查询 knowledge = "根据知识库：重启路由器可以解决80%的网络连接问题。" return {"knowledge_context": knowledge} async def query_tickets_tool(state: AgentState): """节点3：调用工单数据库工具（通过MCP）""" user_input = state['user_input'] # 这里是关键：通过 MCP Client 调用我们之前写的 Server async with stdio_client(["python", "database_mcp_server.py"]) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 构造一个简单的查询，实际中可以让LLM动态生成SQL sql = f"SELECT title, solution FROM tickets WHERE title LIKE '%{user_input[:10]}%' LIMIT 3" result = await session.call_tool("query_tickets", {"query_sql": sql}) ticket_info = result[0].text if result else "未找到相似工单" return {"ticket_context": ticket_info} async def generate_final_answer(state: AgentState): """节点4：综合所有信息，生成最终答案""" prompt = f""" 用户问题：{state['user_input']} 知识库信息：{state.get('knowledge_context', '无')} 历史工单信息：{state.get('ticket_context', '无')} 请生成给用户的最终答复。 """ response = await llm.ainvoke(prompt) return {"final_answer": response.content} # 4. 构建图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("understand", understand_user_input) workflow.add_node("query_knowledge", query_knowledge_tool) workflow.add_node("query_tickets", query_tickets_tool) workflow.add_node("generate", generate_final_answer) # 设置入口 workflow.set_entry_point("understand") # 定义边（路由逻辑） def decide_next_step(state: AgentState): """根据当前步骤决定下一个节点""" step = state['current_step'] if "查询知识库" in step: return "query_knowledge" elif "检索相似工单" in step: return "query_tickets" elif "生成答案" in step: return "generate" else: return "generate" # 默认 workflow.add_conditional_edges( "understand", decide_next_step ) workflow.add_edge("query_knowledge", "generate") workflow.add_edge("query_tickets", "generate") workflow.add_edge("generate", END) # 编译图 app = workflow.compile()

关键点：

• AgentState定义了整个工作流的“记忆体”，所有节点都读写它。
• 每个node是一个清晰的函数，职责单一。
• 路由逻辑decide_next_step让工作流具备了动态决策能力。
• 在query_tickets_tool节点中，我们通过 MCP Client 标准化地调用了外部工具，这是追踪的关键接入点。

4. 注入追踪、评估与可观测性

这是将项目从“能跑”提升到“能用”乃至“可靠”的关键步骤。

4.1 实现链路追踪

我们需要在关键位置埋点，记录执行轨迹。一个简单而有效的方式是利用 LangGraph 的回调（Callbacks）和 MCP 的调用日志。

方案一：使用 LangGraph 内置回调

from langchain.callbacks.base import BaseCallbackHandler from datetime import datetime import uuid class TracingCallbackHandler(BaseCallbackHandler): def __init__(self): self.trace_id = str(uuid.uuid4()) self.logs = [] def on_chain_start(self, serialized, inputs, **kwargs): self.logs.append({ "timestamp": datetime.now().isoformat(), "event": "chain_start", "chain_name": serialized.get("name", "unknown"), "inputs": str(inputs)[:200] # 截断避免过长 }) def on_tool_start(self, serialized, input_str, **kwargs): self.logs.append({ "timestamp": datetime.now().isoformat(), "event": "tool_start", "tool_name": serialized.get("name", "unknown"), "input": input_str }) def on_tool_end(self, output, **kwargs): self.logs.append({ "timestamp": datetime.now().isoformat(), "event": "tool_end", "output": str(output)[:200] }) # 在执行工作流时传入回调 tracer = TracingCallbackHandler() async def run_agent_with_trace(user_query: str): initial_state = AgentState( user_input=user_query, conversation_history=[], knowledge_context="", ticket_context="", current_step="", final_answer="", trace_id=tracer.trace_id ) config = {"callbacks": [tracer]} final_state = await app.ainvoke(initial_state, config=config) return final_state, tracer.logs

方案二：集成 OpenTelemetry（更企业级）

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter # 设置追踪 trace.set_tracer_provider(TracerProvider()) tracer = trace.get_tracer(__name__) # 输出到控制台（生产环境应输出到 Jaeger, Tempo 等） span_processor = BatchSpanProcessor(ConsoleSpanExporter()) trace.get_tracer_provider().add_span_processor(span_processor) # 在工具函数或节点中用 @trace 装饰器或 with 块包裹关键操作 def query_tickets_tool(state: AgentState): with tracer.start_as_current_span("query_tickets_tool") as span: span.set_attribute("user_input", state['user_input'][:50]) # ... 原有的 MCP 调用逻辑 span.set_attribute("sql_executed", sql) span.set_attribute("result_rows", len(results)) return {"ticket_context": ticket_info}

4.2 设计评估体系

评估不能只靠人眼看最终答案。需要设计自动化或半自动化的评估指标。

实现一个简单的答案相关性评估器：

from langchain.evaluation import load_evaluator from langchain.evaluation import EvaluatorType async def evaluate_answer_relevance(question: str, answer: str, context: str): """使用 LangChain 的评估器进行相关性打分""" evaluator = load_evaluator(EvaluatorType.QA) eval_result = await evaluator.aevaluate_strings( prediction=answer, input=question, reference=context # 这里传入检索到的知识作为参考 ) # eval_result 可能是一个字典，包含 'score' 或 'reasoning' return eval_result.get('score', 0), eval_result.get('reasoning', '')

4.3 构建可观测仪表板

将追踪日志和评估指标可视化。最直接的方式是将 OpenTelemetry 数据导出到 Prometheus，再用 Grafana 展示。

1. 配置 OpenTelemetry 导出到 Prometheus：

   from opentelemetry.exporter.prometheus import PrometheusMetricReader from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.metrics import set_meter_provider metric_reader = PrometheusMetricReader() provider = MeterProvider(metric_readers=[metric_reader]) set_meter_provider(provider)

2. 在关键位置记录指标：

   from opentelemetry.metrics import Counter, Histogram meter = meter_provider.get_meter("agent_meter") requests_counter = meter.create_counter("agent.requests.total") tool_duration_histogram = meter.create_histogram("agent.tool.duration.ms") # 在请求开始时 requests_counter.add(1, {"endpoint": "/chat"}) # 在工具调用前后记录耗时 start_time = time.time() # ... 调用工具 duration = (time.time() - start_time) * 1000 tool_duration_histogram.record(duration, {"tool_name": "query_tickets"})

3. 在 Grafana 中创建面板：监控 QPS、平均响应时间、工具调用耗时分布、错误率、答案相关性评分趋势等。

5. 面试复盘与项目深化的关键点

当你把上述内容串联起来，就构成了一个完整的、有深度的回答。在面试中，你需要清晰地传达以下层次：

1. 痛点与架构设计：先讲明白为什么单纯的链式调用不够，企业级需要追踪、评估、可观测。然后引出LangGraph（状态编排）+ MCP（标准化工具与上下文）+ 可观测层（追踪/评估/监控）的三层架构。
2. 核心实现：
   • LangGraph：重点说明State对象如何管理复杂状态，Graph如何定义非线性工作流，以及conditional_edge如何实现动态路由。
   • MCP：强调你如何将外部资源（数据库、API）抽象为独立的、协议化的 Server，从而实现了工具的热插拔和天生的可追踪性。这是区别于简单封装 Tool 类的关键。
   • 追踪：介绍你如何利用回调或 OpenTelemetry 在节点开始、结束、工具调用等关键点埋点，生成包含trace_id的完整链路日志。
   • 评估：说明你不仅评估最终答案，更评估过程（工具调用合理性、成功率）和结果质量（相关性、有用性），并给出了具体的实现方案（如基于规则的打分或调用评估 LLM）。
   • 可观测：提到你将指标和日志对接到了 Prometheus/Grafana 或类似系统，实现了实时监控。
3. 避坑与优化：
   • 状态管理：LangGraph 的 State 要设计得简洁，避免臃肿。对于超长对话，要考虑将历史记录外存到向量数据库。
   • MCP 性能：MCP 调用有序列化/反序列化开销，对于高频、低延迟的工具，要考虑性能权衡，或采用连接池、批处理。
   • 评估成本：用 LLM 评估 LLM 输出成本不低，可以抽样评估，或先使用规则、嵌入模型相似度等轻量级方法过滤。
   • 错误处理：在 LangGraph 图中要设计错误处理节点（try...except），当工具调用失败时，能优雅地重试或降级处理。
   • 安全性：通过 MCP 暴露工具时，要做好权限控制和输入验证（如防止 SQL 注入）。

最后，给面试官的印象应该是：你不仅仅是在调用 API 实现功能，而是在以一个系统架构师的视角，思考如何构建一个稳定、可靠、可运维、可迭代的 AI Agent 系统。你提到的每一个技术选型（LangGraph, MCP）都是为了解决具体的工程问题（状态管理、工具标准化），而追踪、评估、可观测性是你确保这个系统能在生产环境跑下去的必备手段。

把这个项目写在简历上，你可以称之为“基于 LangGraph 与 MCP 协议的可观测智能体系统”，并在描述中突出“实现了全链路追踪、多维度过程评估与系统监控”。这远比“我用 LangChain 写了一个聊天机器人”要有分量得多。