part05运维Agent方案设计与源码分析

agent 红队
记录
发布日期:   2026-10-26

前置准备

我觉得我们设计一个蓝队agent的话，这个运维agent是少不了了。就是我们有报错，有告警的话，全部是人工看确实是很麻烦的一个步骤

这个就是一套重复的体力活

1.看告警信息，提取关键信息（哪个服务？什么错误？什么时间？）

2.去日志平台搜索日志

3.去监控平台看指标曲线

4.根据经验判断原因

5.决定处理方案（观察？重启？联系下游？）

区别于普通的程序，agent做的：告警来了，先查日志，发现超时错误，那就再去查下下游服务的状态，发现下游正常，那就看是不是自己服务的连接池是不是满了，一步步推理下去，最终定位到根本原因

为什么需要运维agent
1.新人培养问题
2.80%都是老面孔，每次都需要重新查一遍
3.各个系统是割裂的，排查得来回切换

运维agent的目标是什么？
让机器去处理重复的事，让人去做有价值的事

运维agent有哪些核心功能
1.实时告警相应与自动排查
2.工程师维护一个错误码/错误特征的知识库，达到智能匹配根本原因
3.告警处理，总结本次排查过程，更新知识库，下次遇到类似的问题，直接复用
在这个过程中不断进步升级

架构设计

plan-execute-replan
1.像老同事一样有条理地规划排查步骤
2.遇到此路不通时能够灵活调整方向
3.统一调度各个系统的工具

先规划，再执行，随时调整

和ReAct的区别就在于，react没有一个完整的计划，每一步都是临场决策
用plan-execute-replan的场景：任务比较复杂，步骤多，需要一步步地推进。比如运维告警排查，自动化报告生成，多系统联动的工作流。这类工作如果“边想边做”，很容易跑偏或者遗漏步骤

用ReAct场景，任务比较灵活，事先不确定需要几步，探索性强。比如回答一个开放性问题，在网上搜索信息等，这个任务如果要先做出计划，反而显得僵硬

plan-execute-replan优势：
1.结构化拆解，把复杂问题变简单
2.动态适应，遇到问题不死磕
3.职责分离，各司其职更加高效（规划，执行，评估三个职责分开，交给不同agent来做）
4.任务进度可以观测

关键代码：app/agent/aiops/ 目录下的 state.py、planner.py、executor.py、replanner.py，以及 app/services/aiops_service.py。

运维 Agent 的核心目标是 规划 → 执行 → 评估 → 调整。整体流程就是三个节点：

  1. Planner：拆解排查步骤，生成执行计划
  2. Executor：从计划中取出第一个步骤，调用工具执行
  3. Replanner：评估执行结果，决定继续、调整计划还是生成最终报告

三个节点通过 LangGraph StateGraph 串联，共享同一份 PlanExecuteState 状态对象在整个流程中传递。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
"""
通用 Plan-Execute-Replan 状态定义
基于 LangGraph 官方教程实现
"""

from typing import List, TypedDict, Annotated
import operator


class PlanExecuteState(TypedDict):
    """Plan-Execute-Replan 状态"""
    
    # 用户输入（任务描述）
    input: str
    
    # 执行计划（步骤列表）
    plan: List[str]
    
    # 已执行的步骤历史
    # 使用 operator.add 实现追加式更新（而非覆盖）
    past_steps: Annotated[List[tuple], operator.add]
    
    # 最终响应/报告
    response: str

past_steps 使用 Annotated[List[tuple], operator.add] 声明，LangGraph 会将每次节点返回的 past_steps 自动追加到列表中，而不是覆盖，无需手动维护历史。

Planner节点

Planner 负责制定执行计划，输出一个结构化的步骤列表。核心流程：

  1. 先调用 retrieve_knowledge 查询知识库，寻找历史经验文档
  2. 获取所有可用工具（本地工具 + MCP 工具），格式化为文字描述
  3. 将工具列表和经验文档注入 prompt，调用 LLM 生成结构化计划
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
"""
Planner 节点：制定执行计划
基于 LangGraph 官方教程实现
"""

from textwrap import dedent
from typing import Dict, Any, List
from langchain_core.prompts import ChatPromptTemplate
from langchain_qwq import ChatQwen
from pydantic import BaseModel, Field
from loguru import logger

from app.config import config
from app.tools import DEFAULT_LOCAL_AGENT_TOOLS, retrieve_knowledge
from app.agent.mcp_client import get_mcp_client_with_retry
from .state import PlanExecuteState
from .utils import format_tools_description


class Plan(BaseModel):
    """计划的输出格式"""
    steps: List[str] = Field(
        description="完成任务所需的不同步骤。这些步骤应该按顺序执行，每一步都建立在前一步的基础上。"
    )


