再聊OpenManus和MCP构建运维Agent实践及思考

推荐语

深入探讨OpenManus和MCP在运维Agent构建中的实践与挑战。

核心内容：
1. OpenManus项目在运维场景中的MVP测试与实施心得
2. 多智能体系统架构及各智能体模块的功能解析
3. OpenManus核心机制实现：规划、反思、记忆机制的详细介绍

杨芳贤

53AI创始人/腾讯云(TVP)最具价值专家

背景

之前写了一篇文章介绍过OpenManus项目，主要是通过本地搭建的开源模型服务来运行OpenManus项目中提供的实例。随着对OpenManus项目中工程设计深入思考和引入到我们运维场景的最小MVP测试, 发现离实际场景可用还有很多的工作需要做, 在结合业务运维场景实施落地一段时间后, 重新再梳理下，记录对OpenManus项目和MCP构建运维Agent实践及思考心得。

技术分析

OpenManus是一个多智能体系统，其架构中包含多个模块化的智能体，比如规划、执行、工具调用这些不同的代理各司其责，协同工作。采用分层架构设计，明确划分为规划、执行、工具调用等不同只能的智能体模块。分别有：

• 规划智能体(PlanningAgent)
  ：负责任务插接与流程编排；
• 执行智能体(SWEAgent)
  ：进行代码生成与工具调用；
• 工具智能体(ToolCallAgent)
  ：负责执行浏览网页、文件管理等具体操作。

谈到智能体，需要重新再温顾下其基本元素。大家普遍比较熟悉的AI Agent定义是“规划 + 记忆 + 工具调用 + 执行”的组合，其中LLM作为系统的核心控制器。

• 规划(Planning)
  ：将任务进行拆解、生成子任务以及自我反思；
• 记忆(Memory)
  ：进行短期记忆和长期记忆的管理，为问题提供上下文背景，可作用于复杂任务场景；
• 工具调用(Tool Usage)
  ：通过调用外部工具来增强Agent的能力，补充额外信息;
• 执行(Executor)
  ：通过执行工具获取工具返回结果，然后反馈给Agent进行分析或总结。

接下来，围绕OpenManus中规划机制、反思机制、记忆机制这几块核心内容实现大致介绍下

1. 规划机制

OpenManus规划过程：用户输入任务请求 -> 创建初始化计划 -> 分解任务为可执行步骤 -> 生成结构化计划。其中，PlanningTool类的作用就是创建和管理任务的计划，生成线性计划，并分配任务给相应的Agent。

当用户输入任务请求后，会调用_create_initial_plan方法，该方法中会调用LLM和PlanningTool生成初始计划，计划包含任务目标、步骤分解及状态追踪。PlanningTool将任务分解为可执行的线性步骤，并生成结构化计划（目标、步骤、状态）。 摘取调用LLM和PlanningTool生成初始计划的接口内容：

curl --location 'http://localhost:11434/api/chat' \
--header 'Content-Type: application/json' \
--data '{
    "model": "qwen2.5:14b",
    "messages": [
        {
            "role": "system",
            "content": "You are a planning assistant. Your task is to create a detailed plan with clear steps."
        },
        {
            "role": "user",
            "content": "Create a detailed plan to accomplish this task: 我想在上海市杨浦区新江湾城附近租一室户，要求租金不超过8000，2室一厅，90平以上，小区环境好，请帮我推荐下"
        }
    ],
    "options": {
        "temperature": 0.0,
        "stream": false
    },
    "tools": [
        {
            "type": "function",
            "function": {
                "name": "planning",
                "description": "\\nA planning tool that allows the agent to create and manage plans for solving complex tasks.\\nThe tool provides functionality for creating plans, updating plan steps, and tracking progress.\\n",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "command": {
                            "description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
                            "enum": [
                                "create",
                                "update",
                                "list",
                                "get",
                                "set_active",
                                "mark_step",
                                "delete"
                            ],
                            "type": "string"
                        },
                        "plan_id": {
                            "description": "Unique identifier for the plan. Required for create, update, set_active, and delete commands. Optional for get and mark_step (uses active plan if not specified).",
                            "type": "string"
                        },
                        "title": {
                            "description": "Title for the plan. Required for create command, optional for update command.",
                            "type": "string"
                        },
                        "steps": {
                            "description": "List of plan steps. Required for create command, optional for update command.",
                            "type": "array",
                            "items": {
                                "type": "string"
                            }
                        },
                        "step_index": {
                            "description": "Index of the step to update (0-based). Required for mark_step command.",
                            "type": "integer"
                        },
                        "step_status": {
                            "description": "Status to set for a step. Used with mark_step command.",
                            "enum": [
                                "not_started",
                                "in_progress",
                                "completed",
                                "blocked"
                            ],
                            "type": "string"
                        },
                        "step_notes": {
                            "description": "Additional notes for a step. Optional for mark_step command.",
                            "type": "string"
                        }
                    },
                    "required": [
                        "command"
                    ],
                    "additionalProperties": false
                }
            }
        }
    ],
    "tool_calls": "required"
}'

