大家好我是专注于AI应用开发与架构探索的技术博主。在构建复杂AI Agent时你是否也遇到过这样的困境Agent的逻辑散落在各处状态管理混乱扩展新能力Skills需要大动干戈与外部工具Tools的集成更是繁琐不堪如果你还在为Agent开发仅仅是调用不同API的“胶水代码”而感到困惑那么这篇文章正是为你准备的。本文将深入解析构建现代、可维护、高扩展性AI Agent的三大核心形态LangGraph状态与流程编排、MCP模型上下文协议与Workflow工作流引擎。我们将超越简单的API调用从架构层面探讨如何通过“渐进式披露”思想优雅地管理Skills并理解Harness治理框架在其中扮演的关键角色。无论你是希望从零构建一个智能助手还是想优化现有Agent项目的架构这篇文章都将提供一套完整的、可落地的思路与实战示例。1. 背景与核心概念为什么Agent开发不止于API调用在AI应用开发的早期我们常常将Agent简单地视为一个“智能路由器”接收用户输入调用一个大语言模型LLMAPI然后根据返回结果再调用一些工具如搜索、计算、数据库查询最后将结果返回给用户。这种模式虽然直接但很快会暴露出诸多问题状态管理困难多轮对话中Agent需要记住历史、维护任务状态如“正在订票中”简单的函数调用难以优雅处理。逻辑耦合严重业务逻辑、工具调用、LLM提示词Prompt全都混在一起代码像“意大利面条”难以维护和测试。扩展性差每增加一个新能力Skill如“发送邮件”或“分析图表”都可能需要修改核心调度逻辑。工具集成繁琐为不同来源、不同协议的工具本地函数、HTTP API、数据库、命令行编写统一的适配层工作量巨大。为了解决这些问题社区和业界提出了更系统的架构模式。我们需要理解几个关键概念Agent智能体一个能够感知环境、进行决策并执行动作以实现目标的自治系统。在代码层面它是一个协调中心。Skill技能Agent所具备的独立能力单元。例如“天气查询”、“代码生成”、“文档总结”都是一个独立的Skill。Skill应该是模块化、可复用的。Tool工具Skill执行其功能时所依赖的具体手段。一个Skill可能调用一个或多个Tools。例如“天气查询”Skill可能依赖“调用天气API”这个Tool。Workflow工作流将多个步骤可能是LLM调用、Skill执行、条件判断等按照特定顺序和规则组织起来以完成一个复杂任务的流程。State状态在Workflow执行过程中需要传递和更新的数据。例如用户的查询、LLM的回复、中间计算结果、当前步骤等。本文要探讨的三大形态正是为了解决上述痛点而生的不同层面的解决方案。2. 环境准备与版本说明在开始实战之前我们需要搭建一个统一的开发环境。本文的示例将主要使用Python生态下的工具因为它们在此领域最为活跃和成熟。核心环境与版本操作系统macOS / Linux / Windows (WSL2推荐)Python版本 3.10 强烈推荐3.10以获得更好的类型提示支持包管理工具pip或poetry(本文使用pip示例)虚拟环境强烈建议使用venv或conda创建独立环境。核心依赖库我们将安装三个核心库分别对应三大形态的典型代表或基础工具。# 创建并激活虚拟环境以venv为例 python -m venv .venv source .venv/bin/activate # Linux/macOS # .venv\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain langchain-community # langgraph: 用于构建有状态的、多环节的Agent工作流。 # langchain: 提供了构建Agent所需的基础组件LLM、Tools、Chains等。 # langchain-community: 包含大量社区贡献的第三方工具和集成。 # 安装MCP相关的客户端库示例使用一个模拟MCP Server的客户端 pip install mcp-client # 注意正式的MCP SDK可能仍在演进此处使用一个概念性客户端进行演示。 # 安装一个简单的工作流引擎示例使用Prefect因其轻量且易理解 pip install prefectLLM配置本文示例需要接入一个LLM。你可以使用OpenAI GPT、Anthropic Claude或开源的Ollama本地模型。这里以OpenAI为例pip install openai然后你需要设置环境变量OPENAI_API_KEY。export OPENAI_API_KEYyour-api-key-here # 或在代码中通过 os.environ[“OPENAI_API_KEY”] ‘your-key’ 设置版本说明AI领域库的版本迭代非常快。本文的代码示例基于langgraph0.0.40,langchain0.1.0等较新版本编写。如果你的环境存在差异请关注官方文档调整导入语句或API调用方式。核心架构思想是通用的。3. 核心形态一LangGraph —— 有状态的工作流编排引擎LangGraph是LangChain团队推出的一个库用于构建有状态、多参与者的图计算应用。它完美契合了复杂Agent对状态生命周期管理和循环控制流的需求。3.1 LangGraph 是什么与LangChain的区别你可以把LangGraph想象成专门为Agent设计的“流程图绘制与执行引擎”。它允许你定义节点Node和边Edge节点执行具体操作如调用LLM、运行工具边决定下一步走向基于状态。与LangChain的核心区别LangChain更像一个“组件工具箱”提供了LLM、Prompt、Memory、Tool、Chain等标准化零件。它擅长构建线性的、无状态的链Chain。LangGraph是一个“编排框架”专注于管理带有复杂状态和循环的图Graph。它使用LangChain的组件作为节点但负责更复杂的流程控制。简单说LangChain提供砖瓦LangGraph设计并建造有多个房间和回廊的房子。3.2 核心概念State、Node、EdgeState一个字典或Pydantic模型贯穿整个图执行过程的所有数据都存储在这里。例如{“messages”: [], “next”: “action”}。Node一个函数接收当前State执行一些操作如调用LLM并返回更新后的State。Edge决定从当前节点结束后下一个该执行哪个节点。可以是固定的也可以是基于State内容动态决定的conditional edge。3.3 实战用LangGraph构建一个基础ReAct AgentReActReasoning Acting是Agent的经典范式思考调用LLM生成推理和动作- 执行调用Tool- 观察获取Tool结果- 循环直到得出最终答案。下面我们构建一个能进行多步数学运算和搜索的ReAct Agent。# 文件react_agent_graph.py from typing import TypedDict, Annotated, Sequence import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.messages import HumanMessage, AIMessage import math # 1. 定义State结构 class AgentState(TypedDict): # 消息历史 messages: Annotated[Sequence, operator.add] # 临时存储上一步的思考 scratchpad: str # 最终答案如果有 final_answer: str # 2. 创建LLM和Tools llm ChatOpenAI(model“gpt-4o-mini”, temperature0) # 定义一些简单的工具 def multiply(a: float, b: float) - float: “”“两个数相乘。”“” return a * b def divide(a: float, b: float) - float: “”“两个数相除b不能为0。”“” if b 0: return “Error: Division by zero” return a / b math_tools [ Tool(name“Multiply”, funcmultiply, description“Useful for multiplying two numbers.”), Tool(name“Divide”, funcdivide, description“Useful for dividing two numbers. Second argument cannot be zero.”), ] search_tool DuckDuckGoSearchRun() # 将所有工具组合 all_tools math_tools [Tool(name“Search”, funcsearch_tool.run, description“Useful for searching the web for current information.”)] # 3. 定义节点函数 def reason_node(state: AgentState) - AgentState: “”“思考节点分析历史决定下一步是回答还是行动。”“” messages state[“messages”] # 构建给LLM的Prompt包含工具描述和历史 tool_descriptions “\n”.join([f”{t.name}: {t.description}” for t in all_tools]) prompt f”””You are a helpful assistant. You have access to the following tools: {tool_descriptions} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{‘, ‘.join([t.name for t in all_tools])}] Action Input: the input to the action Observation: the result of the action … (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original question Begin! {‘\n’.join([msg.content for msg in messages[-5:]])} # 取最近5条消息作为上下文 “”” # 调用LLM response llm.invoke([HumanMessage(contentprompt)]) response_content response.content # 解析LLM的回复 lines response_content.split(‘\n’) thought “” action None action_input None for line in lines: if line.startswith(‘Thought:’): thought line.replace(‘Thought:’, “”).strip() elif line.startswith(‘Action:’): action line.replace(‘Action:’, “”).strip() elif line.startswith(‘Action Input:’): action_input line.replace(‘Action Input:’, “”).strip() # 更新State new_messages state[“messages”] [AIMessage(contentresponse_content)] if “Final Answer:” in response_content: # 如果LLM直接给出了最终答案 final_answer response_content.split(“Final Answer:”)[-1].strip() return {“messages”: new_messages, “scratchpad”: thought, “final_answer”: final_answer} else: # 否则准备执行动作 return {“messages”: new_messages, “scratchpad”: thought, “action”: action, “action_input”: action_input} def act_node(state: AgentState) - AgentState: “”“行动节点执行工具调用。”“” action state.get(“action”) action_input state.get(“action_input”) observation “No action specified.” if action and action_input: # 找到对应的工具 tool_to_use next((t for t in all_tools if t.name action), None) if tool_to_use: try: # 简单解析输入实际应用需要更健壮的解析 observation str(tool_to_use.run(action_input)) except Exception as e: observation f”Error executing {action}: {str(e)}” else: observation f”Unknown action: {action}” # 将观察结果添加到消息历史 observation_msg f”Observation: {observation}” new_messages state[“messages”] [HumanMessage(contentobservation_msg)] return {“messages”: new_messages, “scratchpad”: state[“scratchpad”], “observation”: observation} # 4. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“reason”, reason_node) workflow.add_node(“act”, act_node) # 设置入口点 workflow.set_entry_point(“reason”) # 添加边思考后根据是否有final_answer决定是结束还是行动 def decide_next_step(state: AgentState) - str: if state.get(“final_answer”): return “end” else: return “act” workflow.add_conditional_edges( “reason”, decide_next_step, {“end”: END, “act”: “act”} ) # 行动后总是回到思考节点 workflow.add_edge(“act”, “reason”) # 编译图 app workflow.compile() # 5. 运行Agent if __name__ “__main__”: # 初始化状态 initial_state: AgentState {“messages”: [HumanMessage(content“What is the result of 15 multiplied by 8? And then what‘s the capital of France?”)], “scratchpad”: “”, “final_answer”: “”} # 执行图 final_state app.invoke(initial_state, config{“recursion_limit”: 10}) # 限制递归深度 print(“\n Final Answer ) print(final_state.get(“final_answer”, “No final answer found.”)) print(“\n Full Message History ) for msg in final_state[“messages”]: print(f”{type(msg).__name__}: {msg.content[:200]}…“ if len(msg.content) 200 else f”{type(msg).__name__}: {msg.content}”)这个例子展示了LangGraph如何清晰地管理ReAct的循环reason - (act - reason)* - end。状态AgentState在整个流程中传递和更新逻辑清晰。4. 核心形态二MCP —— 统一的工具集成协议MCPModel Context Protocol是由Anthropic提出的一种开放协议旨在标准化LLM与外部工具、数据源之间的通信方式。它解决了工具集成中的“烟囱”问题。4.1 MCP 解决了什么问题在没有MCP之前每个AI应用项目都需要为每个外部工具数据库、API、文件系统编写特定的客户端代码、认证逻辑和错误处理。这导致了重复劳动每个项目都要重写相似的集成代码。供应商锁定代码与特定工具的实现深度耦合。安全性挑战每个工具都需要单独处理权限和密钥管理。MCP通过定义一套标准的Server-Client模型让工具提供者实现一个MCP Server而任何兼容MCP的AI应用作为MCP Client都可以无缝发现和使用这些工具。4.2 MCP 核心架构MCP Server包装了一个或多个工具或数据源对外提供标准的清单列出可用工具和执行接口。MCP Client通常是AI应用框架如LangChain、Claude Desktop它连接到一个或多个MCP Server动态获取工具列表并在需要时调用它们。SSE/Stdio 传输层MCP Server和Client之间通过Server-Sent Events (SSE) 或标准输入输出Stdio进行通信。4.3 实战模拟一个MCP Server并与LangGraph集成由于正式的MCP Python SDK仍在快速迭代我们通过一个高度简化的模拟来理解其思想我们将创建一个“工具提供者”类它模拟MCP Server的核心功能——注册和调用工具。# 文件mcp_simulated_server.py from typing import Any, Dict, List, Callable import json class SimulatedMCPServer: “”“一个模拟的MCP Server管理一组工具。”“” def __init__(self): self._tools: Dict[str, Callable] {} self._tool_descriptions: List[Dict] [] def register_tool(self, name: str, func: Callable, description: str, input_schema: Dict[str, Any] None): “”“注册一个工具。”“” self._tools[name] func tool_desc { “name”: name, “description”: description, “inputSchema”: input_schema or {“type”: “object”, “properties”: {}} # 简化schema } self._tool_descriptions.append(tool_desc) def list_tools(self) - List[Dict]: “”“列出所有可用工具类似MCP的list_tools。”“” return self._tool_descriptions def call_tool(self, name: str, arguments: Dict[str, Any]) - Any: “”“调用一个工具类似MCP的call_tool。”“” if name not in self._tools: raise ValueError(f”Tool ‘{name}’ not found.”) return self._tools[name](**arguments) # 创建并配置一个模拟Server mcp_server SimulatedMCPServer() # 注册一些工具 def get_weather(city: str) - str: # 模拟API调用 return f”The weather in {city} is sunny, 25°C.” def create_calendar_event(title: str, start_time: str) - str: # 模拟创建日历事件 return f”Calendar event ‘{title}’ created at {start_time}.” mcp_server.register_tool( name“get_weather”, funcget_weather, description“Get the current weather for a given city.”, input_schema{“type”: “object”, “properties”: {“city”: {“type”: “string”}}} ) mcp_server.register_tool( name“create_calendar_event”, funccreate_calendar_event, description“Create a new calendar event.”, input_schema{“type”: “object”, “properties”: {“title”: {“type”: “string”}, “start_time”: {“type”: “string”}}} ) # 模拟Client端动态获取并使用工具 print(“ Available Tools from MCP Server ) tools mcp_server.list_tools() for tool in tools: print(f”- {tool[‘name’]}: {tool[‘description’]}”) print(“\n Calling a Tool ) result mcp_server.call_tool(“get_weather”, {“city”: “Beijing”}) print(f”Result: {result}”)现在我们将这个模拟的MCP Server集成到之前的LangGraph ReAct Agent中。关键在于修改reason_node和act_node使其工具列表来源于MCP Server而不是硬编码。# 文件langgraph_with_mcp.py # … (之前的导入和AgentState定义保持不变) from mcp_simulated_server import mcp_server # 导入我们模拟的Server # 修改 reason_node 中的工具列表来源 def reason_node_with_mcp(state: AgentState) - AgentState: “”“思考节点工具列表动态从MCP Server获取。”“” messages state[“messages”] # 关键变化从MCP Server动态获取工具描述 available_tools mcp_server.list_tools() tool_descriptions “\n”.join([f”{t[‘name’]}: {t[‘description’]}” for t in available_tools]) tool_names [t[‘name’] for t in available_tools] prompt f”””You are a helpful assistant. You have access to the following tools: {tool_descriptions} … (其余Prompt与之前相同使用tool_names) … “”” # … (后续LLM调用和解析逻辑与之前完全相同) … # 假设我们解析出了 action 和 action_input return {“messages”: new_messages, “scratchpad”: thought, “action”: action, “action_input”: action_input} def act_node_with_mcp(state: AgentState) - AgentState: “”“行动节点通过MCP Server调用工具。”“” action state.get(“action”) action_input_str state.get(“action_input”, “”) observation “No action specified.” if action and action_input_str: try: # 关键变化通过MCP Server调用工具 # 需要将字符串类型的action_input解析成字典。这里做简单处理。 # 实际中LLM应返回JSON字符串或使用更复杂的解析器。 import ast # 尝试解析为字典例如{“city”: “London”} try: arguments ast.literal_eval(action_input_str) except: # 如果不是字典假设是单个参数例如“London” arguments {“input”: action_input_str} # 更优做法是根据MCP Server注册的schema来构造参数 # 这里为简化假设工具接受一个名为input的参数或单个位置参数 # 调用MCP Server observation str(mcp_server.call_tool(action, arguments)) except Exception as e: observation f”Error executing {action} via MCP: {str(e)}” # … (后续更新State的逻辑与之前相同) … return {“messages”: new_messages, “scratchpad”: state[“scratchpad”], “observation”: observation} # 然后使用 reason_node_with_mcp 和 act_node_with_mcp 重新构建LangGraph图。 # … (构建图的代码与之前类似) …通过这种集成我们的Agent不再硬编码工具。任何实现了MCP协议的工具Server如一个提供数据库查询的Server、一个提供内部CRM API的Server都可以被Agent动态发现和使用实现了工具层的解耦和标准化。5. 核心形态三Workflow —— 声明式的业务流程管理Workflow工作流引擎关注的是更高层次的业务流程编排。它通常用于处理长时间运行、包含人工审批、错误重试、并行执行等复杂逻辑的任务。LangGraph本身也是一种Workflow引擎专注于Agent循环但这里我们讨论更通用的、声明式的Workflow系统如Prefect、Airflow、Temporal。5.1 为什么需要独立的Workflow引擎当Agent的任务变得非常复杂例如“监控数据 - 生成报告 - 发送邮件审批 - 根据审批结果更新数据库 - 通知用户”这已经超出了单个Agent对话循环的范畴。你需要持久化流程可能运行数小时或数天需要持久化状态。调度与重试定时触发、失败后自动重试。并行与分支同时执行多个独立任务或根据条件走不同分支。可视化与监控清晰地看到整个流程的进度和状态。5.2 实战用Prefect编排一个包含Agent步骤的Workflow我们将使用Prefect来编排一个简单的业务流程其中一步是调用我们之前构建的LangGraph Agent。# 文件prefect_agent_workflow.py from prefect import flow, task from typing import Dict, Any import asyncio # 假设我们有一个运行LangGraph Agent的异步函数来自之前的代码 async def run_agent_async(question: str) - str: “”“一个模拟的异步函数运行我们的LangGraph Agent。”“” # 这里简化实现实际应调用编译好的LangGraph app print(f”[Agent] Processing: {question}”) await asyncio.sleep(1) # 模拟处理时间 # 模拟一个简单的回答 if “weather” in question.lower(): return “The weather is simulated to be sunny.” elif “calculate” in question.lower(): return “The calculated result is 42.” else: return f”I have processed your query: ‘{question}’.” task async def extract_user_query(data_source: str) - str: “”“任务1从数据源提取用户查询。”“” print(f”[Task: Extract] From source: {data_source}”) # 模拟从数据库、API或文件读取 await asyncio.sleep(0.5) simulated_queries [“What‘s the weather like?”, “Calculate 15 * 3”, “Just say hello.”] import random return random.choice(simulated_queries) task async def call_agent_processing(query: str) - Dict[str, Any]: “”“任务2调用LangGraph Agent处理查询。”“” print(f”[Task: Agent] Query: {query}”) answer await run_agent_async(query) return {“original_query”: query, “agent_answer”: answer} task async def log_and_notify(result: Dict[str, Any]): “”“任务3记录结果并发送通知。”“” print(f”[Task: Log] Query: {result[‘original_query’]}”) print(f”[Task: Log] Answer: {result[‘agent_answer’]}”) # 模拟发送通知如邮件、Slack print(f”[Task: Notify] Notification sent for query: {result[‘original_query’][:30]}…”) await asyncio.sleep(0.3) flow(name“Agent-Enhanced Business Workflow”) async def agent_workflow(data_source: str “simulated_db”): “”“主工作流串联提取、Agent处理、日志通知三个步骤。”“” # 1. 提取查询 user_query await extract_user_query(data_source) # 2. 交给Agent处理 processing_result await call_agent_processing(user_query) # 3. 记录和通知 await log_and_notify(processing_result) return processing_result if __name__ “__main__”: # 运行工作流 final_result asyncio.run(agent_workflow(“test_source”)) print(“\n Workflow Execution Finished ) print(f”Final Result: {final_result}”)在这个例子中Prefect负责管理三个任务task的依赖关系和执行。call_agent_processing任务封装了我们的LangGraph Agent。这样Agent就成为了一个更大、更健壮的自动化业务流程中的一个可复用、可监控的组件。6. 渐进式披露Skills与Harness架构设计现在我们融合三大形态探讨如何设计一个优雅的Skill管理系统。核心思想是渐进式披露Progressive DisclosureAgent不需要一开始就知道所有Skills而是在需要时动态发现、学习和调用。6.1 什么是Skills和HarnessSkills如前所述是Agent可执行的高级能力单元。一个Skill可能对应一个简单的Tool也可能对应一个由多个步骤甚至一个子Workflow组成的复杂过程。例如“生成季度报告”Skill可能包含数据提取、分析、图表生成、文档排版等多个步骤。Harness可以理解为Agent的“治理框架”或“运行时环境”。它负责Skill注册与管理维护一个Skill仓库Registry。动态Skill发现从MCP Servers或其他来源发现新的Skills。上下文构建与路由根据用户请求从众多Skills中筛选出最相关的几个并以合适的描述Context提供给LLM。执行与编排调用LangGraph或Workflow引擎来执行选中的Skill。安全与监控权限控制、使用审计、性能监控。6.2 架构设计蓝图------------------- 发现 ---------------------- | MCP Servers | ----------- | Harness | | (Tool Providers) | (清单) | (Skill Registry | ------------------- | Runtime Manager) | --------------------- | 路由 编排 v --------------------- | Core Agent (LLM) | | with LangGraph | --------------------- | 执行 v --------------------- | Skills | | (Simple Tools / | | Complex Workflows) | ----------------------工作流程注册与发现Harness启动时连接配置好的MCP Servers获取所有可用的Tools列表并将其注册为基本的Skills。同时也可以从本地目录加载预定义的复杂Skills这些Skills本身可能是用LangGraph定义的小型工作流。接收请求用户向Harness发送请求。技能路由Harness根据请求内容从Skill Registry中快速检索出最相关的N个Skills例如通过向量相似度匹配Skill的描述。构建上下文Harness将这N个Skills的描述名称、功能、输入输出格式作为上下文与用户请求一起提交给核心AgentLLM。规划与执行核心Agent由LangGraph驱动根据上下文进行思考Reason决定调用哪个Skill并生成正确的调用参数。Harness接收指令执行对应的Skill。如果Skill是一个复杂工作流则交给Prefect等引擎执行。返回结果Skill执行结果返回给核心AgentAgent可能继续思考或直接给出最终答案由Harness返回给用户。6.3 简化版Harness代码示例# 文件simple_harness.py from typing import List, Dict, Any, Optional from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage import asyncio class Skill: “”“技能抽象类。”“” def __init__(self, name: str, description: str, executor): self.name name self.description description self.executor executor # 执行函数或可调用对象 async def execute(self, **kwargs) - Any: return await self.executor(**kwargs) if asyncio.iscoroutinefunction(self.executor) else self.executor(**kwargs) class SimpleHarness: def __init__(self, llm): self.llm llm self.skill_registry: Dict[str, Skill] {} # 可以集成一个简单的向量存储来检索相关技能此处简化 def register_skill(self, skill: Skill): self.skill_registry[skill.name] skill def _select_relevant_skills(self, query: str, top_k: int 3) - List[Skill]: “”“根据查询选择最相关的技能。此处使用简单的关键词匹配作为示例。”“” query_lower query.lower() scored_skills [] for skill in self.skill_registry.values(): score 0 # 简单的关键词评分逻辑 if any(word in query_lower for word in skill.name.lower().split(‘_’)): score 2 if any(word in query_lower for word in skill.description.lower().split()): score 1 if score 0: scored_skills.append((score, skill)) # 按分数排序 scored_skills.sort(keylambda x: x[0], reverseTrue) return [skill for _, skill in scored_skills[:top_k]] async def process_query(self, user_query: str) - str: # 1. 技能路由 relevant_skills self._select_relevant_skills(user_query) if not relevant_skills: return “I don‘t have any relevant skills to handle your request.” # 2. 构建上下文 skill_context “\n”.join([f”- {s.name}: {s.description}” for s in relevant_skills]) system_prompt f”””You are a helpful assistant with access to specific skills. Use the following skills if needed: {skill_context} If a user‘s request can be fulfilled by one of these skills, respond in the format: THOUGHT: I should use the [Skill Name] skill. ACTION: [Skill Name] ACTION_INPUT: {{“arg1”: “value1”, “arg2”: “value2”}} Otherwise, provide a helpful general response. “”” messages [SystemMessage(contentsystem_prompt), HumanMessage(contentuser_query)] # 3. 调用LLM进行规划 response await self.llm.ainvoke(messages) llm_output response.content # 4. 解析并执行 if “ACTION:” in llm_output: lines llm_output.split(‘\n’) action None action_input_str None for line in lines: if line.startswith(‘ACTION:’): action line.replace(‘ACTION:’, “”).strip() elif line.startswith(‘ACTION_INPUT:’): action_input_str line.replace(‘ACTION_INPUT:’, “”).strip() if action and action in self.skill_registry and action_input_str: import json try: args json.loads(action_input_str) skill self.skill_registry[action] result await skill.execute(**args) return f”I used the ‘{action}’ skill. Result: {result}” except Exception as e: return f”Failed to execute skill ‘{action}’: {str(e)}” # 5. 返回LLM的通用回答 return llm_output # 使用示例 async def main(): llm ChatOpenAI(model“gpt-4o-mini”, temperature0) harness SimpleHarness(llm) # 注册一些技能这些技能可以来自MCP Server或本地定义 async def skill_weather(city: str) - str: await asyncio.sleep(0.2) return f”The weather in {city} is simulated.” async def skill_calc(expression: str) - str: await asyncio.sleep(0.1) try: # 警告实际生产中不要用eval此处仅为演示 result eval(expression) return f”The result of {expression} is {result}.” except: return f”Could not calculate {expression}.” harness.register_skill(Skill(“get_weather”, “Get weather for a city.”, skill_weather)) harness.register_skill(Skill(“calculate”, “Evaluate a mathematical expression.”, skill_calc)) queries [“What‘s the weather in Tokyo?”, “What is 15 25?”, “Tell me a joke.”] for q in queries: print(f”\nUser: {q}”) answer await harness.process_query(q) print(f”Harness: {answer}”) if __name__ “__main__”: asyncio.run(main())这个简化的Harness演示了核心思想动态技能选择、上下文构建、LLM规划、技能执行。在一个生产系统中Harness会复杂得多集成向量检索、更安全的技能执行沙箱、权限校验、状态持久化等。7. 常见问题与排查思路在整合LangGraph、MCP、Workflow和自定义Harness时你可能会遇到以下典型问题问题现象常见原因解决思路LangGraph图陷入无限循环条件边conditional_edge逻辑错误或状态未正确更新导致始终无法满足结束条件。1. 检查decide_next_step函数的逻辑确保在完成任务后能返回”end”。2. 在State中添加明确的”is_finished”标志。3. 使用app.invoke(…, config{“recursion_limit”: 50})设置递归限制以防万一。MCP Server连接失败或工具不显示网络问题、Server未启动、认证失败、或Client/Server版本不兼容。1. 检查MCP Server进程是否正常运行。2. 验证连接URL或Stdio配置是否正确。3. 查看Server日志确认initialize和list_tools调用是否成功。4. 确保Client使用的MCP协议版本与Server兼容。LLM无法正确调用工具/技能Prompt中工具描述不清晰或LLM返回的格式无法被正确解析。1. 优化工具/技能的描述使其功能、输入参数清晰明确。2. 使用LangChain的StructuredTool或Pydantic来定义严格的输入模式并利用其内置的解析功能。3. 在Prompt中提供更精确的格式示例Few-shot。4. 考虑使用OpenAIFunctionsAgent或ReActAgent等LangChain内置的高阶Agent它们有更好的工具调用封装。Harness技能检索不准简单的关键词匹配效果差无法理解用户意图。1. 将技能描述和用户查询都编码成向量使用向量数据库如Chroma, FAISS进行语义检索。2. 可以引入一个轻量级的“路由LLM”来先对用户意图进行分类再选择技能。3. 为技能添加更丰富的元数据标签、类别以提高检索精度。复杂Workflow执行失败或状态丢失Workflow引擎如Prefect的部署模式、持久化存储或执行环境配置不当。1. 对于生产环境使用Prefect的Server Agent部署模式而非本地临时运行。2. 确保使用了正确的结果持久化器如本地文件、S3、数据库。3. 为每个Task和Flow添加详细的日志和错误处理。4. 利用Workflow引擎的重试、超时、回退策略。技能执行有安全风险技能特别是计算类eval、文件操作、系统命令直接执行不可信的用户输入。1.绝对禁止在技能中使用eval、exec或直接执行shell命令来处理原始用户输入。2. 对输入参数进行严格的验证和清洗。3. 在沙箱环境如Docker容器中运行高风险技能。4. 实现基于角色的权限控制RBAC不同用户/Agent只能访问特定技能。8. 最佳实践与工程建议构建一个企业级、可维护的Agent系统需要遵循以下工程原则分层与解耦工具层MCP专注于将外部能力包装成标准的、可发现的接口。保持工具实现的纯粹性。编排层LangGraph/Workflow专注于业务流程和状态管理。定义清晰的节点和边每个节点职责单一。治理层Harness专注于技能的动态管理、路由、安全和监控。它是系统的“大脑”和“调度中心”。表现层提供API、CLI或聊天界面给最终用户。Skill的设计原则单一职责一个Skill只做一件事并做好。描述清晰提供准确、详细的自然语言描述和结构化输入模式JSON Schema这是LLM能正确使用它的关键。幂等与安全尽可能设计幂等的Skill并内置输入验证和权限检查。版本化对Skill进行版本管理便于灰度发布和回滚。状态管理使用强类型State在LangGraph中使用Pydantic的BaseModel来定义State而不是简单的字典。这能获得更好的类型提示、自动验证和IDE支持。状态最小化只将必要的变量放入State避免State过于庞大影响性能和可读性。考虑持久化对于长时间运行的Agent需要将State持久化到数据库如Redis、PostgreSQL以便在应用重启后恢复。可观测性全面日志记录在Harness、LangGraph节点、Skill执行处添加结构化日志如使用structlog或loguru。链路追踪为每个用户会话或请求生成唯一的trace_id贯穿所有组件Harness - LangGraph - MCP Server便于问题排查。关键指标监控监控技能调用成功率、延迟、LLM的Token消耗、错误率等。测试策略单元测试对每个独立的Skill、LangGraph的节点函数进行单元测试。集成测试测试Harness与MCP Server的集成、LangGraph图的完整执行流程。端到端测试模拟真实用户场景测试从输入到输出的完整链条。LLM输出稳定性测试对于依赖LLM解析的环节如工具调用使用固定的Prompt和种子测试输出的确定性或使用断言来检查关键字段。生产环境部署配置外部化将LLM API Key、MCP Server地址、数据库连接等配置信息通过环境变量或配置中心如Apollo管理。容器化使用Docker将每个组件Harness、MCP Servers容器化便于部署和扩展。弹性与扩缩容Harness和MCP Server应设计为无状态或状态可外部化便于水平扩展。考虑使用消息队列如RabbitMQ, Kafka来解耦组件间的调用。通过将LangGraph、MCP和Workflow三大形态有机结合并辅以精心设计的Harness架构你构建的AI Agent将不再是脆弱的“脚本合集”而是一个真正强大、灵活、可扩展的智能系统。从明确的状态机LangGraph到标准化的工具集成MCP再到稳健的业务流程Workflow最后通过渐进式披露的Skill管理Harness将它们统一起来这套架构能够应对从简单问答到复杂自动化的大部分场景。