最近在尝试 LLM Multi Agent(多智能体)的应用场景,下面给一个最近觉得还比较好用,也不是很麻烦的案例。
这是基于微软的 AutoGen 框架写的一个脚本(为了阅读友好,完整代码放在了博客的最后面)。
在这个脚本中,我以解决 SRE 的需求为案例,一共有4 个 Agent,可以理解为有 4 个 SRE工程师,但是他们擅长的内容不完全一样。
可以修改脚本最后一段的 message,输入自己的问题,比如这里的“怎么部署 k8s?” ,然后执行脚本,它会在命令行窗口逐一给出上面 4 个 Agent 的回答,如果嫌啰嗦,可以只看最后一个,但是从学习的角度来说,比较建议都看一下,不同的 Agent 给出的答案还是有很大不同的。
initializer.initiate_chat(
manager,
message="""怎么部署 k8s ?""",
clear_history=False,
)
在这个脚本中,我使用的是 DeepSeek(我真是 DeepSeek 的热爱粉丝…),如下代码。
llm_config_deepseek = {
"model": "deepseek-chat",
"api_key": "sk-bbd9c7ce03e247a099e3c0506aa2a0ea",
"base_url": "https://api.deepseek.com/v1",
"temperature": 0.5,
"stream": True
}
而且只使用了这一个模型,但是从我个人的使用经验来看,使用多家不同的模型,效果会更好。我想可能是因为每一家的算法架构、训练过程和训练数据都不一样,所以最终的效果也是不一样的,使用不同的模型可以获取更大的多样性,在经过最后的总结,效果会比较好。
如果选择国内的模型,除了 DeepSeek,通义和豆包两个系列的模型也都挺好的,可以参考上面的代码进行配置。不过如果有条件,还是更推荐使用 gpt4o、claude-3.5-sonnot和 gemini-1.5-pro,这三个,或者其中的某一个。
我为了让这个脚本使用起来尽量简单,很多功能没添加,比如自动执行命令、人类介入等,所以这个脚本目前比较适用于那些相对独立、完整一些的问题,比如上面的“怎么部署 k8s”,不太适合用来处理线上问题,不是说不能处理,只是需要不断地在线上环境和这个脚本之间来回切换,很麻烦。
除了脚本中写好的 4 个适用于 SRE 的 prompt,其实改一下 Prompt 还可以使用在很多其他场景。
编写 Agent 的逻辑大致如下:
下面是代码全文。
代码有三点要说明:
pip install autogen==0.3.1
即可"api_key": "xxxxxxxxxxxxxxxxxxxx",
修改成自己的 keymax_round=20
表示最多允许 Agent 发言 20 次,如果修改脚本,想加入人类输入,或者增加 Agent 的发言次数,可能会需要修改这个上限"""
please execute "pip install autogen==0.3.1" before running this script
"""
import os
import autogen # type: ignore
from typing import Any
os.environ.setdefault("AUTOGEN_USE_DOCKER", "False")
llm_config_deepseek = {
"model": "deepseek-chat",
"api_key": "xxxxxxxxxxxxxxxxxxxx",
"base_url": "https://api.deepseek.com/v1",
"temperature": 0.5,
"stream": True
}
initializer = autogen.UserProxyAgent(
name="Init",
)
sre_engineer_01 = autogen.AssistantAgent(
name="sre_engineer_01",
llm_config=llm_config_deepseek,
system_message="""
你是一位专注于问题解决的 SRE 资深工程师,具备以下特质:
技术专长:
- 精通 Linux/Unix 系统管理和故障排查
- 熟悉容器技术和 Kubernetes 生态系统
- 深入理解分布式系统和微服务架构
- 掌握主流监控、日志和追踪工具
- 具备网络、存储、数据库等基础设施维护经验
核心能力:
- 系统性问题分析和解决能力
- 快速故障定位和恢复能力
- 性能优化和容量规划能力
- 自动化工具开发和优化能力
请作为 SRE 资深工程师,解决问题时可以参考如下步骤:
1. 问题诊断
- 症状分析和分类
- 初步影响评估
- 可能的根因列表
- 需要收集的额外信息
2. 排查方案
- 具体排查步骤
- 使用的工具和命令
- 关键日志和指标
- 排查注意事项
3. 解决方案
- 临时解决方案
- 长期修复方案
- 具体操作步骤
- 验证方法
4. 预防措施
- 监控改进建议
- 告警优化方案
- 自动化改进
- 文档和流程优化
""",
description="我是sre_engineer_01,专注于问题解决的 SRE 资深工程师,当遇到问题时,可以找我",
)
sre_engineer_02 = autogen.AssistantAgent(
name="sre_engineer_02",
llm_config=llm_config_deepseek,
system_message="""
你是一位精通 Python 的 SRE 资深工程师,具备以下核心能力:
技术专长:
- Python 高级开发和系统编程
- 自动化运维工具开发
- 分布式系统监控和故障排查
- 性能优化和容量规划
- 数据分析和可视化
核心能力:
- 系统性问题分析和解决
- 自动化框架设计与实现
- 监控系统开发与优化
- 故障排查和恢复
- CI/CD 流程自动化
请作为 Python SRE 工程师,在解决问题时参考以下几点:
1. 问题分析
- 根因定位方法
- 影响评估报告
2. 技术方案
- 自动化实现代码
- 监控集成方案
- 验证测试方法
3. 长期优化
- 性能优化建议
- 监控告警改进
- 自动化程度提升
""",
description="我是sre_engineer_02,精通 Python 的 SRE 资深工程师,当遇到问题时,可以找我",
)
sre_reflection = autogen.AssistantAgent(
name="sre_reflection",
llm_config=llm_config_deepseek,
system_message="""
你是一位具有系统思维和创新能力的 SRE 资深工程师,专注于方案优化和系统改进。
专业能力:
- 深入的技术洞察力和系统分析能力
- 丰富的方案设计和优化经验
- 全面的技术栈理解和整合能力
- 优秀的问题发现和风险预判能力
核心职责:
- 现有方案的深度分析与反思
- 改进建议的提出与验证
- 最佳实践的定制与推广
- 团队能力的建设与提升
在对方案进行系统性分析和改进时参考以下方法:
1. 可以使用SWOT分析
- Strengths:方案优势
- Weaknesses:存在问题
- Opportunities:改进机会
- Threats:潜在风险
2. 多维度评估
- 可靠性维度
* 高可用性评估
* 容灾能力分析
* 故障自愈机制
- 性能维度
* 响应时间
* 资源利用
* 扩展能力
- 运维维度
* 监控覆盖
* 问题排查
* 变更管理
- 成本维度
* 资源成本
* 人力成本
* 维护成本
请你结合问题,对 sre_engineer_01 和 sre_engineer_02 的回答进行优化
""",
description="我是sre_reflection,具有系统思维和创新能力的 SRE 资深工程师,专注于方案优化和系统改进",
)
sre_engineer_00 = autogen.AssistantAgent(
name="sre_engineer_00",
llm_config=llm_config_deepseek,
system_message="""
你是一位专注于方案简洁和可行性的 SRE 资深工程师,具备以下特质:
核心能力:
- 快速提炼复杂信息的核心要素
- 设计简洁可行的优化方案
- 准确评估方案可行性和风险
- 有效推动方案落地实施
主要职责:
- 分析和总结现有方案
- 提供精简可行的优化建议或者解决方案
- 确保方案的高效实施
以下内容非常重要:
- 全面review sre_engineer_01、sre_engineer_02 和 sre_reflection 的方案,不要有遗漏
- 尽全力提供清晰、简洁、可行的方案
- 只输出必要的文字说明,不要太啰嗦
- 要有完整的方案、步骤、命令或者代码
""",
description="我是 sre_engineer_00,专注于方案优化的 SRE 资深工程师,可以帮助你提供优化建议",
)
graph_dict = {}
graph_dict[initializer] = [sre_engineer_01]
graph_dict[sre_engineer_01] = [sre_engineer_02]
graph_dict[sre_engineer_02] = [sre_reflection]
graph_dict[sre_reflection] = [sre_engineer_00]
agents = [
initializer,
sre_engineer_01,
sre_engineer_02,
sre_reflection,
sre_engineer_00,
]
groupchat = autogen.GroupChat(
agents=agents,
messages=[],
max_round=20,
allowed_or_disallowed_speaker_transitions=graph_dict,
speaker_transitions_type="allowed",
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config_deepseek)
initializer.initiate_chat(
manager,
message="""怎么部署 k8s ?""",
clear_history=False,
)