# Planner 提示词
planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            dedent("""
                作为一个专家级别的规划者，你需要将复杂的任务分解为可执行的步骤。

                可用工具列表（用于制定计划时参考）：

                {tools_description}

                注意：你的职责是制定计划，实际的工具调用由 Executor 负责执行。

                {experience_context}

                对于给定的任务，请创建一个简单的、逐步的计划来完成它。计划应该：
                - 将任务分解为逻辑上独立的步骤
                - 每个步骤应该明确使用哪些工具(如果需要工具的话)来获取信息, 最好能同时提供工具执行所需要的参数
                - 步骤之间应该有清晰的依赖关系
                - 步骤描述要具体、可操作
                - **如果有相关经验文档，请参考其中的方法和步骤制定计划**

                示例输入："分析当前系统的性能问题"
                示例输出（假设有对应工具）：
                步骤1: 使用 get_metrics 工具收集系统的 CPU 和内存使用情况
                步骤2: 使用 query_logs 工具检查最近的错误日志
                步骤3: 使用 query_database 工具分析慢查询日志
                步骤4: 综合以上信息生成性能分析报告
            """).strip(),
        ),
        ("placeholder", "{messages}"),
    ]
)


async def planner(state: PlanExecuteState) -> Dict[str, Any]:
    """
    规划节点：根据用户输入生成执行计划

    流程：
    1. 先查询内部文档，获取相关经验和最佳实践
    2. 基于经验文档和可用工具制定执行计划
    """
    logger.info("=== Planner：制定执行计划 ===")

    input_text = state.get("input", "")
    logger.info(f"用户输入: {input_text}")

    try:
        # 步骤1: 查询内部文档获取相关经验
        logger.info("查询内部文档，寻找相关经验...")
        experience_docs = ""
        try:
            # retrieve_knowledge 使用 response_format="content_and_artifact"
            # ainvoke() 只返回 content（字符串），不是元组
            context_str = await retrieve_knowledge.ainvoke({"query": input_text})
            if context_str and context_str.strip():
                experience_docs = context_str
                logger.info(f"找到相关经验文档，长度: {len(experience_docs)}")
            else:
                logger.info("未找到相关经验文档")
        except Exception as e:
            logger.warning(f"查询内部文档失败: {e}")

        # 步骤2: 获取可用工具列表
        # 获取本地工具
        local_tools = list(DEFAULT_LOCAL_AGENT_TOOLS)

        # 获取 MCP 工具
        mcp_client = await get_mcp_client_with_retry()
        mcp_tools = await mcp_client.get_tools()

        # 合并所有工具
        all_tools = local_tools + mcp_tools
        logger.info(f"可用工具数量: 本地 {len(local_tools)} + MCP {len(mcp_tools)}")

        # 格式化工具描述
        tools_description = format_tools_description(all_tools)

        # 步骤3: 格式化经验文档上下文
        if experience_docs:
            experience_context = dedent(f"""
                ## 相关经验文档

                以下是从知识库中检索到的相关经验和最佳实践，请参考这些经验制定执行计划：

                {experience_docs}

                ---
            """).strip()
        else:
            experience_context = ""

        # 步骤4: 创建 LLM 并生成计划
        llm = ChatQwen(
            model=config.rag_model,
            api_key=config.dashscope_api_key,
            temperature=0
        )

        planner_chain = planner_prompt | llm.with_structured_output(Plan)

        # 调用 LLM 生成计划
        plan_result = await planner_chain.ainvoke({
            "messages": [("user", input_text)],
            "tools_description": tools_description,
            "experience_context": experience_context
        })

        # 提取步骤列表
        if isinstance(plan_result, Plan):
            plan_steps = plan_result.steps
        else:
            # 如果返回的是字典，提取 steps 字段
            plan_steps = plan_result.get("steps", [])  # type: ignore

        logger.info(f"计划已生成，共 {len(plan_steps)} 个步骤")
        for i, step in enumerate(plan_steps, 1):
            logger.info(f"  步骤{i}: {step}")

        return {"plan": plan_steps}

    except Exception as e:
        logger.error(f"生成计划失败: {e}", exc_info=True)
        # 返回一个默认计划
        return {
            "plan": [
                "收集相关信息",
                "分析数据",
                "生成报告"
            ]
        }

计划的输出格式用Pydantic Plan模型约束，通过llm.with_structured_output(Plan)保证llm的输出可以直接解析为步骤列表

planner Prompt

planner的系统提示词要求模型将任务分解为逻辑独立的步骤，每步指明使用哪个工具以及所需参数。如果查到了经验文档，也会作为参考注入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
planner_prompt = ChatPromptTemplate.from_messages([
    ("system", """
作为一个专家级别的规划者，你需要将复杂的任务分解为可执行的步骤。

可用工具列表（用于制定计划时参考）：
{tools_description}

注意：你的职责是制定计划，实际的工具调用由 Executor 负责执行。

{experience_context}

对于给定的任务，请创建一个简单的、逐步的计划：
- 将任务分解为逻辑上独立的步骤
- 每个步骤明确使用哪些工具（如果需要），最好能同时提供工具所需参数
- 步骤之间应有清晰的依赖关系
- 如果有相关经验文档，请参考其中的方法和步骤制定计划
"""),
    ("placeholder", "{messages}"),
])

Executor节点

Executor每次只执行计划中的第一个步骤，使用LangGraph的ToolNode自动处理工具调用，执行完后将该步骤从plan中移除，并将结果追加到Past_steps:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
"""
Executor 节点：执行单个步骤
基于 LangGraph 官方教程实现
"""