接口返回结果如下：

{
  "model": "qwen2.5:14b",
  "created_at": "2025-05-31T08:07:06.248964463Z",
  "message": {
    "role": "assistant",
    "content": "",
    "tool_calls": [
      {
        "function": {
          "name": "planning",
          "arguments": {
            "command": "create",
            "steps": [
              {
                "description": "确定搜索范围和条件，包括租金上限、户型要求等",
                "step_index": 0
              },
              {
                "description": "使用在线平台进行初步筛选，如链家网、58同城等",
                "step_index": 1
              },
              {
                "description": "联系房产中介获取更多信息，并预约看房时间",
                "step_index": 2
              },
              {
                "description": "实地考察潜在房源",
                "step_index": 3
              }
            ],
            "title": "租房计划：上海市杨浦区新江湾城附近"
          }
        }
      }
    ]
  },
  "done": false
}
{
  "model": "qwen2.5:14b",
  "created_at": "2025-05-31T08:07:06.359368765Z",
  "message": {
    "role": "assistant",
    "content": ""
  },
  "done_reason": "stop",
  "done": true,
  "total_duration": 11605808012,
  "load_duration": 27693026,
  "prompt_eval_count": 454,
  "prompt_eval_duration": 218000000,
  "eval_count": 208,
  "eval_duration": 11343000000
}

规划的扩展和挑战

• 规划模式：当前规划采用线性规划，但MetaGPT团队计划引入DAG（有向无环图）结构以支持更复杂的任务依赖关系。例如 Data Interpreter 任务可能需要多步骤的条件判断和动态调整；
• 动态规划的实现：系统通过update_plan_status方法根据工具执行结果动态调整计划。例如，若某步骤耗时过长，系统可能并行执行其他子任务以提高效率；
• 用户反馈式驱动优化：如果任务失败或不理解，可支持用户触发重新规划。

2. 反思机制

反思机制主要体现在采用ReAct模式与实时反馈相结合的方式.

• ReAct模式
  ：OpenManus在任务执行过程中会通过ReAct模式进行动态决策和迭代优化。关于ReAct框架核心机制实现具体体现在ReActAgent类中,定义了think和act两个静态方法。通过LLM分析当前状态，生成思考结果，并选择工具执行下一步操作。在此过程中系统会通过自我反思对之前决策和行动进行验证和修复，从而提升任务完成质量。比如在think 方法中，LLM 会基于历史消息和工具状态生成带工具选项的响应，这一过程本质上是对过去行为的反思性调整。

class ReActAgent(BaseAgent, ABC):
    name: str
    description: Optional[str] = None

    system_prompt: Optional[str] = None
    next_step_prompt: Optional[str] = None

    llm: Optional[LLM] = Field(default_factory=LLM)
    memory: Memory = Field(default_factory=Memory)
    state: AgentState = AgentState.IDLE

    max_steps: int = 10
    current_step: int = 0

    @abstractmethod
    async def think(self) -> bool:
        """Process current state and decide next action"""

    @abstractmethod
    async def act(self) -> str:
        """Execute decided actions"""

    async def step(self) -> str:
        """Execute a single step: think and act."""
        should_act = await self.think()
        if not should_act:
            return "Thinking complete - no action needed"
        return await self.act()

• 实时反馈和动态调整
  ：OpenManus 的实时反馈机制不仅向用户展示思考过程（如任务分解逻辑、工具调用步骤），还允许系统根据执行结果动态更新记忆和上下文。例如，当检测到重复步骤时，is_stuck 方法会触发异常终止，避免陷入无效循环。这种机制通过反思历史步骤来优化后续行动。

相较于传统端到端模型，OpenManus的反思机制使其在任务失败时能更精准地定位问题，但是依然还是有很多待优化改造。当前想到的几点如下：

• OpenManus项目当前是单智能体系统，会导致错误归因范围模糊，当任务链中出现工具级错误时（如浏览器自动化失败），系统仅能定位到当前执行节点，却无法追溯上游规划阶段的潜在决策失误。因此，需要引入多智能体架构的错误传播跟踪，构建任务依赖关系图谱；
• 动态决策机制不足：ReAct循环的状态评估阈值设置过于刚性，在复杂场景应对能力不足。

3. 记忆机制

记忆机制是通过多轮对话、上下文关联。比如动态上下文管理,每次交互会通过update_memory方法记录完整的对话历史(包括有用户输入、工具调用结果、LLM响应等)，形成可追溯的上下文链条。在ReActAgent类中初始化时会实例化两个对象，一个用来实现短期记忆，通过MemoryBuffer维护固定长度的上下文窗口，避免上下文过长导致性能下降或报错；另一个是长期记忆, 使用向量数据库存储关键信息，支持通过语义检索调用历史知识。

