一个 Agent 不够用？AutoGen 多智能体实战：让一队 AI 替你干活

import autogen

# 模型配置（这里用内网 vLLM 部署的开源模型，不花钱也不泄密）
config_list = [
    {
        "model": "qwen2.5-32b",
        "base_url": "http://localhost:8000/v1",
        "api_key": "EMPTY",
    }
]
llm_config = {"config_list": config_list, "temperature": 0.2}

# Agent 1：出题人
assistant = autogen.AssistantAgent(
    name="Coder",
    llm_config=llm_config,
    system_message="你是一位资深 Python 工程师，负责编写高质量代码。"
                   "代码必须包含类型注解和 docstring。",
)

# Agent 2：审查人（绝不轻易放行）
reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config=llm_config,
    system_message="你是一位严格的代码审查员。重点检查：安全漏洞、边界条件、"
                   "异常处理。发现任何问题都要指出并要求修改，"
                   "只有代码完全达标才回复 'APPROVE'。",
)

# Agent 3：人类代表（自动运行，最后把关）
user_proxy = autogen.UserProxyAgent(
    name="Admin",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=3,  # 最多自动来回 3 轮，防止死循环
    is_termination_msg=lambda msg: "APPROVE" in msg.get("content", ""),
    code_execution_config=False,
)

# 启动协作：人类发起任务，Coder 和 Reviewer 自动协作
user_proxy.initiate_chat(
    assistant,
    message="""写一个函数：解析 Nginx 访问日志，统计 Top 10 IP。
要求：日志文件可能很大（10GB+），注意内存效率。"""
)

# 发生的事情（自动展开）：
# 1. Coder 写出第一版代码
# 2. Admin 把代码转给 Reviewer 审查
# 3. Reviewer："第 12 行逐行读文件没问题，但没处理格式异常的行，
#    遇到脏数据会崩，请修复后重新提交"
# 4. Coder 修复，加上了 try-except 和跳过逻辑
# 5. Reviewer："APPROVE"
# 6. 对话结束

import autogen

config_list = [
    {"model": "qwen2.5-32b", "base_url": "http://localhost:8000/v1", "api_key": "EMPTY"}
]
llm_config = {"config_list": config_list, "temperature": 0.2}

# ===== 故障应急响应团队的四个角色 =====

# 1. 监控分析员：看指标
monitor_analyst = autogen.AssistantAgent(
    name="MonitorAnalyst",
    llm_config=llm_config,
    system_message="""你是监控分析专家，负责解读 Prometheus 指标。
你收到告警后要输出：异常指标清单、异常开始时间、影响范围初步判断。
不负责给出修复方案，只负责把"发生了什么"讲清楚。""",
)

# 2. 日志侦探：翻日志找线索
log_detective = autogen.AssistantAgent(
    name="LogDetective",
    llm_config=llm_config,
    system_message="""你是日志分析专家，负责从应用日志中找异常线索。
重点：异常堆栈、错误模式、时间相关性。只陈述日志中的事实，
不做没有证据的推测。""",
)

# 3. 应急指挥官：整合信息做决策
commander = autogen.AssistantAgent(
    name="Commander",
    llm_config=llm_config,
    system_message="""你是故障应急指挥官。综合 MonitorAnalyst 和
LogDetective 的信息，输出：
1. 根因判断（含置信度）
2. 处置方案（按优先级）
3. 是否需要人工决策的说明
方案必须包含回滚预案。""",
)

# 4. 安全审查员：复核方案
safety_reviewer = autogen.AssistantAgent(
    name="SafetyReviewer",
    llm_config=llm_config,
    system_message="""你是变更安全审查员。审查 Commander 的处置方案：
- 是否有影响其他服务的副作用
- 是否符合变更管理规范（生产操作必须可回滚）
- 高危操作（重启、扩容、切流）必须标记需要人工审批
发现问题直接打回，通过则回复 '方案通过'。""",
)

# 人类代表
admin = autogen.UserProxyAgent(
    name="OnCallEngineer",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=15,  # GroupChat 需要更多轮次
    is_termination_msg=lambda msg: "方案通过" in msg.get("content", ""),
    code_execution_config=False,
)

# ===== 组建群聊 =====
groupchat = autogen.GroupChat(
    agents=[admin, monitor_analyst, log_detective, commander, safety_reviewer],
    messages=[],
    max_round=15,               # 最多 15 轮，防失控
    speaker_selection_method="auto",  # LLM 自动选择下一个发言者
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
)

# ===== 模拟一次真实告警 =====
admin.initiate_chat(
    manager,
    message="""【P2 告警】订单服务响应延迟异常
- 告警时间：14:23
- 当前指标：
  * order-service P99 延迟：3200ms（正常 200ms）
  * order-service 错误率：8.2%（正常 0.1%）
  * 数据库连接池使用率：98%
  * Redis 缓存命中率：45%（正常 92%）
- 最近变更：1 小时前 order-service 发布了新版本 v2.3.1
- 应用日志片段（最近100行）：
  14:21:33 ERROR OrderDao - Connection timeout after 30000ms
  14:21:35 WARN  HikariPool - Connection pool nearly exhausted (49/50)
  14:22:10 ERROR CacheService - Redis GET timeout, fallback to DB
  （后续大量重复的 Connection timeout 日志）

请各位协作分析并给出处置方案。""",
)