from typing import Dict, Any
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_qwq import ChatQwen
from langgraph.prebuilt import ToolNode
from loguru import logger

from app.config import config
from app.tools import DEFAULT_LOCAL_AGENT_TOOLS
from app.agent.mcp_client import get_mcp_client_with_retry
from .state import PlanExecuteState


async def executor(state: PlanExecuteState) -> Dict[str, Any]:
    """
    执行节点：执行计划中的下一个步骤
    
    使用 LangGraph 的 ToolNode 自动处理工具调用
    """
    logger.info("=== Executor：执行步骤 ===")

    plan = state.get("plan", [])

    # 如果计划为空，不执行
    if not plan:
        logger.info("计划为空，跳过执行")
        return {}

    # 取出第一个步骤
    task = plan[0]
    logger.info(f"当前任务: {task}")

    try:
        # 获取本地工具
        local_tools = list(DEFAULT_LOCAL_AGENT_TOOLS)

        # 获取 MCP 工具
        mcp_client = await get_mcp_client_with_retry()
        mcp_tools = await mcp_client.get_tools()
        logger.info(f"可用工具数量: 本地 {len(local_tools)} + MCP {len(mcp_tools)}")

        # 合并所有工具
        all_tools = local_tools + mcp_tools

        # 创建 LLM（绑定工具）
        llm = ChatQwen(
            model=config.rag_model,
            api_key=config.dashscope_api_key,
            temperature=0
        )
        llm_with_tools = llm.bind_tools(all_tools)

        # 创建工具节点（自动执行工具调用）
        tool_node = ToolNode(all_tools)

        # 构建消息（只包含当前步骤，避免原始任务干扰）
        messages = [
            SystemMessage(content="""你是一个能力强大的助手，负责执行具体的任务步骤。

你可以使用各种工具来完成任务。对于每个步骤：
1. 理解步骤的目标
2. 选择合适的工具，如果已经指定了工具，则使用指定的工具
3. 调用工具获取信息
4. 返回执行结果

注意：
- 如果工具调用失败，请说明失败原因
- 不要编造数据，只返回实际获取的信息
- 执行结果要清晰、准确
- 专注于当前步骤，不要考虑其他任务"""),
            HumanMessage(content=f"请执行以下任务: {task}")
        ]

        # 第一步：LLM 决定是否调用工具
        llm_response = await llm_with_tools.ainvoke(messages)
        logger.info(f"LLM 响应类型: {type(llm_response)}")

        # 第二步：如果有工具调用，执行工具
        if hasattr(llm_response, "tool_calls") and llm_response.tool_calls:
            logger.info(f"检测到 {len(llm_response.tool_calls)} 个工具调用")
            
            # 使用 ToolNode 自动执行工具
            messages.append(llm_response)
            tool_messages = await tool_node.ainvoke({"messages": messages})
            
            # 第三步：将工具结果返回给 LLM 生成最终答案
            messages.extend(tool_messages["messages"])
            final_response = await llm_with_tools.ainvoke(messages)
            result = final_response.content if hasattr(final_response, 'content') else str(final_response)
        else:
            # 没有工具调用，直接使用 LLM 的输出
            logger.info("LLM 未调用工具，直接返回结果")
            result = llm_response.content if hasattr(llm_response, 'content') else str(llm_response)

        logger.info(f"步骤执行完成，结果长度: {len(result)}")

        # 返回更新：移除已执行的步骤，添加执行历史
        return {
            "plan": plan[1:],  # 移除第一个步骤
            "past_steps": [(task, result)],  # 使用 operator.add 追加
        }

    except Exception as e:
        logger.error(f"执行步骤失败: {e}", exc_info=True)
        return {
            "plan": plan[1:],
            "past_steps": [(task, f"执行失败: {str(e)}")],
        }

Replanner节点

Replanner 根据原始任务、已执行步骤和剩余计划做出三选一的决策：

决策 含义 触发条件
respond 信息充足，立即生成最终报告 最高优先级，已执行 ≥ 3 步且有关键信息
continue 当前计划合理，继续执行 剩余步骤确实必要
replan 调整计划，替换剩余步骤 最低优先级，计划有重大偏差时才使用

决策同样用 Pydantic 模型约束输出：

1
2
3
4
class Act(BaseModel):
    action: str = Field(description="下一步行动: 'continue' | 'replan' | 'respond'")
    new_steps: List[str] = Field(default_factory=list, description="replan 时的新步骤列表")

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
"""
Replanner 节点：重新规划或生成最终响应
基于 LangGraph 官方教程实现
"""

from textwrap import dedent
from typing import Dict, Any, List
from langchain_core.prompts import ChatPromptTemplate
from langchain_qwq import ChatQwen
from pydantic import BaseModel, Field
from loguru import logger

from app.config import config
from app.tools import DEFAULT_LOCAL_AGENT_TOOLS
from app.agent.mcp_client import get_mcp_client_with_retry
from .state import PlanExecuteState
from .utils import format_tools_description


class Response(BaseModel):
    """最终响应的格式"""
    response: str = Field(description="对用户的最终响应")