OpenManus通过Memory类实现对话历史的存储与管理，该组件以消息列表形式记录用户、系统和工具之间的交互记录，并通过update_memory方法动态更新。具体实现包括：

• 消息类型支持：支持用户输入、系统消息、助手回复及工具执行结果的分类型存储；
• 动态更新机制：每次交互后自动评估信息重要性，决定是否纳入记忆库，避免冗余数据堆积。

class Memory(BaseModel):
    messages: List[Message] = Field(default_factory=list)
    max_messages: int = Field(default=100)

    def add_message(self, message: Message) -> None:
        """Add a message to memory"""
        self.messages.append(message)
        # Optional: Implement message limit
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages :]

    def add_messages(self, messages: List[Message]) -> None:
        """Add multiple messages to memory"""
        self.messages.extend(messages)

    def clear(self) -> None:
        """Clear all messages"""
        self.messages.clear()

    def get_recent_messages(self, n: int) -> List[Message]:
        """Get n most recent messages"""
        return self.messages[-n:]

    def to_dict_list(self) -> List[dict]:
        """Convert messages to list of dicts"""
        return [msg.to_dict() for msg in self.messages]

关于长期记忆，目前还在规划中。社区提到引入长期记忆存储已验证知识或稳定状态，以提升重复任务效率。源码显示当前记忆只暂存于内存，但架构设计预留了通过数据库或文件持久化的扩展可能性。

实践案例

    受OpenManus项目启发，我们也在大数据实时引擎侧运维场景进行了试点探索。利用智能体规划，工具调用，执行以及反思过程来构建会自我更新和优化的运维工具。

1. 场景

• 针对Flink实时作业异常进行根因分析，异常类型主要包括：作业失败、数据堆积、断流、Checkpoint失败。

2. 目标

• 解决作业异常根因分析(作业失败、数据堆积、断流、Checkpoint失败)，以提升自动运维及诊断效果。

3. 执行流程

执行流程具体过程如下图:

3.1. 流程关键节点说明：

用户的请求往往信息量不足，为了对问题有更全面的理解，需要引入额外的知识库来补充问题的上下文以及解决思路，这种内容补充，我们将其定义为记忆检索。正如前面介绍OpenManus项目中记忆机制，其核心目标是一致的。关于记忆检索主要实现如下：

class Memory:
    """存储智能体交互历史的记忆系统"""
    def __init__(self):
        self.messages: List[Message] = []  # 消息队列（短期记忆）
        self.knowledge_base = Elasticsearch()  # 知识库（长期记忆）

    def add_message(self, message: Union[Message, dict]):
        """添加结构化消息到记忆流"""
        if isinstance(message, dict):
            message = Message(**message)
        self.messages.append(message)
        self._update_base(message)  # 触发知识库更新

    def _update_base(self, message):
        """基于消息内容构建知识节点"""
        entities = self.ner_extractor.extract(message.content)
        for entity in entities:
            self.knowledge_base.insert(
                label=entity['type'], 
                properties={'name': entity['text'], 'context': message.context}
            )

    def get_relevant_memories(self, query: str, top_k=5) -> List[Message]:
        """基于混合检索获取相关记忆"""
        query_embed = self.encoder.encode(query)
        return sorted(self.messages, 
                     key=lambda x: cosine_similarity(query_embed, x.embed))[:top_k]

利用llm进行规划的时候，需要理解用户问题，并根据提供的上下文内容进行规划，这样可避免llm“自我发挥”，以提升规划的合理性。该过程主要优化点在于Prompt，关于llm规划过程中Prompt内容可以借鉴OpenManus中的规划Prompt学习下。本案例中规划Prompt内容如下：

FLINK_DIAGNOSIS_PLAN_PROMPT = """
你是一个规划代理专家，任务是通过创建结构化计划来解决复杂问题。
你的要求是：
1. 如果问题中明确了思路，则按照该思路构建方案；如果上下文中提供了思路，则按照上下文思路构建方案；如果问题没有明确思路，则根据问题内容生成一个结构化计划。
你的工作是：
1. 分析请求以了解任务范围.
2. 用“计划”工具制定清晰、可行的计划.
3. 根据需要使用可用的工具执行步骤.
4. 历史记录中存在已知的错误消息，需要在下一个生成的结果中纠正该问题.
5. 跟踪进度并动态调整计划.
6. 输出json格式内容中的指令对象为数组类型，数组中的值为str类型。`thoughts`对象的类型是str.

按以下格式输出json列表：

```json
{{
  "plan":
  {{
    "thoughts": str # 详细记录你分析的过程及思路，按照“好的，我的思路是...”方式编写
    "instructions": [
    str # 记录步骤内容
    ]
  }}
}}

把任务分解成有逻辑的、连续的步骤。
"""

