掌握三大设计模式，构建协同智能体集群，电商场景下单转化率提升35%
MAS核心价值：
复杂任务解耦
领域专家分工协作
系统容错性提升
from autogen import AssistantAgent, UserProxyAgent, GroupChatManager
# 1. 创建智能体群组
pricing_agent = AssistantAgent(
name="定价专家",
system_message="你负责动态定价策略，分析竞品价格和用户画像",
llm_config={"config_list": [...]}
)
inventory_agent = AssistantAgent(
name="库存管家",
system_message="你管理实时库存数据，预测补货需求",
llm_config={"config_list": [...]}
)
user_profile_agent = AssistantAgent(
name="用户分析官",
system_message="你分析用户行为和购买历史",
llm_config={"config_list": [...]}
)
# 2. 创建集中控制器
group_chat_manager = GroupChatManager(
name="促销指挥中心",
agents=[pricing_agent, inventory_agent, user_profile_agent],
max_round=10
)
# 3. 用户代理发起任务
user_proxy = UserProxyAgent(
name="用户代理",
human_input_mode="NEVER",
code_execution_config=False
)
# 4. 执行协同任务
user_proxy.initiate_chat(
manager=group_chat_manager,
message="为VIP用户设计iPhone 15促销方案，目标提升销量20%"
)控制流程：
[指挥中心] → 分配任务： 1. 定价专家：分析竞品价格 2. 用户分析官：提取VIP用户特征 3. 库存管家：检查iPhone 15库存 [定价专家] → 建议：限时折扣$799（原价$899） [用户分析官] → 建议：赠送AirPods（VIP用户偏好） [库存管家] → 警告：库存仅剩500台，需限购 [指挥中心] → 生成最终方案： "VIP专享：iPhone 15限时$799，赠AirPods（每人限购2台）"
import pika
import threading
# RabbitMQ连接设置
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
# 声明消息交换中心
channel.exchange_declare(exchange='agent_comm', exchange_type='topic')
# 1. 定价智能体
def pricing_agent():
channel.queue_declare(queue='pricing_in')
channel.queue_bind(exchange='agent_comm', queue='pricing_in', routing_key='price.*')
def callback(ch, method, properties, body):
print(f"定价收到: {body}")
# 处理逻辑...
ch.basic_publish(exchange='agent_comm', routing_key='inventory.update', body="新价格策略")
channel.basic_consume(queue='pricing_in', on_message_callback=callback, auto_ack=True)
channel.start_consuming()
# 2. 库存智能体
def inventory_agent():
# 类似实现...
# 启动智能体线程
threading.Thread(target=pricing_agent).start()
threading.Thread(target=inventory_agent).start()
# 发送任务
channel.basic_publish(
exchange='agent_comm',
routing_key='price.request',
body="用户A请求MacBook Pro报价"
)消息路由拓扑：
import torch
from torch import nn
from fl_aggregator import FederatedAveraging
# 1. 医院本地模型
class HospitalModel(nn.Module):
def __init__(self):
super().__init__()
self.linear = nn.Linear(10, 2) # 简化示例
def forward(self, x):
return self.linear(x)
# 2. 联邦聚合器
class MedicalFLAggregator:
def __init__(self):
self.global_model = HospitalModel()
self.hospital_models = {}
def receive_update(self, hospital_id, model_state):
self.hospital_models[hospital_id] = model_state
def aggregate(self):
# 安全聚合（避免泄露原始数据）
avg_state = {}
for key in self.global_model.state_dict().keys():
params = [model[key] for model in self.hospital_models.values()]
avg_state[key] = torch.stack(params).mean(dim=0)
self.global_model.load_state_dict(avg_state)
return self.global_model.state_dict()
# 3. 医院节点训练
def hospital_train(hospital_id, local_data):
local_model = HospitalModel()
# 加载全局模型参数
local_model.load_state_dict(global_params)
# 本地训练（数据不出院）
optimizer = torch.optim.SGD(local_model.parameters(), lr=0.01)
for epoch in range(10):
for x, y in local_data:
loss = nn.CrossEntropyLoss()(local_model(x), y)
loss.backward()
optimizer.step()
# 上传参数更新
aggregator.receive_update(hospital_id, local_model.state_dict())
# 基于AutoGen的增强版协同
from autogen import ConversableAgent
class PricingAgent(ConversableAgent):
def __init__(self):
super().__init__("pricing_agent")
self.register_function(self.get_dynamic_price)
def get_dynamic_price(self, product_id, user_tier):
"""
动态定价算法：
- 基础价格 × 用户等级系数
- 竞品价格监测
- 库存压力因子
"""
base_price = db.get_price(product_id)
tier_factor = {"regular":1.0, "vip":0.9, "svip":0.8}[user_tier]
comp_price = scrape_competitor(product_id)
inventory_pressure = inventory_agent.get_pressure(product_id)
final_price = base_price * tier_factor * 0.95 if comp_price < base_price else base_price
return max(final_price, base_price * 0.7) # 保底折扣
# 智能体协同协议
agents = [
PricingAgent(),
InventoryAgent(),
UserProfileAgent()
]
def collaborative_promotion(user_query):
# 1. 用户分析
user_tier = user_profile_agent.analyze_user(user_query.user_id)
# 2. 并行获取数据
with ThreadPoolExecutor() as executor:
price_future = executor.submit(
pricing_agent.get_dynamic_price,
user_query.product_id, user_tier
)
stock_future = executor.submit(
inventory_agent.check_stock,
user_query.product_id
)
# 3. 生成促销方案
promotion = f"""
{user_tier}专享优惠：
- 价格：${price_future.result():.2f}
- 库存：{stock_future.result()['available']}件
- 赠品：{user_profile_agent.get_gift_suggestion(user_query.user_id)}
"""
return promotion
# Prometheus监控指标 AGENT_REQUEST_COUNTER = Counter( 'agent_requests_total', 'Total agent requests', ['agent_type', 'status'] ) AGENT_LATENCY = Summary( 'agent_request_latency_seconds', 'Agent processing latency' ) @AGENT_LATENCY.time() def handle_request(request): try: result = agent.process(request) AGENT_REQUEST_COUNTER.labels(agent_type="pricing", status="success").inc() return result except Exception: AGENT_REQUEST_COUNTER.labels(agent_type="pricing", status="fail").inc()
Grafana看板关键指标：
智能体请求量/成功率
平均响应延迟
资源利用率（CPU/内存）
消息队列深度
智能体通信死锁
症状：A等待B的响应，B等待A的输出
解决方案：
# 添加通信超时
def send_message(receiver, msg, timeout=5):
start = time.time()
while not receiver.has_response():
if time.time() - start > timeout:
raise TimeoutError("智能体响应超时")
time.sleep(0.1)分布式状态不一致
场景：库存智能体显示有货，实际已售罄
解决方案：
# 实现分布式事务
def update_inventory(product_id, delta):
with distributed_lock(product_id): # 获取分布式锁
current = db.get_inventory(product_id)
if current + delta < 0:
raise InventoryError("库存不足")
db.update_inventory(product_id, current+delta)联邦学习梯度泄露
风险：从梯度反推原始数据
防护方案：
# 添加差分隐私噪声 def secure_aggregate(gradients): noise = torch.randn_like(gradients) * 0.1 # 添加高斯噪声 return gradients + noise
每个智能体保持单一职责
消息协议标准化（JSON Schema）
关键操作需幂等设计
生产系统必须添加速率限制
遵循此指南，可构建高可靠多智能体系统，建议从电商促销或智能客服场景切入实践！更多AI大模型应用开发学习视频内容和资料，尽在聚客AI学院。