# 自动展开的过程：
# 1. MonitorAnalyst：P99 和错误率 14:21 左右开始飙升，
#    连接池 98% 是最突出的异常，缓存命中率骤降会加剧 DB 压力
# 2. LogDetective：日志确认连接超时从 14:21:33 开始，
#    早于 v2.3.1 发布？不对，发布是 13:23，日志 14:21 才开始异常，
#    说明不是发布直接导致——更像是连接池泄漏逐渐积累到爆
# 3. Commander：综合判断，根因可能是 v2.3.1 新代码存在连接泄漏，
#    1 小时后累积到池上限。方案：① 重启 order-service 快速恢复
#    ② 回滚到 v2.3.0 根治 ③ 排查新版本连接关闭逻辑
# 4. SafetyReviewer：重启和回滚都属于高危操作，标记需人工审批；
#    回滚前确认 v2.3.0 无未修复的安全补丁——方案通过（附人工确认项）
# 5. 对话结束，OnCallEngineer 拿到一份带证据链的完整应急方案

import autogen
import requests
from datetime import datetime, timedelta

config_list = [
    {"model": "qwen2.5-32b", "base_url": "http://localhost:8000/v1", "api_key": "EMPTY"}
]

admin = autogen.UserProxyAgent(
    name="OnCallEngineer",
    human_input_mode="NEVER",
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
    code_execution_config=False,
)

monitor_analyst = autogen.AssistantAgent(
    name="MonitorAnalyst",
    llm_config={"config_list": config_list, "temperature": 0.2},
    system_message="你是监控分析专家。使用提供的工具查询指标，基于真实数据回答，"
                   "禁止编造数字。分析完成后回复 TERMINATE。",
)


# ===== 注册真实工具：查 Prometheus =====
@monitor_analyst.register_for_llm(
    name="query_prometheus",
    description="查询 Prometheus 指标。参数：query(PromQL语句), "
                "duration_minutes(回溯时长分钟数)"
)
def query_prometheus(query: str, duration_minutes: int = 30):
    """从 Prometheus 查询指标数据"""
    end = datetime.now()
    start = end - timedelta(minutes=duration_minutes)

    resp = requests.get(
        "http://prometheus.internal:9090/api/v1/query_range",
        params={
            "query": query,
            "start": start.isoformat(),
            "end": end.isoformat(),
            "step": "60s",
        },
        timeout=10,
    )
    data = resp.json()

    if data.get("status") != "success":
        return {"error": data.get("error", "unknown")}

    # 精简返回，只给 Agent 需要的信息（省 token）
    result = []
    for item in data["data"]["result"][:5]:
        metric = item.get("metric", {})
        values = item.get("values", [])
        if values:
            vals = [float(v[1]) for v in values]
            result.append({
                "metric": metric.get("__name__", ""),
                "instance": metric.get("instance", ""),
                "current": vals[-1],
                "max": max(vals),
                "min": min(vals),
                "avg": round(sum(vals) / len(vals), 2),
            })
    return result


@monitor_analyst.register_for_llm(
    name="get_active_alerts",
    description="获取当前活跃的告警列表"
)
def get_active_alerts():
    """获取 Prometheus 当前告警"""
    resp = requests.get(
        "http://prometheus.internal:9090/api/v1/alerts", timeout=10
    )
    data = resp.json()
    alerts = []
    for a in data.get("data", {}).get("alerts", []):
        alerts.append({
            "name": a["labels"].get("alertname", ""),
            "severity": a["labels"].get("severity", ""),
            "instance": a["labels"].get("instance", ""),
            "description": a.get("annotations", {}).get("description", ""),
            "active_since": a.get("activeAt", ""),
        })
    return alerts


# 工具执行注册到 admin（由人类代表侧实际执行）
# （生产实践中通常用 docker 容器执行或本地函数白名单）


# ===== 实战运行 =====
admin.initiate_chat(
    monitor_analyst,
    message="""订单服务响应慢，请你：
1. 先查当前活跃告警
2. 查询 order-service 最近 30 分钟的 P99 延迟趋势
3. 查询数据库连接池使用率
基于查询到的真实数据分析问题。分析完成后回复 TERMINATE。""",
)

# Agent 会自主决定调用哪些工具、以什么参数调用：
# → 调用 get_active_alerts()
#   发现：DatabaseConnectionPoolNearlyFull (critical)
# → 调用 query_prometheus('http_server_p99_seconds{app="order-service"}', 30)
#   发现：P99 从 0.2s 攀升到 3.2s
# → 调用 query_prometheus('hikaricp_connections_usage_ratio{app="order-service"}', 30)
#   发现：连接池使用率 30 分钟内从 40% 爬到 98%
# → 综合输出："连接池在 30 分钟内被逐渐耗尽，趋势是爬坡式的，
#   符合连接泄漏特征，而非流量突增（流量指标平稳）..."

from autogen import GroupChat

# 显式指定发言顺序的合法转移，避免群聊跑偏
allowed_transitions = {
    admin: [monitor_analyst, log_detective],
    monitor_analyst: [commander, log_detective],
    log_detective: [commander, monitor_analyst],
    commander: [safety_reviewer, monitor_analyst],
    safety_reviewer: [commander, admin],  # 审查打回指挥官，通过则回人类
}

groupchat = autogen.GroupChat(
    agents=[admin, monitor_analyst, log_detective, commander, safety_reviewer],
    messages=[],
    max_round=12,
    speaker_transitions_type="allowed",
    allowed_or_speaker_transitions_dict=allowed_transitions,
)