基于LangGraph与MCP构建可追踪、可评估的企业级AI Agent框架

这次我们来看一个能直接用在企业级项目里的 AI Agent 开发方案。如果你正在准备面试或者想把手头的 Agent 项目从玩具级升级到生产级这篇文章可以直接收藏。核心不是讲概念而是给出一套能跑通、能追踪、能评估的实战代码框架。这个方案的核心是LangGraph和MCPModel Context Protocol。LangGraph 负责构建 Agent 的“大脑”和决策流程让 Agent 能处理复杂的多步骤任务而 MCP 则像一套标准化的“插件”协议让 Agent 可以安全、统一地调用外部工具和数据源。最关键的是这套组合拳解决了企业最关心的三个问题任务执行过程的可追踪性、效果的可评估性、以及系统状态的可观测性。这意味着你的 Agent 不再是黑盒每一步决策、每一次工具调用、每一个中间结果都能被记录、分析和复盘。本文会带你从零搭建一个具备这些能力的 Agent 项目。我们会重点关注如何用 LangGraph 设计一个支持分支、循环和状态管理的 Agent 工作流如何通过 MCP 集成外部能力比如数据库、搜索引擎、API以及如何在整个流程中嵌入追踪Tracing和评估Evaluation的钩子Hooks让你能清晰地看到 Agent 的“思考”过程并评判其表现。这套方案不依赖特定的大模型你可以用 OpenAI、DeepSeek 或本地部署的 Ollama 来驱动。1. 核心能力速览能力项说明项目类型企业级 AI Agent 开发框架与最佳实践核心技术栈LangGraph (工作流编排) MCP (工具/上下文协议) 可观测性组件核心解决痛点Agent 任务流程黑盒、执行过程不可追溯、效果难以量化评估硬件/环境门槛无特殊 GPU 要求。主要依赖 Python 环境需要能访问所选的大模型 API如 OpenAI或本地模型服务如 Ollama。启动与部署基于 Python 脚本启动可封装为 REST API 服务方便集成到现有系统。是否支持 API是。核心 Agent 可包装为 Web 服务提供任务提交、状态查询、结果获取等接口。是否支持批量/异步任务是。通过 LangGraph 的状态管理和异步执行可处理任务队列适合批量处理场景。关键产出可复用的 Agent 工作流代码、完整的追踪日志、结构化的评估报告、可观测性仪表板数据。适合场景面试项目深度构建、PoC验证、企业内部自动化助手、需审计和复盘的生产级Agent应用。2. 适用场景与使用边界这个项目方案最适合以下几类开发者面试备战者需要超越简单的LangChain OpenAI调用展示对 Agent 生命周期管理、可观测性等高级话题的理解。PoC概念验证负责人需要快速搭建一个功能完整、且具备监控评估能力的 Agent 原型向团队或客户证明其可行性。中小型项目技术选型者希望采用一套成熟、模块化且易于扩展的架构来开发生产环境的 Agent 应用。它能解决什么问题流程可视化将 Agent 的“思考-行动-观察”循环以图谱形式呈现清晰展示任务分解与执行路径。问题定位当 Agent 输出错误或陷入循环时能通过追踪日志快速定位是提示词问题、工具调用失败还是状态逻辑错误。效果量化通过嵌入评估节点或后处理分析对 Agent 完成任务的质量、成本、耗时进行量化评分支持持续优化。安全与合规通过 MCP 规范工具调用可以集中管理权限、记录审计日志满足企业安全要求。需要谨慎对待的边界并非开箱即用的产品这是一个开发框架和模式你需要根据具体业务逻辑填充提示词、工具和评估标准。性能依赖底层模型Agent 的智能程度、响应速度和大规模并发能力很大程度上取决于所选大模型的能力和你的工程优化。评估体系需自定义项目提供了评估的“插座”但“评分标准”需要你根据业务目标来定义例如信息检索的准确性、代码生成的可运行性。数据安全与隐私如果 Agent 通过 MCP 访问内部数据库或敏感 API必须确保认证、授权和传输过程的安全并遵守相关数据隐私法规。3. 环境准备与前置条件在开始编码前请确保你的开发环境满足以下要求。这套方案对硬件没有特殊要求重点在于软件栈的配置。Python 环境推荐使用 Python 3.10 或 3.11。更高版本可能存在依赖兼容性问题。建议使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境 (以 conda 为例) conda create -n ai-agent python3.10 conda activate ai-agent大模型访问权限准备一个可用的 LLM API。以下任选其一OpenAI需要OPENAI_API_KEY。DeepSeek需要DEEPSEEK_API_KEY。本地模型 (如 Ollama)需要在本地运行 Ollama 服务并拉取相应模型如llama3.2、qwen2.5。其他兼容 OpenAI 格式的 API如 Together AI, Groq 等。基础依赖安装我们将使用langgraph,langchain, 以及mcp相关的库。由于 MCP 生态较新部分库可能需要从源码或特定渠道安装。# 安装核心依赖 pip install langgraph langchain langchain-openai langchain-community # 安装 MCP 相关库 (示例具体包名可能随时间变化) # pip install mcp langchain-mcp # 如果官方包已发布 # 或者从 Github 克隆相关 SDK可选可观测性工具为了更好的可视化可以集成LangSmith(LangChain 官方平台) 或开源的Phoenix。本文会以代码日志和简单存储为例但会给出集成这些平台的指引。LangSmith: 注册账号获取LANGSMITH_API_KEY。Phoenix:pip install arize-phoenix代码编辑器/IDE推荐 VS Code 或 PyCharm。4. 项目结构与核心模块设计在动手写代码前先规划好项目结构。一个清晰的结构是项目可维护、可扩展的基础。your_agent_project/ ├── agents/ │ ├── __init__.py │ ├── base_agent.py # 基础 Agent 类封装通用逻辑 │ └── research_agent.py # 示例研究型 Agent 实现 ├── graphs/ │ ├── __init__.py │ └── research_graph.py # 使用 LangGraph 定义的研究工作流 ├── tools/ │ ├── __init__.py │ ├── mcp_tools.py # 通过 MCP 集成的工具 (如数据库查询) │ └── custom_tools.py # 自定义 Python 工具函数 ├── evaluation/ │ ├── __init__.py │ ├── evaluators.py # 评估器定义 │ └── metrics.py # 评估指标 (如正确性、成本) ├── tracing/ │ ├── __init__.py │ └── custom_handler.py # 自定义追踪处理器记录日志或发送到监控系统 ├── config/ │ └── settings.py # 配置文件管理 API Key、模型参数等 ├── data/ │ ├── inputs/ # 存放输入样例 │ └── outputs/ # 存放运行结果和追踪日志 ├── scripts/ │ └── run_agent.py # 主运行脚本 ├── requirements.txt └── README.md核心模块解读agents/: 定义不同类型的 Agent。base_agent.py提供公共父类处理模型初始化、基础对话逻辑。graphs/:核心目录。使用 LangGraph 定义具体的工作流。一个工作流是一个有向图节点是函数或子Agent边是条件逻辑。tools/: 存放 Agent 可以调用的工具。mcp_tools.py是关键它通过 MCP 客户端与外部服务如数据库 Server、搜索引擎 Server通信。evaluation/: 评估模块。可以在工作流的特定节点如结束节点插入评估函数对中间或最终结果进行打分。tracing/: 追踪模块。通过 LangGraph 的Checkpointer或自定义回调函数记录每个节点的输入、输出、耗时和错误信息。5. 实战构建一个可追踪的研究型 Agent我们以一个“研究型 Agent”为例它需要完成以下任务根据用户问题进行网络搜索总结信息并评估自身回答的质量。这个流程天然适合用 LangGraph 来编排。5.1 定义 Agent 状态首先在graphs/research_graph.py中我们定义 Agent 运行过程中需要维护的状态。这通常是一个 TypedDict。from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): Agent 工作流的状态定义 # 用户输入的问题 question: str # 模型生成的思考步骤或计划 plan: str # 从工具如搜索获取的原始信息 gathered_info: List[str] # 模型生成的最终答案草稿 draft_answer: str # 评估模块对答案的评分和反馈 evaluation: dict # 最终润色后的答案 final_answer: str # 记录迭代次数防止无限循环 iteration: Annotated[int, operator.add]5.2 创建工具集成 MCP假设我们通过 MCP 集成了一个网络搜索工具。在tools/mcp_tools.py中import httpx from langchain.tools import tool from config.settings import MCP_SERVER_URL class MCPClient: 一个简化的 MCP 客户端示例 def __init__(self, server_url: str MCP_SERVER_URL): self.server_url server_url self.client httpx.AsyncClient(base_urlserver_url) async def call_tool(self, tool_name: str, arguments: dict) - dict: 调用 MCP Server 上的工具 # 实际协议会更复杂这里做简化演示 response await self.client.post( f/tools/{tool_name}/execute, json{arguments: arguments} ) response.raise_for_status() return response.json() # 创建客户端实例单例模式或依赖注入 mcp_client MCPClient() tool async def web_search(query: str) - str: 使用 MCP 协议调用后端搜索工具。输入是一个搜索查询字符串。 try: result await mcp_client.call_tool(web_search, {query: query}) # 假设返回结构中有 content 字段 return result.get(content, No content found.) except Exception as e: return f搜索工具调用失败: {str(e)}关键点MCP 将工具的实现与 Agent 代码解耦。工具的实际逻辑如调用 Serper API 或 Google Search API运行在独立的MCP Server上Agent 只通过标准的协议与之通信。这提升了安全性和可维护性。5.3 使用 LangGraph 构建工作流现在在graphs/research_graph.py中构建图。我们将定义几个节点函数并用StateGraph把它们连接起来。from langgraph.graph import StateGraph, END from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver from tools.mcp_tools import web_search from langchain_openai import ChatOpenAI from evaluation.evaluators import answer_correctness_evaluator import asyncio # 初始化模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) def plan_node(state: AgentState) - dict: 节点1分析问题制定计划 messages [ (system, 你是一个研究助手。请分析用户问题并制定一个分步研究计划。), (human, state[question]) ] response llm.invoke(messages) return {plan: response.content} async def research_node(state: AgentState) - dict: 节点2执行研究调用搜索工具 # 基于计划或问题生成搜索词这里简化处理 search_query state[question] search_result await web_search.ainvoke({query: search_query}) # 将新信息追加到列表 current_info state.get(gathered_info, []) current_info.append(search_result) return {gathered_info: current_info} def answer_node(state: AgentState) - dict: 节点3基于收集的信息生成答案草稿 info_text \n.join(state[gathered_info]) messages [ (system, 你是一个严谨的助手。请根据以下信息生成一个结构清晰、准确的答案。), (human, f问题{state[question]}\n\n收集到的信息\n{info_text}) ] response llm.invoke(messages) return {draft_answer: response.content} def evaluate_node(state: AgentState) - dict: 节点4评估生成的答案草稿 evaluation_result answer_correctness_evaluator( questionstate[question], context\n.join(state[gathered_info]), answerstate[draft_answer] ) # evaluation_result 可能是一个字典如 {score: 0.8, feedback: ...} return {evaluation: evaluation_result} def refine_node(state: AgentState) - dict: 节点5根据评估反馈润色答案 feedback state[evaluation].get(feedback, ) messages [ (system, 请根据以下评估反馈优化你之前的答案。), (human, f原始答案{state[draft_answer]}\n\n评估反馈{feedback}\n\n请输出优化后的最终答案。) ] response llm.invoke(messages) return {final_answer: response.content, iteration: state[iteration] 1} def should_continue(state: AgentState) - str: 条件边决定是继续迭代还是结束 score state[evaluation].get(score, 0) iteration state.get(iteration, 0) # 如果评分够高或迭代次数太多则结束 if score 0.7 or iteration 3: return end else: return refine # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(research, research_node) workflow.add_node(answer, answer_node) workflow.add_node(evaluate, evaluate_node) workflow.add_node(refine, refine_node) # 设置边定义执行顺序 workflow.set_entry_point(plan) workflow.add_edge(plan, research) workflow.add_edge(research, answer) workflow.add_edge(answer, evaluate) # 条件边根据评估结果决定是润色还是结束 workflow.add_conditional_edges( evaluate, should_continue, { end: END, # 结束 refine: refine, # 去润色 } ) workflow.add_edge(refine, evaluate) # 润色后再次评估形成循环 # 启用检查点用于持久化和追踪 memory AsyncSqliteSaver.from_conn_string(:memory:) # 实际项目可用文件路径 app workflow.compile(checkpointermemory)这个图定义了完整的工作流计划 - 研究 - 生成答案 - 评估 - (如果评分低) - 润色 - 再次评估... 直到评分达标或超限。5.4 嵌入追踪与评估追踪在上面的代码中AsyncSqliteSaver就是一个简单的检查点存储器它会自动记录每个节点执行前后的状态。在生产环境中你可以将其替换为更强大的后端如数据库或者集成LangSmith的 callback。# 集成 LangSmith 追踪 (示例) from langsmith import Client from langchain.callbacks.tracers.langchain import LangChainTracer client Client() tracer LangChainTracer(project_namemy-research-agent) # 在调用 app.invoke() 时传入 callbacks[tracer]评估我们在图中专门设计了evaluate_node。评估器 (evaluation/evaluators.py) 可以用另一个 LLM 来评判也可以基于规则。# evaluation/evaluators.py from langchain_openai import ChatOpenAI from langchain.evaluation import load_evaluator llm ChatOpenAI(modelgpt-4o-mini, temperature0) def answer_correctness_evaluator(question: str, context: str, answer: str) - dict: 使用 LLM 作为评估器判断答案的正确性 eval_prompt f 你是一个严格的评估员。 问题{question} 参考上下文{context} 待评估答案{answer} 请从以下方面评估 1. 事实准确性0-1分答案是否与上下文提供的事实一致 2. 完整性0-1分答案是否充分回答了问题的所有部分 3. 清晰度0-1分答案是否表达清晰、易于理解 请输出一个 JSON 对象包含 score (三项平均分) 和 feedback (具体的改进建议)。 response llm.invoke([(human, eval_prompt)]) # 这里需要解析 response.content 中的 JSON简化处理直接返回 import json try: return json.loads(response.content) except: return {score: 0.5, feedback: 评估器解析失败。}6. 运行 Agent 与查看结果创建一个主运行脚本scripts/run_agent.py。import asyncio import json from datetime import datetime from graphs.research_graph import app, AgentState async def main(): # 1. 初始化输入状态 initial_state: AgentState { question: LangGraph 和 LangChain 在构建 Agent 时的主要区别是什么, plan: , gathered_info: [], draft_answer: , evaluation: {}, final_answer: , iteration: 0 } # 2. 配置一个线程ID用于追踪本次会话 config {configurable: {thread_id: fthread_{datetime.now().isoformat()}}} print(开始执行研究型 Agent...) # 3. 执行工作流 final_state await app.ainvoke(initial_state, configconfig) # 4. 输出最终结果 print(\n *50) print(【最终答案】) print(final_state.get(final_answer, N/A)) print(\n【评估结果】) print(json.dumps(final_state.get(evaluation, {}), indent2, ensure_asciiFalse)) print(\n【迭代次数】) print(final_state.get(iteration, 0)) # 5. (可选) 从检查点中读取完整的执行轨迹 # checkpointer app.checkpointer # history await checkpointer.list(config) # 可以遍历 history 查看每个步骤的输入输出 if __name__ __main__: asyncio.run(main())运行这个脚本你将看到 Agent 逐步执行并输出最终答案和评估分数。所有中间状态和节点调用都被checkpointer记录了下来。7. 扩展为 API 服务与批量任务要让这个 Agent 能被其他系统调用我们需要将其包装成一个 Web 服务。这里使用 FastAPI 示例。# api/main.py from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from typing import Optional import uuid from .agent_runner import run_agent_async # 将之前的运行逻辑封装在此函数中 app FastAPI(titleResearch Agent API) # 内存中存储任务状态生产环境应用数据库 tasks {} class AgentRequest(BaseModel): question: str callback_url: Optional[str] None # 支持异步回调 class TaskStatus(BaseModel): task_id: str status: str # pending, running, completed, failed result: Optional[dict] None app.post(/research, response_modelTaskStatus) async def create_research_task(request: AgentRequest, background_tasks: BackgroundTasks): 提交一个研究任务 task_id str(uuid.uuid4()) tasks[task_id] {status: pending, result: None} # 将任务放入后台执行 background_tasks.add_task( execute_agent_task, task_idtask_id, questionrequest.question, callback_urlrequest.callback_url ) return TaskStatus(task_idtask_id, statuspending) app.get(/research/{task_id}, response_modelTaskStatus) async def get_task_status(task_id: str): 查询任务状态和结果 task tasks.get(task_id) if not task: return {task_id: task_id, status: not_found} return TaskStatus(task_idtask_id, **task) async def execute_agent_task(task_id: str, question: str, callback_url: Optional[str]): 后台执行 Agent 任务 try: tasks[task_id][status] running # 调用我们之前封装的 Agent 运行逻辑 result await run_agent_async(question) tasks[task_id].update({status: completed, result: result}) except Exception as e: tasks[task_id].update({status: failed, result: {error: str(e)}}) # 如果有回调 URL通知调用方 if callback_url: # 使用 httpx 异步发送 POST 请求 pass批量任务处理基于这个 API你可以轻松实现批量处理。写一个脚本读取问题列表并发或顺序地调用/research接口并收集所有task_id然后轮询/research/{task_id}获取结果。使用asyncio或Celery可以管理更复杂的任务队列。8. 资源占用、性能观察与优化建议由于本项目核心是逻辑编排和 API 调用资源占用主要取决于大模型 API 调用成本与延迟。使用 GPT-4 等高级模型费用和耗时更高。工具调用如网络搜索、数据库查询的延迟。本地内存/CPU用于运行 Python 代码、维护状态图和存储检查点数据。性能观察点节点执行时间在关键节点函数开始和结束处记录时间戳计算耗时。Token 消耗通过 LangChain 回调或模型供应商的日志统计每次 LLM 调用的输入/输出 token 数用于成本分析。工具调用成功率记录 MCP 工具调用的成功/失败次数和错误类型。循环迭代次数监控iteration状态防止因评估标准过严导致无限循环。优化建议缓存对频繁查询且结果不变的工具调用如某些数据查询引入缓存机制。超时与重试为网络请求LLM API、MCP 工具设置合理的超时和重试策略。流式输出如果最终答案较长考虑使用 LLM 的流式响应提升用户体验。检查点存储优化生产环境将AsyncSqliteSaver替换为更高效的后端如 Redis并定期清理旧数据。9. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 LangGraph 或 MCP 库失败依赖未正确安装或版本冲突。检查pip list确认langgraph,langchain等版本。查看错误信息。创建干净的虚拟环境根据官方文档或requirements.txt重新安装。关注 MCP 相关库的安装方式。运行时报错NotImplementedError(MCP相关)MCP 工具客户端未正确实现或服务未启动。检查tools/mcp_tools.py中的MCPClient.call_tool方法是否与你的 MCP Server 协议匹配。确认 MCP Server 是否在运行且可达。参考 MCP 官方 SDK 示例实现客户端。确保 MCP Server URL (MCP_SERVER_URL) 配置正确。Agent 陷入无限循环should_continue条件函数逻辑有误或评估分数始终不达标。打印每次循环后的state[evaluation]和state[iteration]。检查评估器打分是否过于苛刻。调整should_continue中的阈值如将score 0.7改为score 0.6。为迭代次数设置硬性上限。LangSmith 追踪看不到数据LANGSMITH_API_KEY未设置或project_name不正确或回调未传入。检查环境变量。确认在调用app.invoke()时传入了callbacks[tracer]。访问 LangSmith 网站查看项目列表。正确设置环境变量。确保追踪器在每次调用时都被正确初始化并传入。API 服务启动后调用 Agent 超时Agent 单次执行时间过长超过了 HTTP 默认超时时间。查看 Agent 执行日志定位耗时最长的节点是 LLM 调用还是工具调用。1. 优化慢节点。2. 将 API 设计为异步任务模式如本文所示立即返回task_id通过另外的端点查询结果。3. 调整 Web 服务器如 Uvicorn的超时配置。评估器LLM给出的分数不稳定评估提示词Prompt不够明确导致 LLM 评判标准波动。用同一组(question, context, answer)多次运行评估器观察分数差异。优化评估提示词要求 LLM 输出更结构化的结果如必须包含具体扣分项。考虑使用更稳定的模型如 GPT-4做评估或引入基于规则的辅助评估。检查点存储文件过大长时间运行后SQLite 文件记录了过多历史状态。检查data/目录下数据库文件大小。实现定期清理策略例如只保留最近 N 天的任务记录或在编译图时使用内存型检查点 (:memory:) 但牺牲了持久化。10. 最佳实践与项目进阶方向最佳实践配置化管理将所有 API Key、模型参数、服务器地址、超时设置等放在config/settings.py中通过环境变量加载。日志标准化使用logging模块为不同模块设置不同日志级别并输出到文件方便调试和审计。测试驱动为你的graphs、tools、evaluators编写单元测试和集成测试确保逻辑正确。版本控制对提示词Prompt、工作流图Graph定义、评估标准进行版本管理跟踪每次变更对效果的影响。安全第一通过 MCP Server 集中管理敏感操作如数据库写操作、支付接口在 Server 端实施严格的权限控制和输入验证。进阶方向更复杂的图结构尝试子图Subgraph、并行执行、人工审核节点等高级特性。集成向量数据库让 Agent 拥有长期记忆能够参考历史对话和知识库。多 Agent 协作创建多个具备不同技能的 Agent如研究员、写手、校对员让它们通过 LangGraph 协同完成更复杂的任务。前端可视化利用 LangGraph 的序列化能力将执行过程生成流程图在前端实时展示 Agent 的“思考”过程。自动化评估与再训练收集大量运行结果和人工反馈构建评估数据集用于微调驱动 Agent 的 LLM 或优化提示词。这套基于 LangGraph MCP 的 AI Agent 项目方案将可追踪、可评估、可观测的理念落到了代码层面。它不仅能帮你打造一个出色的面试作品更能为构建真正可靠、可维护的企业级智能应用打下坚实基础。建议从本文的示例代码出发替换成你自己的业务逻辑和工具逐步完善功能。