自从MCP在AI Agent领域爆火后，很多相关集成MCP的应用或框架都络绎不绝。MCP的优势在于整合了工具定义规范、自动注册、易用性调用等特点。因此，把工具都整合到已有的运维服务(Java)中，只需将运维服务升级为支持MCP框架就可以了。参考《基于AI的MCP协议解读及实践应用》文章，可了解具体如何构建支持MCP的Java服务。

通过由独立的运维服务整合了所需相关工具后，只需对外提供一个API接口即可支持调用了。同时服务内部llm执行逻辑是：根据问题调用工具，结果返回工具，执行工具获得结果，然后继续llm执行，直到llm返回结果没有工具则总结内容输出。内部封装了反馈过程，我们只需要在调优system_prompt就行。关于服务提供对外接口执行展示如下：

目前反思主要包括两个环节：工具调用不准确和规划的不合理。关于工具调用不准确，主要优化点在MCP内部逻辑过程，该过程主要通过对工具准确描述定义和Sytem Prompt来进行优化。而规划的不合理主要依赖于录入知识库中的内容、检索的内容准确性、对模型要求的规划Prompt以及规划流程的优化。其中重点规划流程的优化，需要考虑当前规划后的结果判断是否进行进一步规划这类在流程上的自优化迭代过程。总的来说，当前我们采用的规划流程处于线性迭代和约束性条件设定，整体上比较简单，且再规划迭代只支持两次。

从相关的资料可以获取到，在复杂场景中，AI Agent的规划流程需要突破传统线性决策框架的限制，构建具有动态感知、弹性调整和自主进化能力的智能系统。有以下迭代优化方向及技术实现路径：

• 闭环反馈驱动的动态再规划机制：核心逻辑是将再规划流程构建为"执行-反馈-修正"的闭环系统，实现实时环境响应。其中反馈又分为了：即时操作反馈、语义层反馈、用户性反馈。
• 多粒度任务分解与策略重组：目标是从固定规则的任务拆解，转向基于场景动态的弹性分解策略。其中，核心技术点之一需要有个动态复杂度评估模型来预测任务的难度，然后结合规则引擎（预定义行业模板）与神经网络（学习最优拆解路径）的混合决策机制。
• 多Agent协同架构：就是构建具有角色分工和知识共享的Agent集群系统，突破单Agent认知局限。其核心点就是根据任务阶段选择合适的Agent，同时需要构建记忆共享机制，方便Agent获取历史信息。
• 基于强化学习的自主策略优化：需要基于训练框架，来构建"环境-动作-奖励"的持续学习闭环。

总的来说规划反思过程是一个非常重要的优化环节，未来将会有更多实际可行性落地解决方案。需要我们持续关注和探索。

4. 结果展示

在构建好诊断Agent能力后，为了让用户直接体验咨询，可利用企微提供的接口来封装为一个BOT或应用服务号，然后用户可直接和BOT或应用服务号对话。问答过程中，BOT回答会包括两块：用户请求分析规划内容展示和执行过程及结果展示。目的是让用户知道agent的规划过程和具体工具调用结果展示以及最后结论。大致效果如下：

思考

1. 谈谈当前实践项目最大不足点

抛开实践案例中已实现当前Agent必备的相关核心模块功能外，有一项核心内容其实是缺失的，即：评估机制。目前实践案例中，对结果的正确与否，主要依赖人工评估。之前有篇文章写到吴恩达教授谈到构建Agent所需的关键技能时，他认为评估机制的缺失，是当前 Agent 构建过程中最大的“看不见的问题”。他主张快速搭建一个评估系统，哪怕是很初级的，这样可以承担很多重复性的判断任务。更重要的是基于真实数据、真实失败路径的“触觉型直觉”，才是系统构建中最宝贵的经验。因此，后续中，我们将在开发一个评估机制，以更好辅助优化Agent能力。

2. 如何让 LLM 拥有自主决策推理深度的智慧

之前看过一篇文章关于介绍了Thinkless 框架，让模型学会了“聪明偷懒”，更在效率与性能之间找到了平衡。因为推理模型往往不管复杂还是简单问题，都会展开 lengthy 的推理过程。这样会带来很大的计算资源浪费。那在Agent智能系统中，对llm要求是多样的，有些场景需要不需要推理，有些场景需要推理。比如路由过程，要求快和准。规划过程要求模型有自己的思考，不能一味根据提供的上下文内容来规划，这样灵活性和智能性将会降低。从应用开发者角度讲，核心点还是结合不同场景设定llm模型的角色和类型。让llm学会“偷懒”，同时也效率进一步提升。