class Act(BaseModel):
    """重新规划的输出格式"""
    action: str = Field(
        description="""下一步的行动，必须是以下三种之一：
        - 'continue': 当前计划合理，继续执行下一个步骤
        - 'replan': 当前计划需要调整，提供新的步骤列表
        - 'respond': 计划已完成且信息充足，生成最终响应"""
    )
    # action 为 'replan' 时，新的步骤列表（会替换当前剩余计划）
    new_steps: List[str] = Field(
        default_factory=list,
        description="新的步骤列表（如果 action 是 'replan'，这些步骤会替换剩余计划）"
    )


# Replanner 提示词
replanner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            dedent("""
                作为一个重新规划专家，你需要根据已执行的步骤决定下一步行动。

                可用工具列表（用于制定计划时参考）：

                {tools_description}

                注意：你的职责是制定或调整计划，实际的工具调用由 Executor 负责执行。

                你有三个选择（按优先级排序）：

                **1. 'respond' - 信息充足，立即生成最终响应** 【最高优先级】
                   - 使用场景：当前信息已经足够回答用户问题
                   - 决策标准：
                     * 已执行步骤 >= 3 且获取了关键信息
                     * 或者已执行步骤 >= 5（无论结果如何）
                     * 或者当前信息完全满足任务需求
                   - ⚠️ 不要等到"完美"才响应，"足够好"就应该立即 respond

                **2. 'continue' - 当前计划合理，继续执行** 【次优先级】
                   - 使用场景：剩余计划合理且必要
                   - 决策标准：剩余步骤确实能提供关键信息
                   - ⚠️ 如果剩余步骤不是"必需"的，应选择 respond

                **3. 'replan' - 当前计划有严重问题** 【最低优先级，谨慎使用】
                   - 使用场景：原计划明显错误或遗漏关键步骤
                   - ⚠️ **严格限制**：
                     * 新步骤数量必须 <= 当前剩余步骤数
                     * 优先简化计划，不要添加不必要的步骤
                     * 总步骤数已执行 >= 5 次时，禁止 replan，只能 respond

                评估标准：
                - 当前信息是否已经足够解决用户问题？【最关键】
                - 已执行步骤是否成功获取了核心信息？
                - 剩余步骤是否真的"必需"？
                - 已执行步骤数是否过多（>= 5）？如果是，立即 respond

                **决策优先级口诀：** 
                "优先结束 > 保持不变 > 调整计划"
                "信息足够就响应，不要追求完美"
            """).strip(),
        ),
        ("placeholder", "{messages}"),
    ]
)

# 最终响应生成提示词
response_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            dedent("""
                根据原始任务和已执行步骤的结果，生成一个全面的最终响应。

                响应要求：
                - 清晰、结构化
                - 基于实际数据，不要编造
                - 如果某些步骤失败，要诚实说明
                - 使用 Markdown 格式
            """).strip(),
        ),
        ("placeholder", "{messages}"),
    ]
)


async def replanner(state: PlanExecuteState) -> Dict[str, Any]:
    """
    重新规划节点：决定是继续、调整计划还是生成最终响应

    三种决策：
    1. continue - 继续执行当前计划
    2. replan - 调整计划（替换剩余步骤）
    3. respond - 生成最终响应
    """
    logger.info("=== Replanner：重新规划 ===")

    input_text = state.get("input", "")
    plan = state.get("plan", [])
    past_steps = state.get("past_steps", [])

    logger.info(f"剩余计划步骤: {len(plan)}")
    logger.info(f"已执行步骤: {len(past_steps)}")

    # ⚠️ 强制限制：如果已执行步骤过多，直接生成响应
    MAX_STEPS = 8
    if len(past_steps) >= MAX_STEPS:
        logger.warning(f"已执行 {len(past_steps)} 个步骤，超过最大限制 {MAX_STEPS}，强制生成最终响应")
        llm = ChatQwen(
            model=config.rag_model,
            api_key=config.dashscope_api_key,
            temperature=0
        )
        return await _generate_response(state, llm)

    # 获取可用工具列表
    try:
        # 获取本地工具
        local_tools = list(DEFAULT_LOCAL_AGENT_TOOLS)

        # 获取 MCP 工具
        mcp_client = await get_mcp_client_with_retry()
        mcp_tools = await mcp_client.get_tools()

        # 合并所有工具
        all_tools = local_tools + mcp_tools
        logger.info(f"可用工具数量: 本地 {len(local_tools)} + MCP {len(mcp_tools)}")

        # 格式化工具描述
        tools_description = format_tools_description(all_tools)
    except Exception as e:
        logger.warning(f"获取工具列表失败: {e}")
        tools_description = "无法获取工具列表"

    # 创建 LLM
    llm = ChatQwen(
        model=config.rag_model,
        api_key=config.dashscope_api_key,
        temperature=0
    )

    # 格式化已执行的步骤
    steps_summary = "\n".join([
        f"步骤: {step}\n结果: {result[:300]}..."
        for step, result in past_steps
    ])

    # 如果还有剩余计划，进行决策
    if plan:
        logger.info("还有剩余计划，评估下一步行动")

        replanner_chain = replanner_prompt | llm.with_structured_output(Act)

        try:
            messages = [
                ("user", f"原始任务: {input_text}"),
                ("user", f"已执行的步骤:\n{steps_summary}"),
                ("user", f"剩余计划: {', '.join(plan)}"),
                ("user", f"⚠️ 重要提示：已执行 {len(past_steps)} 个步骤，请优先考虑是否信息已足够生成响应（respond）")
            ]

            act = await replanner_chain.ainvoke({
                "messages": messages,
                "tools_description": tools_description
            })

            # 处理返回结果
            if isinstance(act, Act):
                action = act.action
                new_steps = act.new_steps
            else:
                # 如果返回的是字典
                action = act.get("action", "continue")  # type: ignore
                new_steps = act.get("new_steps", [])  # type: ignore

            logger.info(f"Replanner 决策: {action}")

            if action == "respond":
                logger.info("决定生成最终响应")
                return await _generate_response(state, llm)

            elif action == "replan":
                # ⚠️ 强制限制：新步骤数不能超过当前剩余步骤数
                if len(new_steps) > len(plan):
                    logger.warning(
                        f"新步骤数 {len(new_steps)} > 剩余步骤数 {len(plan)}，"
                        f"强制截断为 {len(plan)} 个步骤"
                    )
                    new_steps = new_steps[:len(plan)]
                
                # ⚠️ 二次检查：如果已执行步骤 >= 5，禁止 replan
                if len(past_steps) >= 5:
                    logger.warning(f"已执行 {len(past_steps)} 个步骤，禁止重新规划，强制生成响应")
                    return await _generate_response(state, llm)
                
                logger.info(f"决定调整计划，新步骤数量: {len(new_steps)}")
                if new_steps:
                    # 替换剩余计划
                    return {"plan": new_steps}
                else:
                    logger.warning("replan 但未提供新步骤，继续执行原计划")
                    return {}

            else:  # action == "continue"
                logger.info("决定继续执行当前计划")
                return {}  # 不修改状态，继续执行

        except Exception as e:
            logger.error(f"重新规划失败: {e}, 继续执行剩余计划")
            return {}

    else:
        # 没有剩余计划，生成最终响应
        logger.info("计划已执行完毕，生成最终响应")
        return await _generate_response(state, llm)


