1. 先搞清楚“三大形态”到底在解决什么问题如果你正在做AI Agent开发并且感觉自己的项目还停留在“调用API、处理返回、结束任务”的简单循环里那这篇文章就是为你准备的。很多人一听到LangGraph、MCP、Workflow这些词第一反应是“又一个新框架”然后就开始纠结选型。但更关键的问题是它们各自解决了Agent开发中哪些不同的痛点为什么说“三大形态”而不是“三个框架”简单来说这三种形态对应着Agent能力构建和任务编排的三个不同层次和场景LangGraph解决的是“有状态的、复杂的、多步骤的任务流编排”问题。当你的Agent需要记住之前对话的内容、根据中间结果决定下一步、或者在多个工具之间循环调用时LangGraph的图Graph模型比简单的链Chain更合适。MCPModel Context Protocol解决的是“如何安全、统一地为Agent扩展外部能力和数据”的问题。它让你不必把所有工具代码都硬编码到Agent里而是通过一个标准协议动态地接入数据库、API、文件系统等资源让Agent的“技能”Skills可以即插即用。Workflow则是一个更上层的概念尤其在强调“可视化、可观测、可维护的业务流程”场景。它关注的是将包含Agent节点在内的整个业务过程如审批、数据处理流水线进行建模、监控和管理。所以别再只盯着API调用了。真正的Agent开发核心是设计Agent如何思考、如何行动、如何与复杂世界交互。下面我们就从最基础的LangGraph开始拆解如何一步步构建一个具备“渐进式技能披露”和“Harness”管控能力的智能体。2. 从LangGraph开始构建有状态的任务大脑LangGraph的核心是“图”。你可以把它理解为给Agent设计一个流程图节点Node是执行步骤比如调用LLM、运行工具边Edge决定了步骤之间的流转逻辑。这和LangChain的“链”最大区别在于图支持循环Cycles和状态State持久化。2.1 为什么需要“状态”想象一个订机票的Agent用户说“我想去上海。”Agent需要问“请问您的出发城市和出行日期是”用户回答“从北京出发下周五。”Agent需要记住“目的地上海”、“出发地北京”、“日期下周五”然后去查询航班。这个“记住”的过程就是状态管理。在LangGraph中你会定义一个State对象贯穿整个图的执行过程。2.2 一个最简单的LangGraph实战我们先不搞复杂的用一个“聊天助手”的图来感受一下。这个助手会判断用户意图如果是问候就回复问候如果是问天气就调用天气工具否则就通用回复。环境准备确保你安装了langgraph和langchain-openai或其他LLM库。建议使用Python虚拟环境。pip install langgraph langchain-openai第一步定义状态状态就是一个字典记录当前对话的上下文。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 存储完整的对话历史 messages: Annotated[list, add_messages] # 可以添加其他状态如用户意图、工具调用结果等 user_intent: str tool_result: str第二步创建节点节点就是普通的函数它接收当前State返回更新后的State。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) def intent_classifier(state: State): 节点1判断用户意图 last_message state[“messages”][-1].content # 这里简化处理实际可以用LLM或规则判断 if “天气” in last_message: intent “query_weather” elif “你好” in last_message or “嗨” in last_message: intent “greeting” else: intent “general_chat” return {“user_intent”: intent} def greeting_node(state: State): 节点2处理问候 return {“messages”: [“您好我是您的智能助手。”]} def general_chat_node(state: State): 节点3通用聊天 response llm.invoke(state[“messages”]) return {“messages”: [response.content]} # 注意weather_tool_node 需要调用外部API我们先假设它存在 def weather_tool_node(state: State): 节点4调用天气工具 # 模拟工具调用 city “北京” # 实际应从消息中提取 result f”{city}的天气是晴25摄氏度。” return {“tool_result”: result, “messages”: [result]}第三步定义边和路由逻辑这是LangGraph的灵魂。我们需要决定在一个节点执行完后下一个该执行谁。from langgraph.graph import END, StateGraph, START def router(state: State): 路由函数根据意图决定下一个节点 intent state.get(“user_intent”) if intent “greeting”: return “greeting_node” elif intent “query_weather”: return “weather_tool_node” else: return “general_chat_node” # 创建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“intent_classifier”, intent_classifier) workflow.add_node(“greeting_node”, greeting_node) workflow.add_node(“general_chat_node”, general_chat_node) workflow.add_node(“weather_tool_node”, weather_tool_node) # 设置入口 workflow.set_entry_point(“intent_classifier”) # 设置边路由 workflow.add_conditional_edges( “intent_classifier”, router, # 路由函数 { “greeting_node”: “greeting_node”, “query_weather”: “weather_tool_node”, “general_chat”: “general_chat_node” } ) # 设置终点 workflow.add_edge(“greeting_node”, END) workflow.add_edge(“weather_tool_node”, END) workflow.add_edge(“general_chat_node”, END) # 编译图 app workflow.compile()第四步运行测试# 初始化状态 initial_state {“messages”: [“北京今天天气怎么样”], “user_intent”: “”, “tool_result”: “”} # 运行图 final_state app.invoke(initial_state) print(final_state[“messages”][-1]) # 输出北京的天气是晴25摄氏度。通过这个例子你应该能感受到LangGraph把Agent的“决策流程”显式地定义了出来。你可以清晰地看到intent_classifier是一个路由枢纽它根据状态决定分支。这对于调试复杂Agent逻辑至关重要。2.3 关键点与避坑状态设计是核心一开始就要想清楚你的Agent需要记住什么。不要把所有东西都塞进messages像工具结果、中间变量、用户偏好等应该设计独立的字段避免状态污染。从简单图开始不要一上来就设计包含几十个节点的复杂工作流。先用3-5个节点把主流程跑通再逐步增加分支和循环。调试工具利用LangGraph Studio如果可用或简单的日志打印来追踪状态变化和流程走向。在关键节点打印state内容是排查逻辑错误最快的方法。循环的控制LangGraph支持循环但要避免死循环。通常需要设置一个最大循环次数或者在状态中设置一个should_continue的标记由某个节点决定是否跳出循环。3. 引入MCP为Agent动态加载“技能包”现在你的Agent有了一个聪明的大脑LangGraph图但它还是个“光杆司令”能力有限。传统做法是把所有工具函数写死在代码里但这带来两个问题1) Agent体积臃肿2) 每增加一个工具技能就要改代码、重新部署。MCP协议就是为了解决这个问题。它定义了一个标准方式让Agent客户端可以发现、描述和调用运行在独立进程服务器上的工具。你可以把MCP Server想象成一个技能插件商店Agent需要什么就临时加载什么。3.1 MCP的核心概念Server, Client, ToolsMCP Server一个独立的进程它对外提供一组Tools工具和Resources资源。例如一个“天气MCP Server”提供get_weather工具一个“数据库MCP Server”提供query_data工具和数据库连接资源。MCP Client你的Agent程序。它通过标准协议如STDIO或HTTP连接到MCP Server获取可用的工具列表然后在需要时调用它们。Tools一个可执行函数有明确的输入参数和输出。在MCP中工具的定义名称、描述、参数schema会被标准化地传递给Client。3.2 实战为天气Agent添加MCP技能假设我们已经有一个运行在本地50051端口的“天气MCP Server”。现在我们要让LangGraph Agent能调用它。第一步在LangGraph中集成MCP Client你需要一个MCP Client库来连接Server。这里以概念代码为例。# 假设有一个mcp_client库 import mcp_client from langchain.tools import Tool # 1. 连接到MCP Server weather_server mcp_client.connect(“http://localhost:50051”) # 2. 获取Server提供的所有工具列表 available_tools weather_server.list_tools() print(available_tools) # 例如[{‘name’: ‘get_weather’, ‘description’: ‘…’, ‘parameters’: {…}}] # 3. 将MCP工具包装成LangChain/LangGraph能识别的Tool对象 def call_mcp_weather_tool(city: str) - str: “”“调用远程MCP工具的包装函数”“” result weather_server.call_tool(“get_weather”, {“city”: city}) return result[“weather_info”] weather_tool Tool( name“get_weather”, funccall_mcp_weather_tool, description“根据城市名称查询天气情况” )第二步修改LangGraph节点使用动态工具现在我们可以用这个weather_tool来替换之前硬编码的weather_tool_node。from langgraph.prebuilt import ToolNode # 创建一个工具调用节点它能自动处理工具调用和状态更新 tool_node ToolNode(tools[weather_tool]) # 在你的图中原来指向weather_tool_node的边现在可以指向这个tool_node # 同时路由逻辑和状态结构可能需要微调以适配ToolNode的输出格式这样做的好处立刻显现解耦天气查询逻辑完全在独立的MCP Server中。你可以单独更新、扩展天气服务而无需重启或修改Agent主程序。安全Agent不需要知道天气API的密钥或数据库密码这些机密信息可以只保存在MCP Server端。动态性Agent可以在运行时发现新的MCP Server并加载其工具实现真正的“技能即插即用”。3.3 MCP实战要点Server稳定性MCP Server是独立进程它的崩溃会导致Agent调用失败。在生产环境中需要考虑Server的高可用、健康检查和重试机制。工具发现与管理当有多个MCP Server时Agent需要一个“工具注册中心”或简单的配置列表来管理连接。避免在代码中硬连接多个Server地址。性能与延迟网络调用必然带来延迟。对于高频调用的简单工具需要权衡是否值得用MCP。通常涉及外部资源数据库、API、复杂逻辑或需要隔离权限的工具是MCP的最佳应用场景。错误处理MCP调用可能因为网络、Server错误或参数错误而失败。在LangGraph的节点中必须做好异常捕获并设计错误处理路径例如返回一个友好的错误信息给用户或者重试。4. Workflow形态将Agent嵌入可观测的业务流程当你把LangGraph任务流和MCP技能组合起来一个功能强大的Agent就成型了。但站在业务角度这个Agent可能只是更大业务流程中的一个环节。这就是Workflow形态要解决的问题。Workflow关注的是任务编排、执行监控、错误处理、重试策略和结果收集。例如一个“用户反馈处理Workflow”可能包含1) Agent分析反馈情感和分类2) 根据分类路由到不同工单系统MCP工具3) 等待人工审核一个长时间等待的节点4) 发送处理结果通知。4.1 LangGraph本身就是一种Workflow引擎是的你之前构建的LangGraph图就是一个最简单的工作流。但生产级的Workflow框架如Airflow、Prefect、甚至LangGraph自身的高级特性通常还提供可视化界面拖拽编排节点直观查看流程。历史与日志记录每一次工作流执行的详细日志和状态便于回溯和审计。并发与队列管理大量工作流实例的执行合理分配资源。重试与警报当某个节点失败时自动重试或发送警报。版本控制对工作流定义进行版本管理。4.2 设计一个包含Agent的Workflow假设我们要实现一个“智能周报生成Workflow”触发每周五下午6点自动触发。收集数据调用多个MCP ServerGit Server、JIRA Server、Calendar Server收集本周代码提交、任务完成和会议信息。Agent分析汇总运行一个LangGraph Agent它接收原始数据分析重点撰写周报草稿。人工审核可选将草稿发送到Slack频道等待负责人确认或修改。最终发布将最终版周报发布到Confluence另一个MCP工具。在这个Workflow中第3步“Agent分析汇总”就是你用LangGraph MCP构建的核心模块。Workflow框架负责调度它、为它准备输入数据、处理它的输出、并管理整个流程的推进。4.3 渐进式披露Skills与Harness架构现在让我们把标题中的两个高级概念串联起来渐进式披露Skills这正是MCP的用武之地。你的Agent在启动时并不需要加载所有可能的工具。根据Workflow的上下文、用户的身份或当前任务阶段动态地连接不同的MCP Server获取当前所需的技能集。例如在处理“财务报销”Workflow时才加载“发票识别”和“财务系统提交”的MCP工具。Harness架构你可以把Harness理解为一个“驾驶舱”或“管控层”。它包裹着你的核心AgentLangGraph并负责技能管理动态加载、卸载MCP工具。输入/输出适配将外部的请求如HTTP API调用、消息队列事件转化为Agent能处理的状态State并将Agent的输出转化为外部系统需要的格式。策略执行限流、鉴权、审计、成本控制例如限制调用某昂贵MCP工具的频率。状态持久化将LangGraph的State存储到数据库实现长对话或异步任务。一个简化的Harness伪代码结构可能是class AgentHarness: def __init__(self, core_graph): self.core_graph core_graph # 编译好的LangGraph App self.mcp_clients {} # 当前活跃的MCP连接 self.state_store RedisStore() # 状态存储 async def handle_request(self, user_input, session_id): # 1. 根据session_id恢复或初始化状态 state self.state_store.load(session_id) or initial_state # 2. 根据当前状态和输入决定需要哪些SkillsMCP工具 required_skills self._determine_skills(state, user_input) # 3. 动态加载Skills连接到对应的MCP Server await self._load_mcp_tools(required_skills) # 4. 将动态加载的工具注入到核心Graph的执行上下文中 # 这需要LangGraph支持动态修改图或通过State传递工具列表 execution_state {**state, “available_tools”: self._get_tool_list()} # 5. 运行核心Agent result_state await self.core_graph.ainvoke(execution_state) # 6. 保存状态清理临时资源 self.state_store.save(session_id, result_state) self._cleanup_tools(required_skills) # 7. 返回响应 return self._format_response(result_state)Harness让你的核心Agent逻辑保持纯净和可测试而所有与环境、资源、管控相关的“脏活累活”都由Harness层处理。这是构建健壮、可扩展的生产级Agent系统的关键。5. 从Demo到生产避坑清单与迭代路径把LangGraph、MCP、Workflow组合起来听起来很美好但直接上生产很容易踩坑。下面是我从实验到落地过程中总结的几点核心建议。5.1 开发阶段先跑通再优化第一步用纯LangGraph实现核心逻辑。先别管MCP和复杂的Workflow。在内存里用硬编码的工具函数把你的Agent任务流图的逻辑跑通。确保状态流转、条件分支、循环控制都是正确的。第二步将1-2个工具抽离为MCP Server。选择那些最独立、最可能复用或最需要安全隔离的工具如数据库查询、发送邮件。实现一个简单的MCP Server并让LangGraph Agent成功调用它。这一步验证解耦的可行性。第三步设计Harness雏形。写一个简单的包装类处理会话ID、状态初始化、调用Agent、返回结果。此时可以引入一个简单的状态存储比如字典或文件。第四步嵌入到Workflow中。将你的HarnessAgent作为一个节点放入一个更大的业务工作流中比如使用Airflow或Prefect。测试它如何被触发、如何处理输入/输出、如何与上下游节点协作。5.2 生产化考量状态存储内存存储只适用于Demo。生产环境需要选择Redis、PostgreSQL或专用向量数据库来持久化State并设计合理的TTL和清理策略。MCP Server治理MCP Server不能成为单点故障。需要考虑服务发现、负载均衡、健康检查。为每个MCP工具调用设置合理的超时和重试。可观测性在Harness层和关键LangGraph节点中加入详细的日志输入、输出、耗时、错误。使用OpenTelemetry等工具集成链路追踪让你能清晰看到一个用户请求流经了哪些MCP Server和Graph节点。测试策略单元测试单个节点函数集成测试整个LangGraph图Mock掉MCP调用端到端测试完整Workflow。版本与回滚对LangGraph图定义、MCP Server接口、Harness配置进行版本控制。确保能快速回滚到稳定版本。5.3 常见问题排查清单当你的Agent表现异常时按这个顺序查输入是否正确检查Harness传递给Agent的初始State格式是否与图定义的State类型匹配。消息内容、参数是否完整图逻辑是否按预期流转在关键节点打印或日志记录state的变化。使用LangGraph Studio可视化执行轨迹。检查条件边conditional_edges的判断逻辑。MCP工具调用成功了吗检查网络连通性、MCP Server日志、工具参数格式。Harness层是否捕获了调用异常并做了恰当处理如重试、降级资源是否耗尽长时间运行的Agent或Workflow可能导致内存泄漏尤其是大语言模型上下文累积。监控进程的内存和CPU使用情况。对于长会话考虑定期清理旧的对话历史。输出格式是否匹配下游检查Agent最终输出的State是否包含了Workflow中下一个节点所需的所有字段格式是否正确最终记住一个原则Agent开发不是魔法而是工程。LangGraph提供了强大的编排能力MCP提供了灵活的扩展能力Workflow提供了可靠的管理能力。但如何将它们有机组合设计出清晰的状态流、稳健的错误处理和高效的工具调度才是真正体现你架构功力的地方。从一个小而美的图开始逐步引入MCP和Harness最终融入业务Workflow这条渐进式路径能帮你更稳地走向成功。