深度实战：基于LangGraph的ReAct新一代ai多智能体架构在腾讯云TKE故障自愈中的深度实践

from typing import TypedDict, List, Annotated
from langgraph.graph.message import add_messages
import operator

class AgentState(TypedDict):
    # 消息历史，add_messages用于自动合并新消息
    messages: Annotated[List[BaseMessage], add_messages]
    # 故障根因推理结果
    root_cause: str
    # 当前执行步骤（用于防止死循环）
    iteration: int
    # 腾讯云资源ID（集群ID/实例ID）
    resource_id: str
    # 工具执行结果暂存区
    intermediate_steps: Annotated[List[Tuple[str, str]], operator.add]

graph TD
    Start[Prometheus告警触发] --> Supervisor[Supervisor Agent决策]
    Supervisor --> |需要日志分析| LogAgent[CLS日志分析Agent]
    Supervisor --> |需要资源变更| OpsAgent[TKE资源操作Agent]
    
    LogAgent --> Tool1[query_cls_logs]
    OpsAgent --> Tool2[scale_deployment / cordon_node]
    
    Tool1 --> Check{根因定位成功?}
    Tool2 --> Check
    Check --> |否| Supervisor
    Check --> |是| Execute[执行自愈策略]
    Execute --> End[恢复状态上报]

pip install langgraph langchain-openai tencentcloud-sdk-python kubernetes asyncio

import asyncio
from tencentcloud.common import credential
from tencentcloud.cls.v20201016 import cls_client, models
from langchain_core.tools import tool

@tool
async def query_cls_error_logs(topic_id: str, query_keyword: str, hours_back: int = 1):
    """
    异步查询腾讯云CLS日志，专注于提取ERROR及异常堆栈。
    Args:
        topic_id: 日志主题ID
        query_keyword: 查询关键词（如 pod_name）
        hours_back: 回溯小时数
    """
    cred = credential.Credential(os.getenv("TENCENT_SECRET_ID"), os.getenv("TENCENT_SECRET_KEY"))
    client = cls_client.ClsClient(cred, "ap-guangzhou")
    
    req = models.SearchLogRequest()
    end_time = int(time.time()) * 1000
    start_time = (int(time.time()) - hours_back * 3600) * 1000
    
    req.TopicId = topic_id
    req.From = start_time
    req.To = end_time
    req.Query = f"{query_keyword} AND ERROR"  # 精准过滤
    req.Limit = 100  # 单次抽取限制
    
    # 处理高并发下的限频重试
    retry_count = 3
    for i in range(retry_count):
        try:
            resp = client.SearchLog(req)
            # 解析JSON格式的日志内容
            logs = []
            if resp.Results:
                for result in resp.Results:
                    # 动态解析LogJson字段
                    logs.append(json.loads(result.LogJson))
            return logs
        except Exception as e:
            if "RequestLimitExceeded" in str(e):
                await asyncio.sleep(2 ** i)  # 指数退避
                continue
            return {"error": str(e)}
    return {"error": "Max retries exceeded"}

from kubernetes import client, config
from kubernetes.client.rest import ApiException

@tool
async def tke_cordon_node(node_name: str, cluster_id: str):
    """
    封锁TKE集群中的指定Node，使其不再调度新Pod。
    使用腾讯云TKE原生SDK，而非直接操作Kube-API，以绕过RBAC权限细粒度问题。
    """
    from tencentcloud.tke.v20180525 import tke_client, models as tke_models
    
    cred = credential.Credential(os.getenv("TENCENT_SECRET_ID"), os.getenv("TENCENT_SECRET_KEY"))
    tke_cli = tke_client.TkeClient(cred, "ap-guangzhou")
    
    req = tke_models.UpdateClusterNodePoolRequest()
    req.ClusterId = cluster_id
    # 实际生产环境建议通过NodePool操作，此处简化逻辑
    
    # 模拟通过label打标实现软封锁（详细实现需结合K8s API）
    try:
        # 加载集群内kubeconfig（若Agent部署在TKE内）
        config.load_incluster_config()
        v1 = client.CoreV1Api()
        # 设置污点 NoSchedule
        body = {
            "spec": {
                "taints": [{"key": "node.kubernetes.io/unreachable", "value": "auto-heal", "effect": "NoSchedule"}]
            }
        }
        v1.patch_node(node_name, body)
        return {"status": "success", "message": f"Node {node_name} has been cordoned."}
    except ApiException as e:
        return {"status": "failed", "message": f"K8s API error: {e.reason}"}

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 初始化带工具绑定的模型
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.1)
tools = [query_cls_error_logs, tke_cordon_node]
llm_with_tools = llm.bind_tools(tools)