async def _generate_response(state: PlanExecuteState, llm: ChatQwen) -> Dict[str, Any]:
    """生成最终响应"""
    logger.info("生成最终响应...")

    input_text = state.get("input", "")
    past_steps = state.get("past_steps", [])

    # 格式化执行历史
    execution_history = "\n\n".join([
        f"### 步骤: {step}\n**结果:**\n{result}"
        for step, result in past_steps
    ])

    response_gen = response_prompt | llm.with_structured_output(Response)

    try:
        messages = [
            ("user", f"原始任务: {input_text}"),
            ("user", f"执行历史:\n{execution_history}"),
            ("user", "请基于以上信息生成全面的最终响应")
        ]

        response_obj = await response_gen.ainvoke({"messages": messages})

        # 处理返回结果
        if isinstance(response_obj, Response):
            final_response = response_obj.response
        else:
            # 如果返回的是字典
            final_response = response_obj.get("response", "")  # type: ignore

        logger.info(f"最终响应生成完成，长度: {len(final_response)}")

        return {"response": final_response}

    except Exception as e:
        logger.error(f"生成响应失败: {e}")
        # 生成简单的后备响应
        fallback_response = f"""# 任务执行结果

## 原始任务
{input_text}

## 执行的步骤
{_format_simple_steps(past_steps)}

## 说明
由于系统异常，无法生成完整响应。以上是已收集的信息。
"""
        return {"response": fallback_response}


def _format_simple_steps(past_steps: list) -> str:
    """格式化步骤列表（简单版）"""
    if not past_steps:
        return "无"

    formatted = []
    for i, (step, result) in enumerate(past_steps, 1):
        result_preview = result[:200] + "..." if len(result) > 200 else result
        formatted.append(f"{i}. **{step}**\n   {result_preview}\n")

    return "\n".join(formatted)

构建 LangGraph 工作流

三个节点通过 StateGraph 连接，replanner 之后根据状态中是否存在 response 进行条件路由：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def _build_graph(self):
    workflow = StateGraph(PlanExecuteState)

    # 添加三个节点
    workflow.add_node("planner", planner)
    workflow.add_node("executor", executor)
    workflow.add_node("replanner", replanner)

    # 固定边: planner -> executor -> replanner
    workflow.set_entry_point("planner")
    workflow.add_edge("planner", "executor")
    workflow.add_edge("executor", "replanner")

    # 条件边: replanner 根据状态决定结束还是继续执行
    def should_continue(state: PlanExecuteState) -> str:
        if state.get("response"):
            return END          # 已生成最终响应，结束流程
        if state.get("plan"):
            return "executor"   # 还有步骤，继续执行
        return END

    workflow.add_conditional_edges("replanner", should_continue, {
        "executor": "executor",
        END: END
    })

    return workflow.compile(checkpointer=MemorySaver())

具体工作流结构

