ARTICLE DETAIL

资讯详情

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

LangGraph、MCP与Workflow:构建复杂AI Agent的架构指南

LangGraph、MCP与Workflow:构建复杂AI Agent的架构指南 这次我们来看一个关于 AI Agent 开发的技术框架组合LangGraph、MCP 和 Workflow。如果你还在用简单的 API 调用来构建 Agent感觉功能单一、流程僵化那么这个组合或许能打开新思路。它不是一个具体的“一键启动”工具包而是一套用于构建复杂、可扩展、具备长期记忆和动态工作流能力的 Agent 开发范式。核心在于它解决了传统 Agent 开发的几个痛点状态管理混乱、技能Skills难以复用和组合、与外部工具集成繁琐、以及复杂业务流程的编排困难。通过 LangGraph 定义 Agent 的“大脑”和决策流程通过 MCPModel Context Protocol协议标准化外部工具和数据的接入再结合 Workflow 引擎来编排多步骤、有状态的长期任务最终通过 Harness 这样的架构来统一管理和调度 Skills。本文将带你一次搞懂这三大形态LangGraph, MCP, Workflow如何协同工作并深入解析渐进式披露 Skills 与 Harness 架构的设计理念。你会了解到这套技术栈能做什么、不适合做什么、以及如何着手搭建自己的第一个具备动态工作流能力的 Agent。1. 核心能力速览首先我们通过一个表格快速了解这三大核心组件的定位和关键能力。组件/概念核心定位关键能力解决的问题LangGraphAgent 的状态机与决策引擎1. 基于图Graph定义多节点工作流。2. 管理 Agent 的长期记忆State。3. 支持条件分支、循环、并行等复杂逻辑。4. 与 LLM 深度集成驱动决策。Agent 内部状态流转混乱无法处理多轮、有记忆的复杂对话或任务。MCP (Model Context Protocol)外部工具与数据的统一接入层1. 标准化协议用于连接 LLM 与外部工具如数据库、API、文件系统。2. 动态发现和调用 Skills技能。3. 提供安全的上下文Context管理。每个工具都需要写特定适配代码技能难以复用存在安全风险。Workflow (动态工作流)宏观业务流程的编排器1. 编排多个 Agent 或 Skills 完成一个长期目标。2. 支持人工干预、异常处理、任务重试。3. 可视化定义和管理业务流程。复杂业务场景如客服工单、数据分析流水线需要跨多个步骤和系统的协调。Skills Harness技能的抽象与管理架构1.Skills: 可复用的原子能力单元如搜索、计算、写文件。2.Harness: 技能的运行时容器与管理框架负责加载、调度、监控 Skills。技能代码散落各处缺乏统一的生命周期管理和资源调度。这套组合拳的最终效果是你可以构建一个这样的 Agent——它能记住与用户的整个对话历史LangGraph State根据当前对话状态决定下一步是调用搜索引擎通过 MCP 封装的 Search Skill还是查询数据库另一个 MCP Skill并将一系列调用编排成一个完整的客户服务流程Workflow。所有技能都通过 Harness 统一管理易于扩展和维护。2. 适用场景与使用边界适合谁解决什么问题中级及以上 AI 应用开发者已经熟悉基础 LLM API 调用和简单 Agent 构建希望提升系统的复杂度和可靠性。需要构建“智能助理”类产品如智能客服、个人办公助手、自动化数据分析助手这些场景需要多轮对话、记忆和工具调用。面临复杂业务流程自动化例如一个需求从录入、分析、编码到测试的自动化流水线涉及多个系统和决策点。追求架构清晰和可维护性希望将 Agent 的逻辑、技能实现、业务流程进行解耦方便团队协作和迭代。不适合什么场景一次性或极其简单的 Prompt 工程如果任务只是发送一段提示词给 GPT 并返回结果直接调用 API 更简单快捷。对延迟极其敏感的实时交互多层架构LangGraph决策 - MCP调用 - Workflow编排会引入额外开销可能不适合超低延迟场景。资源极其有限或技术栈锁死的环境引入这套架构需要学习新的概念和库并可能增加系统复杂度。技能Skills完全静态且单一的场景如果 Agent 永远只做一件事且不需要与外部工具交互则无需 MCP 和复杂 Workflow。合规与安全边界技能Skills安全通过 MCP 接入外部工具时必须严格定义权限边界。例如一个“写文件”Skill 应限制其可访问的目录防止任意文件写入漏洞。数据隐私LangGraph 的长期记忆可能存储用户对话历史。必须明确数据存储位置内存/数据库、加密方式、留存期限并遵守相关隐私法规。工具调用审核对于执行删除、支付、发送消息等高风险操作的 Skills应在 Workflow 中设计人工审核节点或二次确认机制。版权与授权Agent 通过 Skills 获取的内容如网页搜索、图片生成需确保使用方式符合版权规定。3. 环境准备与前置条件在开始编码之前你需要准备好开发环境。这套技术栈主要基于 Python 生态。操作系统: Linux (推荐 Ubuntu/Debian), macOS, Windows (WSL2 推荐)。Python: 版本 3.10 或以上。这是 LangGraph 等库的主流支持版本。包管理工具: 使用pip或poetry管理依赖。强烈建议使用虚拟环境venv或conda。LLM 访问权限: 你需要一个大型语言模型的 API 密钥例如OpenAI GPT 系列 (GPT-4o, GPT-4 Turbo)Anthropic Claude 系列 (Claude 3)或开源的本地模型通过ollama,vLLM等部署但本地模型对智能体决策能力要求较高。基础概念理解:了解基本的 Prompt Engineering。了解 Agent 的基本概念ReAct, Plan-and-Execute 等。对异步编程 (async/await) 有基本了解会更佳。可选工具:Docker: 用于容器化部署 Skills 或整个 Harness。数据库: 如果需要持久化 LangGraph 的状态State可能需要 Redis 或 PostgreSQL。4. 安装部署与启动方式这不是一个“一键启动”的桌面应用而是一套需要编码集成的开发框架。因此部署的核心是安装必要的 Python 库和设置配置。4.1 核心库安装创建一个新的虚拟环境然后安装核心依赖# 创建并激活虚拟环境 (以 venv 为例) python -m venv agent-env source agent-env/bin/activate # Linux/macOS # agent-env\Scripts\activate # Windows # 安装 LangChain 和 LangGraph。LangGraph 通常与 LangChain 协同使用。 pip install langchain langgraph # 安装 LangChain 社区包包含许多现成的工具和集成 pip install langchain-community # 如果你计划使用 OpenAI 模型 pip install openai # 安装 MCP 相关的库。MCP 的 Python SDK 可用于创建和运行 MCP 服务器。 # 注意MCP 生态正在快速发展包名可能变化请以官方文档为准。 # 假设官方包名为 model-context-protocol pip install model-context-protocol # 用于构建 HTTP 类 Skills 或 Harness pip install fastapi uvicorn4.2 项目结构初始化一个清晰的项目结构有助于管理复杂的 Agent 系统。建议如下your_agent_project/ ├── skills/ # 存放所有 Skills │ ├── __init__.py │ ├── calculator.py # 计算器 Skill │ ├── web_search.py # 网络搜索 Skill (通过 MCP 暴露) │ └── weather.py # 天气查询 Skill ├── workflows/ # 存放 LangGraph 工作流定义 │ ├── __init__.py │ └── customer_support.py ├── harness/ # Harness 运行时管理 │ ├── __init__.py │ ├── app.py # FastAPI 主应用 │ └── skill_manager.py # Skill 加载与管理 ├── config.py # 配置文件 (API Keys, 模型设置) ├── requirements.txt └── main.py # 应用入口启动 LangGraph Agent4.3 启动一个最简单的 LangGraph Agent让我们先绕过 MCP 和复杂 Workflow创建一个具备内存的最基础 LangGraph Agent 来验证环境。# main.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph.message import add_messages # 1. 定义状态 State class AgentState(TypedDict): messages: Annotated[list, add_messages] # 自动管理消息历史 turn_count: int # 自定义状态记录对话轮数 # 2. 初始化 LLM llm ChatOpenAI(modelgpt-4o, temperature0) # 3. 定义节点函数 def call_model(state: AgentState): 调用 LLM 生成回复 messages state[messages] response llm.invoke(messages) return {messages: [response], turn_count: state[turn_count] 1} def should_continue(state: AgentState): 条件判断根据最后一条消息决定是否结束 last_message state[messages][-1] # 简单逻辑如果用户说“结束”则停止 if isinstance(last_message, HumanMessage) and 结束 in last_message.content: return END return call_model # 否则继续调用模型 # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(call_model, call_model) workflow.set_entry_point(call_model) workflow.add_conditional_edges( call_model, should_continue, {END: END, call_model: call_model} # 条件映射 ) # 5. 编译图 app workflow.compile() # 6. 运行 Agent if __name__ __main__: # 初始状态 initial_state AgentState(messages[HumanMessage(content你好我是小明。)], turn_count0) # 执行图 for event in app.stream(initial_state, stream_modevalues): if call_model in event: print(fAI: {event[call_model][messages][-1].content}) print(f当前轮次: {event[call_model][turn_count]})运行python main.py你会看到 AI 的回复和状态更新。这证明了你的 LangGraph 基础环境是正常的。5. 功能测试与效果验证我们将分步验证三大核心形态的能力。5.1 验证 LangGraph 的状态管理与循环目标测试 Agent 是否能记住对话历史并进行多轮交互。测试脚本使用上面main.py的代码。操作修改main.py末尾的交互部分模拟一个简单对话循环。# ... 省略前面的编译代码 ... app workflow.compile() # 模拟对话 test_messages [ “你好我叫张三。”, “我的名字是什么”, “结束” ] state AgentState(messages[], turn_count0) for msg in test_messages: state[messages].append(HumanMessage(contentmsg)) print(f用户: {msg}) # 执行一步 result app.invoke(state) state result # 更新状态 ai_msg result[messages][-1] if isinstance(ai_msg, AIMessage): print(fAI: {ai_msg.content}) print(---)预期结果AI 应该能在第二轮回答出用户的名字是“张三”证明状态消息历史被正确传递和记忆。成功标准AI 的第二轮回复中包含“张三”。turn_count状态正确递增。失败排查检查AgentState定义中add_messages注解是否正确检查 LLM 调用是否成功检查状态更新逻辑。5.2 验证 MCP Skill 的集成与调用目标测试如何将一个简单的计算器功能封装为 MCP Skill并被 LangGraph Agent 调用。创建 MCP Server (Skill)我们创建一个提供加法运算的 MCP Server。# skills/calculator_mcp_server.py from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types import asyncio # 创建 Server 实例 server Server(calculator-server) # 定义工具即 Skill server.list_tools() async def handle_list_tools() - list[types.Tool]: return [ types.Tool( nameadd_numbers, descriptionAdd two numbers together., inputSchema{ type: object, properties: { a: {type: number, description: The first number}, b: {type: number, description: The second number}, }, required: [a, b], }, ) ] # 实现工具逻辑 server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) - list[types.TextContent]: if name add_numbers: a arguments[a] b arguments[b] result a b return [types.TextContent(typetext, textstr(result))] raise ValueError(fUnknown tool: {name}) # 运行 Server (通过 stdio 通信这是 MCP 的常见方式) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namecalculator, server_version0.1.0, ), NotificationOptions(), ) if __name__ __main__: asyncio.run(main())在 LangGraph 中调用该 Skill这需要用到 LangChain 的 MCP 集成。假设我们通过一个MCPToolkit来加载这个 server。# 在 main.py 中集成 from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_community.tools.mcp import create_mcp_toolkit from langgraph.prebuilt import create_react_agent # 1. 创建 MCP 工具包 (需要指定 MCP Server 的路径或配置) # 注意此处为示例实际连接方式可能因 LangChain 版本和 MCP 实现而异 toolkit await create_mcp_toolkit( server_namecalculator, # 这里需要配置如何连接到上面运行的 calculator_mcp_server # 可能是命令行路径、端口或 stdio 配置 ) tools toolkit.get_tools() # 获取到 add_numbers 工具 # 2. 创建 LangGraph Agent (使用 ReAct 模式) llm ChatOpenAI(modelgpt-4o, temperature0) agent create_react_agent(llm, tools) # 3. 执行 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: “请计算 123 加 456 等于多少”}) print(result[output])预期结果Agent 应能识别出需要计算调用add_numbers工具并返回结果579。成功标准控制台日志显示调用了add_numbers工具并输出了正确结果。失败排查确保 MCP Server 已启动并正常运行检查 LangChain 中 MCP 集成的配置是否正确检查网络或进程间通信是否畅通。5.3 验证动态 Workflow 编排目标测试一个包含条件判断和多个步骤的 Workflow例如根据用户查询决定是搜索网络还是查询本地数据库。设计 Workflow节点1:route_query- 分析用户意图路由到search_web或query_db。节点2:search_web- 调用网络搜索 Skill (模拟)。节点3:query_db- 调用数据库查询 Skill (模拟)。节点4:generate_response- 综合结果生成最终回复。LangGraph 实现from langgraph.graph import StateGraph, END from typing import Literal class WorkflowState(TypedDict): query: str intent: Literal[search, query_db, unknown] # 路由意图 search_result: str db_result: str final_answer: str def route_query(state: WorkflowState): 路由节点简单根据关键词判断 query state[query].lower() if “最新消息” in query or “新闻” in query: return {intent: search} elif “用户资料” in query or “订单” in query: return {intent: query_db} else: return {intent: unknown} def search_web(state: WorkflowState): 模拟网络搜索 # 这里应调用真实的 MCP Search Skill print(f“模拟搜索: {state[query]}”) return {search_result: f“关于{state[query]}的模拟搜索结果。”} def query_database(state: WorkflowState): 模拟数据库查询 print(f“模拟查询数据库: {state[query]}”) return {db_result: f“用户 张三 的模拟订单数据。”} def generate_response(state: WorkflowState): 生成最终回复 if state[intent] search: source state[search_result] elif state[intent] query_db: source state[db_result] else: source “无法处理该类型查询。” answer f“根据您的查询『{state[query]}』我为您找到以下信息\n{source}” return {final_answer: answer} # 构建图 workflow StateGraph(WorkflowState) workflow.add_node(“router”, route_query) workflow.add_node(“searcher”, search_web) workflow.add_node(“querier”, query_database) workflow.add_node(“responder”, generate_response) workflow.set_entry_point(“router”) # 条件边根据 router 输出的 intent 决定下一步 workflow.add_conditional_edges( “router”, lambda state: state[intent], # 根据 intent 字段路由 { “search”: “searcher”, “query_db”: “querier”, “unknown”: “responder”, # 未知意图直接生成回复 } ) # 普通边 workflow.add_edge(“searcher”, “responder”) workflow.add_edge(“querier”, “responder”) workflow.add_edge(“responder”, END) app workflow.compile() # 测试 test_queries [“今天有什么科技新闻”, “查询我的订单状态”, “你好吗”] for q in test_queries: print(f\n 处理查询: {q}) result app.invoke({query: q}) print(f结果: {result[final_answer]})预期结果对于不同的查询Workflow 会走不同的分支并调用对应的模拟技能最后生成包含不同来源信息的回复。成功标准控制台输出显示正确的路由路径“模拟搜索”或“模拟查询数据库”和最终回复。失败排查检查route_query函数的逻辑是否正确检查节点之间的边是否连接正确检查状态字段的读写是否一致。6. 接口 API 与批量任务6.1 将 LangGraph Agent 暴露为 HTTP API一个实用的 Agent 系统需要提供 API 供其他服务调用。我们可以用 FastAPI 快速包装上面编译好的app。# harness/app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .your_langgraph_module import app as agent_app # 导入编译好的 LangGraph app import asyncio app FastAPI(titleAI Agent API) class QueryRequest(BaseModel): message: str session_id: str | None None # 用于区分不同会话/用户 class QueryResponse(BaseModel): response: str session_id: str app.post(/chat, response_modelQueryResponse) async def chat_endpoint(request: QueryRequest): try: # 这里需要根据 session_id 获取或创建对应的 Agent 状态 # 简化示例每次都是新的会话 initial_state {messages: [{role: user, content: request.message}]} # 流式或非流式调用 final_state agent_app.invoke(initial_state) # 从最终状态中提取最后一条 AI 消息 last_message ... # 提取逻辑取决于你的状态结构 response_text last_message[content] if isinstance(last_message, dict) else str(last_message) return QueryResponse(responseresponse_text, session_idrequest.session_id or default) except Exception as e: raise HTTPException(status_code500, detailfAgent execution failed: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务后即可通过POST /chat接口与 Agent 交互。6.2 批量任务处理对于需要处理大量独立任务的场景如批量分析文档、处理客服日志可以设计一个任务队列。# batch_processor.py import asyncio import json from typing import List from harness.app import agent_app # 导入你的 Agent async def process_batch(task_list: List[str], output_file: str): 批量处理任务列表结果写入文件 results [] for i, task in enumerate(task_list): print(fProcessing task {i1}/{len(task_list)}: {task[:50]}...) try: state {input: task} # 调用你的 LangGraph workflow result await asyncio.to_thread(agent_app.invoke, state) # 提取结果 processed_result extract_result(result) # 自定义提取函数 results.append({ id: i, input: task, output: processed_result, status: success }) except Exception as e: results.append({ id: i, input: task, error: str(e), status: failed }) # 可选添加延迟避免对上游服务造成压力 # await asyncio.sleep(0.1) # 写入结果 with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(fBatch processing completed. Results saved to {output_file}) # 使用示例 if __name__ __main__: tasks [ “分析一下这份合同的主要风险点。”, “总结这篇技术文章的核心思想。”, “将以下英文翻译成中文: Hello, world!”, ] asyncio.run(process_batch(tasks, “batch_results.json”))关键点异步与线程使用asyncio.to_thread将同步的invoke调用放入线程池避免阻塞事件循环。错误处理每个任务独立处理单个失败不影响整体。结果持久化及时保存结果防止进程崩溃导致数据丢失。速率限制根据调用的模型 API 或自身服务能力添加适当的延迟 (asyncio.sleep)。7. 资源占用与性能观察由于这是一个开发框架而非预训练模型其“资源占用”主要体现在运行时的内存、CPU 以及对外部服务如 LLM API、数据库、MCP Servers的调用开销上。内存占用LangGraph State如果使用内存存储对话状态内存占用随会话长度和状态复杂度线性增长。对于高并发建议将状态存储到外部数据库如 Redis。Python 进程基础的 LangGraph、LangChain 应用进程内存通常在几百 MB。每增加一个复杂的 Skill 或加载大模型内存会增加。监控方法使用psutil库或系统命令如top,htop监控 Python 进程的RES(常驻内存)。CPU 占用主要消耗在本地计算如图处理、文本处理和序列化/反序列化上。纯编排和调用远程 API 的 AgentCPU 占用通常不高。高 CPU 场景如果 Skills 包含本地密集计算如本地嵌入模型计算、图像处理。网络 I/O 与延迟主要延迟源调用远程 LLM API如 GPT-4的延迟通常是最大的可能从几百毫秒到数秒。MCP 调用如果 MCP Server 是本地进程stdio开销很小如果是远程 HTTP Server则增加网络往返。性能观察为关键节点LLM 调用、工具调用添加计时日志。import time def call_model_with_timing(state): start time.time() result call_model(state) # 你的 LLM 调用函数 end time.time() print(fLLM call took {end - start:.2f} seconds) return result优化建议缓存对频繁且结果不变的 LLM 调用或工具调用结果进行缓存。异步在 Workflow 中如果多个节点间没有依赖可以考虑使用asyncio.gather并行执行。状态存储外置将会话状态存储到 Redis 等外部高速缓存避免进程内存无限增长。精简 Skills只加载当前 Workflow 必需的 Skills延迟加载其他 Skills。8. 常见问题与排查方法在开发过程中你可能会遇到以下典型问题问题现象可能原因排查方式解决方案LangGraph 状态不更新或丢失1. State 定义中Annotated字段的 reducer 函数如add_messages使用错误。2. 节点函数返回值没有包含所有需要更新的状态字段。1. 打印每个节点执行前后的完整 State。2. 检查 reducer 函数逻辑。1. 确保 State 定义正确参考官方文档。2. 节点函数必须返回一个字典包含要更新的所有字段即使值不变也要返回原值。MCP Server 连接失败1. MCP Server 进程未启动或崩溃。2. 通信协议stdio/HTTP配置错误。3. 防火墙或端口问题。1. 检查 MCP Server 进程是否在运行。2. 查看 LangChain 端连接 MCP 的配置参数。3. 检查stderr输出。1. 确保先启动 MCP Server。2. 核对create_mcp_toolkit的参数如 server 命令路径、端口等。3. 使用简单的测试客户端验证 MCP Server 本身是否正常。LLM 不调用 Tools/Skills1. LLM 的temperature可能太高导致输出不稳定。2. 提供给 LLM 的 Tool 描述不清晰。3. Prompt 没有正确引导 Agent 使用工具。1. 设置temperature0测试。2. 查看 LLM 接收到的完整 Prompt 和 Tool 定义。3. 使用verboseTrue查看 Agent 的思考链。1. 调整 LLM 参数。2. 优化 Tool 的name和description使其更精确。3. 使用 LangChain 提供的标准 Agent 模板如create_react_agent。Workflow 陷入无限循环1. 条件边 (conditional_edges) 的判断逻辑有误始终无法跳转到END。2. 图结构存在循环但没有终止条件。1. 在条件判断函数中打印日志。2. 可视化你的 Graph 结构检查循环路径。1. 仔细检查条件判断函数的返回值确保所有可能状态都有对应的边。2. 为循环设置最大迭代次数。LangGraph 可以通过在 State 中设置计数器并在条件边中检查来实现。API 服务并发性能差1. LLM API 调用是同步阻塞的。2. Agent 执行是 CPU 密集或 I/O 阻塞操作。1. 使用async版本的 LLM 客户端和工具。2. 使用httpx.AsyncClient等异步 HTTP 客户端。3. 使用性能分析工具如py-spy。1. 将 Agent 执行放入线程池asyncio.to_thread或使用完全异步的 Agent 实现。2. 增加 API 服务的 worker 数量如 Uvicorn workers。3. 对慢速 Skills 进行超时设置和重试。Skills 加载冲突或依赖缺失1. 不同 Skills 有相同的 Python 包依赖但版本冲突。2. Skill 代码运行时找不到模块。1. 检查pip list确认包版本。2. 查看 Skill 启动时的 ImportError 日志。1. 使用虚拟环境或 Docker 隔离不同 Skills 的依赖。2. 在 Harness 中统一管理公共依赖或为每个 Skill 提供独立的依赖清单。9. 最佳实践与使用建议从简到繁渐进式开发第一步先用 LangGraph 实现一个带记忆的简单对话 Agent。第二步引入 1-2 个本地 MCP Skills如计算器、时间查询。第三步设计一个包含条件分支的简单 Workflow。第四步集成远程 API Skills 和复杂业务流程。第五步构建 Harness实现 Skills 的动态发现和生命周期管理。状态设计要精简LangGraph 的 State 应只包含必要的数据。避免将整个会话历史的所有原始消息都塞进去可以考虑只存储摘要或关键信息。Skills 设计原则单一职责一个 Skill 只做一件事。明确接口通过 MCP 协议定义清晰的输入输出 Schema。错误处理Skill 内部应妥善处理异常并返回结构化的错误信息而不是直接崩溃。资源管理对于有状态的 Skill如数据库连接要在 Harness 中管理其初始化和销毁。Workflow 设计原则可观测性为每个节点添加详细的日志记录输入、输出和耗时。可恢复性考虑工作流执行中断后如何恢复。可以为关键节点实现幂等性。超时与重试对调用外部服务LLM、API的节点设置超时和重试机制。安全与合规输入验证对所有来自外部的输入用户输入、API 参数进行严格的验证和清洗。权限控制在 Harness 层或 Skill 层实现基于角色或上下文的权限检查。审计日志记录所有 Tool 调用和关键决策便于事后审计和问题排查。10. 总结与下一步LangGraph MCP Workflow 这套组合为构建下一代复杂 AI Agent 提供了强大的底层架构。它的价值不在于提供一个开箱即用的“黑盒”而在于提供了一套清晰、灵活、可扩展的“乐高积木”。最值得尝试的点用 LangGraph 管理复杂状态告别混乱的全局变量用有向图来清晰定义 Agent 的思维流。用 MCP 统一技能接入像插件一样即插即用各种能力让 Agent 的“工具箱”可以无限扩展。用 Workflow 编排长期任务将单次问答升级为可持续数天、包含人工节点的业务流程自动化。最先应该验证的功能 按照本文第 5 节的顺序从“带记忆的 LangGraph 对话”开始成功后再尝试集成一个最简单的 MCP Skill如计算器最后设计一个包含路由判断的双分支 Workflow。这三个小实验能帮你快速建立对整套体系的理解。最容易踩的坑状态管理LangGraph State 的更新机制需要仔细理解新手容易在这里出错导致状态丢失。MCP 连接MCP 的通信方式stdio/HTTP和配置需要对照官方示例仔细调试。LLM 提示工程即使架构再好如果引导 LLM 使用工具的 Prompt 没写好Agent 也可能表现不佳。后续扩展方向探索可视化使用LangGraph Studio来可视化和调试你的工作流图。集成向量数据库将 LangGraph 的长期记忆与向量检索结合实现更强大的记忆和上下文管理。实现复杂的 Harness开发一个能够动态加载、卸载、监控和负载均衡 Skills 的完整运行时框架。投入生产考虑容器化部署、配置管理、监控告警、以及如何与现有的微服务体系集成。这套架构的学习曲线确实比直接调用 API 要陡峭但它带来的清晰度、可维护性和扩展性对于构建严肃的、生产级的 AI 应用来说是值得的。建议收藏本文在动手搭建时作为参考路线图。
返回列表