# 系统提示词：扮演SRE专家，严格按照ReAct范式输出
SYSTEM_PROMPT = """
你是一名资深的腾讯云TKE SRE专家。你的任务是通过调用工具集分析故障根因并执行自愈操作。
你必须严格遵守以下格式：
Thought: 分析当前已知信息，推理下一步需要什么数据。
Action: 调用具体的工具(如 query_cls_error_logs)。
... (观察工具返回) ...
Thought: 我现在知道根因了，可以给出最终自愈方案。
Final Answer: 执行的具体操作指令。
"""

def reasoning_node(state: AgentState):
    """
    推理节点：调用LLM，决定下一步调用哪个工具或直接返回最终答案。
    """
    messages = state.get("messages", [])
    # 注入系统提示
    if not messages or not isinstance(messages[0], SystemMessage):
        messages = [SystemMessage(content=SYSTEM_PROMPT)] + messages
    
    response = llm_with_tools.invoke(messages)
    
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1
    }

from langgraph.prebuilt import ToolNode
import json

# 构建标准ToolNode，内部自动处理tool_calls的分发
tool_node = ToolNode(tools)

def should_continue(state: AgentState):
    """
    条件边判断函数：决定是继续调用工具还是结束流程。
    """
    last_message = state["messages"][-1]
    # 如果LLM产生了工具调用请求，则进入执行节点
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "execution"
    # 否则直接结束（代表已经给出最终答案）
    return "end"

from langgraph.checkpoint.memory import MemorySaver

# 构建图
builder = StateGraph(AgentState)

# 添加节点
builder.add_node("reasoning", reasoning_node)
builder.add_node("execution", tool_node)

# 设置入口
builder.set_entry_point("reasoning")

# 添加条件边：推理 -> 执行 或 结束
builder.add_conditional_edges(
    "reasoning",
    should_continue,
    {
        "execution": "execution",
        "end": END
    }
)

# 添加普通边：执行完成后回到推理（形成ReAct闭环）
builder.add_edge("execution", "reasoning")

# 内存检查点（生产环境替换为RedisSaver）
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# FastAPI 路由入口
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook/auto-heal")
async def handle_alert(request: Request):
    payload = await request.json()
    cluster_id = payload["cluster_id"]
    namespace = payload["namespace"]
    pod_name = payload["pod_name"]
    
    # 初始化状态
    initial_state = {
        "messages": [HumanMessage(content=f"集群 {cluster_id} 的Pod {pod_name} 频繁重启，请诊断并自愈。")],
        "resource_id": pod_name,
        "cluster_id": cluster_id,
        "iteration": 0,
        "intermediate_steps": []
    }
    
    # 运行图（thread_id用于区分不同故障工单）
    config = {"configurable": {"thread_id": f"heal-{pod_name}-{int(time.time())}"}}
    final_state = await graph.ainvoke(initial_state, config=config)
    
    return {"status": "processed", "result": final_state["messages"][-1].content}

from pydantic import BaseModel, Field

class ClsQueryInput(BaseModel):
    topic_id: str = Field(description="必须从腾讯云控制台获取的Topic ID，格式如 'xxxxxxxx-xxxx-xxxx'")
    query_keyword: str = Field(description="K8s Pod名称或Label")

@tool(args_schema=ClsQueryInput)
async def query_cls_error_logs(topic_id: str, query_keyword: str):
    # 此处可加入正则校验，若不符合uuid格式直接抛出ToolException，LLM会重新规划路径
    pass

def should_continue(state: AgentState):
    if state["iteration"] > 5:  # 最大重试5次
        return "end"
    # ... 原有逻辑