1
2
3
4
5
6
7
8
9
[用户任务] → planner → executor → replanner
                ↑                    |
                |      continue/replan
                └────────────────────┘
                                     |
                            respond / 空计划
                                     ↓
                                   [END]

执行工作流（流式输出）

AIOpsService.execute 使用 graph.astream(stream_mode="updates") 流式执行，每个节点完成后立即产生事件，可以实时推送给前端展示执行进度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
async def execute(self, user_input: str, session_id: str) -> AsyncGenerator:
    initial_state: PlanExecuteState = {
        "input": user_input,
        "plan": [],
        "past_steps": [],
        "response": ""
    }

    async for event in self.graph.astream(
        input=initial_state,
        config={"configurable": {"thread_id": session_id}},
        stream_mode="updates"  # 每个节点更新后立即推送
    ):
        for node_name, node_output in event.items():
            if node_name == "planner":
                yield self._format_planner_event(node_output)      # type: plan
            elif node_name == "executor":
                yield self._format_executor_event(node_output)    # type: step_complete
            elif node_name == "replanner":
                yield self._format_replanner_event(node_output)   # type: report / status

    # 所有节点完成，推送最终响应
    final_state = self.graph.get_state({"configurable": {"thread_id": session_id}})
    final_response = final_state.values.get("response", "") if final_state else ""

    yield {
        "type": "complete",
        "stage": "complete",
        "message": "任务执行完成",
        "response": final_response
    }

流式事件对照表

type stage 含义
plan plan_created Planner 生成了执行计划
step_complete step_executed Executor 执行完一个步骤
report final_report Replanner 生成了最终报告
status 各节点名 节点运行中的状态通知
complete complete 整个工作流结束
error error 执行出错

对应前面代码里的 yield：

  • planner → type: plan, stage: plan_created
  • executor → type: step_complete, stage: step_executed
  • replanner 分两种：report(final_report) / status
  • 循环结束后推送 complete
  • 异常捕获分支抛出 error

api接口与Agent的整合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
"""
AIOps 智能运维接口
"""

import json
from fastapi import APIRouter
from sse_starlette.sse import EventSourceResponse
from loguru import logger

from app.models.aiops import AIOpsRequest
from app.services.aiops_service import aiops_service

router = APIRouter()


@router.post("/aiops")
async def diagnose_stream(request: AIOpsRequest):
    """
    AIOps 故障诊断接口（流式 SSE）

    **功能说明：**
    - 自动获取当前系统的活动告警
    - 使用 Plan-Execute-Replan 模式进行智能诊断
    - 流式返回诊断过程和结果

    **SSE 事件类型：**

    1. `status` - 状态更新
       ```json
       {
         "type": "status",
         "stage": "fetching_alerts",
         "message": "正在获取系统告警信息..."
       }
2. `plan` - 诊断计划制定完成
   
1
2
3
4
5
6
7
{
  "type": "plan",
  "stage": "plan_created",
  "message": "诊断计划已制定，共 6 个步骤",
  "target_alert": {...},
  "plan": ["步骤1: ...", "步骤2: ..."]
}


3. `step_complete` - 步骤执行完成
   
1
2
3
4
5
6
7
8
{
  "type": "step_complete",
  "stage": "step_executed",
  "message": "步骤执行完成 (2/6)",
  "current_step": "查询系统日志",
  "result_preview": "...",
  "remaining_steps": 4
}


4. `report` - 最终诊断报告
   
1
2
3
4
5
6
7
{
  "type": "report",
  "stage": "final_report",
  "message": "最终诊断报告已生成",
  "report": "# 故障诊断报告\\n...",
  "evidence": {...}
}


5. `complete` - 诊断完成
   
1
2
3
4
5
6
{
  "type": "complete",
  "stage": "diagnosis_complete",
  "message": "诊断流程完成",
  "diagnosis": {...}
}


6. `error` - 错误信息
   
1
2
3
4
5
{
  "type": "error",
  "stage": "error",
  "message": "诊断过程发生错误: ..."
}


**使用示例：**

1
2
3
4
curl -X POST "http://localhost:9900/api/aiops" \\
  -H "Content-Type: application/json" \\
  -d '{"session_id": "session-123"}' \\
  --no-buffer


**前端使用示例：**

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const eventSource = new EventSource('/api/aiops');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);

  if (data.type === 'plan') {
    console.log('诊断计划:', data.plan);
  } else if (data.type === 'step_complete') {
    console.log('步骤完成:', data.current_step);
  } else if (data.type === 'report') {
    console.log('最终报告:', data.report);
  } else if (data.type === 'complete') {
    console.log('诊断完成');
    eventSource.close();
  }
};


Args:
    request: AIOps 诊断请求

Returns:
    SSE 事件流
"""
session_id = request.session_id or "default"
logger.info(f"[会话 {session_id}] 收到 AIOps 诊断请求（流式）")

async def event_generator():
    try:
        async for event in aiops_service.diagnose(session_id=session_id):
            # 发送事件
            yield {
                "event": "message",
                "data": json.dumps(event, ensure_ascii=False)
            }

            # 如果是完成或错误事件，结束流
            if event.get("type") in ["complete", "error"]:
                break

        logger.info(f"[会话 {session_id}] AIOps 诊断流式响应完成")

    except Exception as e:
        logger.error(f"[会话 {session_id}] AIOps 诊断流式响应异常: {e}", exc_info=True)
        yield {
            "event": "message",
            "data": json.dumps({
                "type": "error",
                "stage": "exception",
                "message": f"诊断异常: {str(e)}"
            }, ensure_ascii=False)
        }

return EventSourceResponse(event_generator())

1
2
3

以及

“””
通用 Plan-Execute-Replan 服务
基于 LangGraph 官方教程实现
“””

from typing import AsyncGenerator, Dict, Any
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from loguru import logger

from app.agent.aiops import PlanExecuteState, planner, executor, replanner

节点名称常量

NODE_PLANNER = “planner”
NODE_EXECUTOR = “executor”
NODE_REPLANNER = “replanner”

class AIOpsService:
“””通用 Plan-Execute-Replan 服务”””

def __init__(self):
    """初始化服务"""
    self.checkpointer = MemorySaver()
    self.graph = self._build_graph()
    logger.info("Plan-Execute-Replan Service 初始化完成")

def _build_graph(self):
    """构建 Plan-Execute-Replan 工作流"""
    logger.info("构建工作流图...")

    # 创建状态图
    workflow = StateGraph(PlanExecuteState)

    # 添加节点
    workflow.add_node(NODE_PLANNER, planner)      # 制定计划
    workflow.add_node(NODE_EXECUTOR, executor)  # 执行步骤
    workflow.add_node(NODE_REPLANNER, replanner)  # 重新规划

    # 设置入口点
    workflow.set_entry_point(NODE_PLANNER)

    # 定义边
    workflow.add_edge(NODE_PLANNER, NODE_EXECUTOR)     # planner -> executor
    workflow.add_edge(NODE_EXECUTOR, NODE_REPLANNER)   # executor -> replanner

    # replanner 的条件边
    def should_continue(state: PlanExecuteState) -> str:
        """判断是否继续执行"""
        # 如果已经生成了最终响应，结束
        if state.get("response"):
            logger.info("已生成最终响应，结束流程")
            return END

        # 如果还有计划步骤，继续执行
        plan = state.get("plan", [])
        if plan:
            logger.info(f"继续执行，剩余 {len(plan)} 个步骤")
            return NODE_EXECUTOR

        # 计划为空但没有响应，返回 replanner 生成响应
        logger.info("计划执行完毕，生成最终响应")
        return END

    workflow.add_conditional_edges(
        NODE_REPLANNER,
        should_continue,
        {
            NODE_EXECUTOR: NODE_EXECUTOR,
            END: END
        }
    )

    # 编译工作流
    compiled_graph = workflow.compile(checkpointer=self.checkpointer)

    logger.info("工作流图构建完成")
    return compiled_graph

async def execute(
    self,
    user_input: str,
    session_id: str = "default"
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    执行 Plan-Execute-Replan 流程

    Args:
        user_input: 用户的任务描述
        session_id: 会话ID

    Yields:
        Dict[str, Any]: 流式事件
    """
    logger.info(f"[会话 {session_id}] 开始执行任务: {user_input}")

    try:
        # 初始化状态
        initial_state: PlanExecuteState = {
            "input": user_input,
            "plan": [],
            "past_steps": [],
            "response": ""
        }

        # 流式执行工作流
        config_dict = {
            "configurable": {
                "thread_id": session_id
            }
        }

        async for event in self.graph.astream(
            input=initial_state,
            config=config_dict,
            stream_mode="updates"
        ):
            # 解析事件
            for node_name, node_output in event.items():
                logger.info(f"节点 '{node_name}' 输出事件")

                # 根据节点类型生成不同的事件
                if node_name == NODE_PLANNER:
                    yield self._format_planner_event(node_output)

                elif node_name == NODE_EXECUTOR:
                    yield self._format_executor_event(node_output)

                elif node_name == NODE_REPLANNER:
                    yield self._format_replanner_event(node_output)

        # 获取最终状态
        final_state = self.graph.get_state(config_dict)
        final_response = ""

        # 安全地获取响应（处理 values 可能为 None 的情况）
        if final_state and final_state.values:
            final_response = final_state.values.get("response", "")

        # 发送完成事件
        yield {
            "type": "complete",
            "stage": "complete",
            "message": "任务执行完成",
            "response": final_response
        }

        logger.info(f"[会话 {session_id}] 任务执行完成")

    except Exception as e:
        logger.error(f"[会话 {session_id}] 任务执行失败: {e}", exc_info=True)
        yield {
            "type": "error",
            "stage": "error",
            "message": f"任务执行出错: {str(e)}"
        }

async def diagnose(
    self,
    session_id: str = "default"
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    AIOps 诊断接口（兼容旧接口）

    Args:
        session_id: 会话ID

    Yields:
        Dict[str, Any]: 诊断过程的流式事件
    """
    # 使用固定的 AIOps 任务描述
    from textwrap import dedent
    aiops_task = dedent("""诊断当前系统是否存在告警，如果存在告警请详细分析告警原因并生成诊断报告，诊断报告输出格式要求：
            
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# 告警分析报告
    
---
    
## 📋 活跃告警清单
    
| 告警名称 | 级别 | 目标服务 | 首次触发时间 | 最新触发时间 | 状态 |
|---------|------|----------|-------------|-------------|------|
| [告警1名称] | [级别] | [服务名] | [时间] | [时间] | 活跃 |
| [告警2名称] | [级别] | [服务名] | [时间] | [时间] | 活跃 |
    
---
    
## 🔍 告警根因分析1 - [告警名称]
    
### 告警详情
- **告警级别**: [级别]
- **受影响服务**: [服务名]
- **持续时间**: [X分钟]
    
### 症状描述
[根据监控指标描述症状]
    
### 日志证据
[引用查询到的关键日志]
    
### 根因结论
[基于证据得出的根本原因]
    
---
    
## 🛠️ 处理方案执行1 - [告警名称]
    
### 已执行的排查步骤
1. [步骤1]
2. [步骤2]
    
### 处理建议
[给出具体的处理建议]
    
### 预期效果
[说明预期的效果]
    
---
    
## 🔍 告警根因分析2 - [告警名称]
[如果有第2个告警，重复上述格式]
    
---
    
## 📊 结论
    
### 整体评估
[总结所有告警的整体情况]
    
### 关键发现
- [发现1]
- [发现2]
    
### 后续建议
1. [建议1]
2. [建议2]
    
### 风险评估
[评估当前风险等级和影响范围]


            **重要提醒**：
            - 最终输出必须是纯 Markdown 文本，不要包含 JSON 结构
            - 所有内容必须基于工具查询的真实数据，严禁编造
            - 如果某个步骤失败，在结论中如实说明，不要跳过""")

    async for event in self.execute(aiops_task, session_id):
        # 转换事件格式以兼容旧的 API
        if event.get("type") == "complete":
            # 将 response 包装为 diagnosis 格式
            yield {
                "type": "complete",
                "stage": "diagnosis_complete",
                "message": "诊断流程完成",
                "diagnosis": {
                    "status": "completed",
                    "report": event.get("response", "")
                }
            }
        else:
            yield event

def _format_planner_event(self, state: Dict | None) -> Dict:
    """格式化 Planner 节点事件"""
    if not state:
        return {
            "type": "status",
            "stage": "planner",
            "message": "规划节点执行中"
        }

    plan = state.get("plan", [])

    return {
        "type": "plan",
        "stage": "plan_created",
        "message": f"执行计划已制定，共 {len(plan)} 个步骤",
        "plan": plan
    }

def _format_executor_event(self, state: Dict | None) -> Dict:
    """格式化 Executor 节点事件"""
    if not state:
        return {
            "type": "status",
            "stage": "executor",
            "message": "执行节点运行中"
        }

    plan = state.get("plan", [])
    past_steps = state.get("past_steps", [])

    if past_steps:
        last_step, _ = past_steps[-1]
        return {
            "type": "step_complete",
            "stage": "step_executed",
            "message": f"步骤执行完成 ({len(past_steps)}/{len(past_steps) + len(plan)})",
            "current_step": last_step,
            "remaining_steps": len(plan)
        }
    else:
        return {
            "type": "status",
            "stage": "executor",
            "message": "开始执行步骤"
        }

def _format_replanner_event(self, state: Dict | None) -> Dict:
    """格式化 Replanner 节点事件"""
    if not state:
        return {
            "type": "status",
            "stage": "replanner",
            "message": "评估节点运行中"
        }

    response = state.get("response", "")
    plan = state.get("plan", [])

    if response:
        # 已生成最终响应
        return {
            "type": "report",
            "stage": "final_report",
            "message": "最终报告已生成",
            "report": response
        }
    else:
        # 重新规划
        return {
            "type": "status",
            "stage": "replanner",
            "message": f"评估完成，{'继续执行剩余步骤' if plan else '准备生成最终响应'}",
            "remaining_steps": len(plan)
        }

全局单例

aiops_service = AIOpsService()


1. 接口无需用户传入具体问题，Agent 自己主动查询活跃告警，prompt 中已嵌入了完整的诊断任务描述
2. 接口调用 aiops_service.diagnose，内部使用固定的 AIOps 任务描述启动 Plan-Execute-Replan 工作流
3. 工作流通过 graph.astream 流式执行，每个节点完成后立即通过 SSE 推送事件给前端
4. 接收到 complete 或 error 事件后，关闭 SSE 流



`execute` 方法以固定的初始状态启动 LangGraph 工作流，通过 `stream_mode="updates"` 实时推送各节点的执行结果：

`diagnose` 方法内部构造固定的 AIOps 任务描述，将其传入通用的 `execute` 方法启动工作流。任务描述中已明确告知 Agent 该做什么，无需用户填写——这就是这个接口不需要请求参数的原因：
文章作者: wuk0Ng
文章链接: https://web-wuk0ng.github.io/2026/10/26/part05%E8%BF%90%E7%BB%B4Agent%E6%96%B9%E6%A1%88%E8%AE%BE%E8%AE%A1%E4%B8%8E%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90/
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 wuk0Ng !
agent 红队
评论
  目录