AutoGen实战：用Python构建智能客服系统的3种消息处理模式（附完整代码）
·
AutoGen实战：用Python构建智能客服系统的3种消息处理模式（附完整代码）
想象一下，你正在为一家快速发展的电商公司搭建智能客服系统。每天涌入成千上万的用户咨询，有简单的订单查询，有紧急的售后问题，还有需要专家处理的复杂技术咨询。如果让一个AI客服处理所有问题，它要么忙不过来，要么专业度不够。这时候，多智能体协作就成了解决问题的关键。
AutoGen作为微软开源的AI智能体框架，正是为解决这类复杂协作场景而生。它不像传统的单智能体系统那样“一人包办”，而是让多个专业智能体各司其职，通过高效的消息传递机制协同工作。今天，我将带你深入探索AutoGen在智能客服场景下的三种核心消息处理模式，每种模式都配有可直接运行的完整代码，让你能够快速应用到实际项目中。
1. 智能客服系统的架构挑战与AutoGen的解决方案
在深入代码之前，我们先理解一下智能客服系统面临的真实挑战。传统的单智能体客服系统通常采用“一问一答”的线性模式，这种架构在面对复杂业务场景时显得力不从心。比如，一个用户咨询可能涉及订单状态查询、物流跟踪、售后政策等多个环节，单一智能体很难同时具备所有这些专业知识。
更糟糕的是，当多个用户同时咨询时，系统要么排队处理导致响应延迟，要么并行处理但质量下降。VIP客户和普通客户的需求优先级如何区分？技术问题如何转接给专家？这些都是实际部署中必须解决的问题。
AutoGen通过多智能体协作架构，将复杂问题分解为多个子任务，由专门的智能体处理。这种设计有几个显著优势：
- 专业化分工：每个智能体专注于特定领域，提供更精准的服务
- 并行处理能力：多个智能体可以同时处理不同用户的请求
- 灵活路由机制：根据消息类型、优先级等条件智能分配任务
- 可扩展性：新业务上线时，只需添加相应的智能体，无需重构整个系统
下面这个表格对比了传统单智能体与AutoGen多智能体架构的关键差异：
| 维度 | 传统单智能体系统 | AutoGen多智能体系统 |
|---|---|---|
| 处理能力 | 单一模型处理所有任务 | 多个专业模型分工协作 |
| 并发性能 | 顺序处理，容易拥堵 | 并行处理，高并发支持 |
| 专业度 | 通用知识，深度不足 | 领域专家，深度服务 |
| 扩展性 | 整体升级，成本高 | 模块化添加，成本低 |
| 容错性 | 单点故障影响全局 | 局部故障不影响整体 |
理解了这些优势，我们开始进入实战环节。我将从最简单的广播模式开始，逐步深入到更复杂的路由和点对点通信模式。
2. 模式一：单消息广播 - 客服日志与通知系统
让我们从最基础的消息处理模式开始。在智能客服系统中，经常有这样的需求：一条用户消息需要被多个处理模块同时知晓。比如，用户提交了一个咨询，系统需要同时记录日志、更新用户画像、触发满意度调查准备。
这就是典型的“一对多”场景，AutoGen通过主题订阅机制完美支持这种需求。多个智能体订阅同一个主题，当消息发布到该主题时，所有订阅者都会收到消息并并行处理。
2.1 广播模式的核心实现
下面是一个完整的广播模式实现示例，模拟了客服系统中的日志记录场景：
import asyncio
import time
from dataclasses import dataclass
from typing import List
from autogen_core import (
AgentId, RoutedAgent, SingleThreadedAgentRuntime,
DefaultTopicId, default_subscription, message_handler
)
@dataclass
class CustomerQuery:
"""客户查询数据类"""
query_id: str
customer_id: str
question: str
timestamp: float = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
# 使用默认主题订阅装饰器
@default_subscription
class LoggingAgent(RoutedAgent):
"""日志记录智能体 - 负责记录所有用户查询"""
@message_handler
async def handle_query(self, message: CustomerQuery, ctx) -> None:
# 模拟日志记录过程
print(f"[{self.id}] 开始记录查询 {message.query_id}")
await asyncio.sleep(0.5) # 模拟IO操作
log_entry = f"时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(message.timestamp))} | "
log_entry += f"客户: {message.customer_id} | 查询: {message.question[:50]}..."
print(f"[{self.id}] 日志记录完成: {log_entry}")
@default_subscription
class AnalyticsAgent(RoutedAgent):
"""分析智能体 - 实时分析用户查询模式"""
@message_handler
async def handle_query(self, message: CustomerQuery, ctx) -> None:
print(f"[{self.id}] 开始分析查询 {message.query_id}")
await asyncio.sleep(0.8) # 模拟分析处理
# 简单的关键词分析
keywords = ["退货", "退款", "换货", "维修"]
detected = [kw for kw in keywords if kw in message.question]
if detected:
print(f"[{self.id}] 分析结果: 查询包含售后关键词 {detected}")
else:
print(f"[{self.id}] 分析结果: 普通咨询类查询")
@default_subscription
class SurveyAgent(RoutedAgent):
"""调查准备智能体 - 为后续满意度调查做准备"""
def __init__(self, description: str):
super().__init__(description)
self.pending_surveys = []
@message_handler
async def handle_query(self, message: CustomerQuery, ctx) -> None:
print(f"[{self.id}] 准备满意度调查 for {message.customer_id}")
await asyncio.sleep(0.3)
# 根据问题类型准备不同的调查问卷
if "不满意" in message.question or "投诉" in message.question:
survey_type = "投诉处理满意度调查"
elif "感谢" in message.question or "好评" in message.question:
survey_type = "服务体验调查"
else:
survey_type = "常规服务满意度调查"
self.pending_surveys.append({
"customer_id": message.customer_id,
"query_id": message.query_id,
"survey_type": survey_type,
"trigger_time": message.timestamp + 3600 # 1小时后触发
})
print(f"[{self.id}] 调查准备完成: {survey_type}")
async def main():
"""主函数 - 演示广播模式"""
# 创建运行时环境
runtime = SingleThreadedAgentRuntime()
# 注册三个智能体实例
agents = [
("log_agent_1", "日志记录器-主", LoggingAgent),
("log_agent_2", "日志记录器-备份", LoggingAgent),
("analytics_agent", "实时分析引擎", AnalyticsAgent),
("survey_agent", "调查准备系统", SurveyAgent)
]
for agent_id, description, agent_class in agents:
await agent_class.register(
runtime,
agent_id,
lambda desc=description, cls=agent_class: cls(desc)
)
# 启动运行时
print("=" * 60)
print("启动客服系统广播模式演示...")
print("=" * 60)
runtime.start()
# 模拟用户查询
sample_queries = [
CustomerQuery("Q001", "CUST1001", "我的订单12345什么时候能发货？"),
CustomerQuery("Q002", "CUST1002", "收到的商品有破损，要求退货退款！"),
CustomerQuery("Q003", "CUST1003", "客服服务很专业，非常感谢！"),
CustomerQuery("Q004", "CUST1004", "如何使用优惠券？操作不太明白")
]
# 发布消息到默认主题
for query in sample_queries:
print(f"\n[系统] 收到客户查询: {query.customer_id} - {query.question}")
await runtime.publish_message(query, topic_id=DefaultTopicId())
await asyncio.sleep(1) # 给处理留出时间
# 等待所有消息处理完成
await runtime.stop_when_idle()
# 显示调查准备情况
survey_agent_instance = runtime.get_agent("survey_agent")
if survey_agent_instance and hasattr(survey_agent_instance, 'pending_surveys'):
print(f"\n{'='*60}")
print("待触发的满意度调查:")
for survey in survey_agent_instance.pending_surveys:
trigger_time = time.strftime('%H:%M:%S',
time.localtime(survey['trigger_time']))
print(f" - 客户 {survey['customer_id']}: {survey['survey_type']} "
f"(计划于 {trigger_time} 发送)")
if __name__ == "__main__":
asyncio.run(main())
2.2 广播模式的关键机制解析
运行上面的代码，你会看到类似这样的输出：
============================================================
启动客服系统广播模式演示...
============================================================
[系统] 收到客户查询: CUST1001 - 我的订单12345什么时候能发货？
[日志记录器-主] 开始记录查询 Q001
[日志记录器-备份] 开始记录查询 Q001
[实时分析引擎] 开始分析查询 Q001
[调查准备系统] 准备满意度调查 for CUST1001
[调查准备系统] 调查准备完成: 常规服务满意度调查
[日志记录器-主] 日志记录完成: 时间: 2024-01-15 10:30:00 | 客户: CUST1001 | 查询: 我的订单12345什么时候能发货？...
[实时分析引擎] 分析结果: 普通咨询类查询
[日志记录器-备份] 日志记录完成: 时间: 2024-01-15 10:30:00 | 客户: CUST1001 | 查询: 我的订单12345什么时候能发货？...
从输出中可以看到，一条客户查询被四个智能体同时处理。这种模式的核心机制是：
- 主题订阅：所有智能体都使用
@default_subscription装饰器订阅默认主题 - 消息发布：通过
runtime.publish_message()将消息发布到默认主题 - 并行处理：所有订阅者同时收到消息并独立处理
- 异步执行：使用
asyncio.sleep()模拟实际处理时间，展示真正的并行性
注意：在实际生产环境中，你需要考虑消息处理的顺序依赖问题。如果某些处理必须在其他处理完成后才能进行，广播模式可能不是最佳选择。这时可以考虑使用后续介绍的路由模式。
2.3 广播模式的应用场景与优化
广播模式特别适合以下客服场景：
- 审计与合规：所有对话都需要永久存档
- 实时监控：运营团队需要实时看到客服对话
- 多维度分析：同时进行情感分析、意图识别、关键词提取
- 备份冗余：主备日志系统同时记录确保数据安全
在实际部署时，有几个优化点需要考虑：
# 优化1：添加错误处理机制
@default_subscription
class RobustLoggingAgent(RoutedAgent):
@message_handler
async def handle_query(self, message: CustomerQuery, ctx) -> None:
try:
# 业务逻辑
await self._process_query(message)
except Exception as e:
# 记录错误但不影响其他智能体
print(f"[{self.id}] 处理失败: {e}")
# 可以重试或放入死信队列
await self._retry_or_dead_letter(message, e)
async def _process_query(self, message):
# 实际的业务逻辑
await asyncio.sleep(0.5)
async def _retry_or_dead_letter(self, message, error):
# 重试或死信队列逻辑
pass
# 优化2：添加处理优先级
class PriorityBroadcastSystem:
"""带优先级的广播系统"""
def __init__(self):
self.high_priority_agents = [] # 高优先级处理者
self.normal_priority_agents = [] # 普通优先级处理者
async def broadcast(self, message, priority="normal"):
"""根据优先级选择广播范围"""
if priority == "high":
targets = self.high_priority_agents + self.normal_priority_agents
else:
targets = self.normal_priority_agents
# 并行发送给所有目标
tasks = [agent.process(message) for agent in targets]
await asyncio.gather(*tasks, return_exceptions=True)
广播模式虽然简单，但为构建更复杂的消息处理系统奠定了基础。接下来，我们将看到如何根据消息内容进行智能路由。
3. 模式二：优先级路由 - VIP客户与紧急工单处理
在真实的客服系统中，不是所有消息都应该被平等对待。VIP客户的咨询需要优先响应，紧急的技术问题需要转给专家，普通的查询可以由通用客服处理。这就是类型化主题路由的用武之地。
3.1 路由模式的核心概念
路由模式的核心思想是：根据消息的特征（类型、标签、内容等）将其发送到特定的处理通道。在AutoGen中，这通过TopicId和type_subscription装饰器实现。
让我们构建一个完整的客服路由系统：
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Any
from autogen_core import (
RoutedAgent, SingleThreadedAgentRuntime,
TopicId, type_subscription, message_handler
)
class QueryPriority(Enum):
"""查询优先级枚举"""
NORMAL = "normal" # 普通咨询
VIP = "vip" # VIP客户
URGENT = "urgent" # 紧急问题
TECHNICAL = "technical" # 技术问题
@dataclass
class CustomerServiceQuery:
"""客服查询数据类"""
query_id: str
customer_id: str
customer_tier: str # 客户等级: standard, premium, vip
question: str
category: str # 问题分类: billing, technical, general, complaint
priority: QueryPriority = QueryPriority.NORMAL
def __post_init__(self):
# 根据客户等级自动提升优先级
if self.customer_tier == "vip":
self.priority = QueryPriority.VIP
# 根据问题分类设置优先级
elif self.category == "technical":
self.priority = QueryPriority.TECHNICAL
elif "紧急" in self.question or "立刻" in self.question:
self.priority = QueryPriority.URGENT
@dataclass
class ServiceResponse:
"""服务响应数据类"""
query_id: str
agent_id: str
response: str
processing_time: float
resolved: bool = False
# 定义结果收集主题
RESULTS_TOPIC = TopicId(type="service_results", source="customer_service")
# VIP客户专属处理通道
@type_subscription(topic_type=QueryPriority.VIP.value)
class VIPAgent(RoutedAgent):
"""VIP客户专属客服 - 提供最高优先级服务"""
def __init__(self, description: str):
super().__init__(description)
self.vip_customers = {} # VIP客户历史记录
@message_handler
async def handle_vip_query(self, message: CustomerServiceQuery, ctx) -> None:
print(f"🎖️ [{self.id}] 处理VIP客户 {message.customer_id} 的查询")
# VIP客户专属逻辑
start_time = asyncio.get_event_loop().time()
# 检查客户历史
if message.customer_id in self.vip_customers:
history = self.vip_customers[message.customer_id]
greeting = f"再次欢迎您，尊贵的VIP客户！您有 {history['query_count']} 次咨询记录。"
else:
greeting = "欢迎尊贵的VIP客户！我是您的专属客服，将为您提供优先服务。"
self.vip_customers[message.customer_id] = {
"query_count": 0,
"first_seen": start_time
}
# 模拟VIP专属处理（更快的响应）
await asyncio.sleep(0.3) # VIP处理更快
# 更新历史
if message.customer_id in self.vip_customers:
self.vip_customers[message.customer_id]["query_count"] += 1
processing_time = asyncio.get_event_loop().time() - start_time
response_text = f"{greeting} 您的问题'{message.question}'已收到，我们将优先处理。预计解决时间：5分钟内。"
# 发送响应到结果主题
response = ServiceResponse(
query_id=message.query_id,
agent_id=self.id,
response=response_text,
processing_time=processing_time,
resolved=True
)
await self.publish_message(response, topic_id=RESULTS_TOPIC)
# 紧急问题处理通道
@type_subscription(topic_type=QueryPriority.URGENT.value)
class UrgentSupportAgent(RoutedAgent):
"""紧急支持团队 - 处理高优先级问题"""
@message_handler
async def handle_urgent_query(self, message: CustomerServiceQuery, ctx) -> None:
print(f"🚨 [{self.id}] 处理紧急问题: {message.question[:30]}...")
start_time = asyncio.get_event_loop().time()
# 紧急问题处理逻辑
urgent_keywords = ["宕机", "无法使用", "紧急", "立刻", "马上"]
is_critical = any(keyword in message.question for keyword in urgent_keywords)
if is_critical:
# 关键问题，立即升级
await asyncio.sleep(0.2) # 极速响应
response_text = "【紧急响应】您的问题已被标记为最高优先级，技术团队正在处理中。"
else:
# 一般紧急问题
await asyncio.sleep(0.5)
response_text = "您的问题已进入紧急处理队列，客服专员将尽快联系您。"
processing_time = asyncio.get_event_loop().time() - start_time
response = ServiceResponse(
query_id=message.query_id,
agent_id=self.id,
response=response_text,
processing_time=processing_time,
resolved=is_critical # 关键问题标记为已解决
)
await self.publish_message(response, topic_id=RESULTS_TOPIC)
# 技术问题处理通道
@type_subscription(topic_type=QueryPriority.TECHNICAL.value)
class TechnicalSupportAgent(RoutedAgent):
"""技术支持专家 - 处理技术性问题"""
def __init__(self, description: str, expertise: list):
super().__init__(description)
self.expertise = expertise # 专业领域列表
@message_handler
async def handle_technical_query(self, message: CustomerServiceQuery, ctx) -> None:
print(f"🔧 [{self.id}] 处理技术问题: {message.question[:30]}...")
start_time = asyncio.get_event_loop().time()
# 技术问题分类处理
technical_areas = {
"api": ["API", "接口", "调用", "SDK"],
"database": ["数据库", "SQL", "连接", "查询"],
"performance": ["性能", "慢", "卡顿", "延迟"],
"bug": ["错误", "bug", "故障", "异常"]
}
# 识别问题领域
detected_area = None
for area, keywords in technical_areas.items():
if any(keyword in message.question for keyword in keywords):
detected_area = area
break
if detected_area and detected_area in self.expertise:
# 专业领域内的问题
await asyncio.sleep(1.0) # 技术问题需要更多时间
response_text = f"【技术专家】您的问题属于{detected_area.upper()}领域，我正在为您深入分析。"
resolved = False # 技术问题通常需要进一步跟进
else:
# 非专业领域，转给其他专家
await asyncio.sleep(0.3)
response_text = "您的问题需要其他领域专家处理，正在为您转接..."
resolved = False
processing_time = asyncio.get_event_loop().time() - start_time
response = ServiceResponse(
query_id=message.query_id,
agent_id=self.id,
response=response_text,
processing_time=processing_time,
resolved=resolved
)
await self.publish_message(response, topic_id=RESULTS_TOPIC)
# 普通咨询处理通道
@type_subscription(topic_type=QueryPriority.NORMAL.value)
class GeneralSupportAgent(RoutedAgent):
"""通用客服 - 处理普通咨询"""
def __init__(self, description: str, capacity: int = 10):
super().__init__(description)
self.capacity = capacity
self.current_load = 0
@message_handler
async def handle_normal_query(self, message: CustomerServiceQuery, ctx) -> None:
# 检查负载
if self.current_load >= self.capacity:
print(f"⚠️ [{self.id}] 负载过高，排队中...")
await asyncio.sleep(2.0) # 模拟排队
else:
self.current_load += 1
print(f"📞 [{self.id}] 处理普通咨询: {message.question[:30]}...")
start_time = asyncio.get_event_loop().time()
await asyncio.sleep(1.5) # 普通咨询处理时间
# 通用响应模板
templates = [
"您好，关于您的问题'{question}'，我们的标准处理流程是...",
"感谢您的咨询，针对'{question}'，建议您先尝试...",
"您的问题已记录，客服专员将在24小时内回复您。"
]
import random
response_text = random.choice(templates).format(question=message.question[:20])
processing_time = asyncio.get_event_loop().time() - start_time
response = ServiceResponse(
query_id=message.query_id,
agent_id=self.id,
response=response_text,
processing_time=processing_time,
resolved=False # 普通咨询通常需要人工跟进
)
await self.publish_message(response, topic_id=RESULTS_TOPIC)
self.current_load -= 1
# 结果收集器
class ResultCollector(RoutedAgent):
"""收集所有处理结果"""
def __init__(self, description: str):
super().__init__(description)
self.results = []
self.stats = {
"total": 0,
"resolved": 0,
"avg_time": 0.0,
"by_agent": {}
}
@message_handler
async def collect_result(self, message: ServiceResponse, ctx) -> None:
self.results.append(message)
self.stats["total"] += 1
if message.resolved:
self.stats["resolved"] += 1
# 更新按智能体统计
if message.agent_id not in self.stats["by_agent"]:
self.stats["by_agent"][message.agent_id] = {
"count": 0,
"total_time": 0.0
}
agent_stats = self.stats["by_agent"][message.agent_id]
agent_stats["count"] += 1
agent_stats["total_time"] += message.processing_time
# 计算平均时间
total_time = sum(r.processing_time for r in self.results)
self.stats["avg_time"] = total_time / len(self.results) if self.results else 0
print(f"📊 [{self.id}] 收到结果: {message.agent_id} 处理了 {message.query_id} "
f"(耗时: {message.processing_time:.2f}s, 解决: {message.resolved})")
def print_summary(self):
"""打印统计摘要"""
print("\n" + "="*60)
print("客服系统处理统计:")
print("="*60)
print(f"总查询数: {self.stats['total']}")
print(f"已解决数: {self.stats['resolved']}")
print(f"解决率: {self.stats['resolved']/self.stats['total']*100:.1f}%")
print(f"平均处理时间: {self.stats['avg_time']:.2f}秒")
print("\n按智能体统计:")
for agent_id, stats in self.stats["by_agent"].items():
avg_time = stats["total_time"] / stats["count"] if stats["count"] > 0 else 0
print(f" {agent_id}: {stats['count']} 个查询, 平均 {avg_time:.2f}秒/个")
async def main():
"""主函数 - 演示优先级路由系统"""
runtime = SingleThreadedAgentRuntime()
# 创建并注册各种智能体
agents_config = [
("vip_agent_1", "VIP专属客服-01", VIPAgent),
("vip_agent_2", "VIP专属客服-02", VIPAgent),
("urgent_agent", "紧急支持中心", UrgentSupportAgent),
("tech_agent_api", "API技术专家", TechnicalSupportAgent, {"expertise": ["api", "performance"]}),
("tech_agent_db", "数据库专家", TechnicalSupportAgent, {"expertise": ["database", "bug"]}),
("general_agent_1", "普通客服-01", GeneralSupportAgent, {"capacity": 5}),
("general_agent_2", "普通客服-02", GeneralSupportAgent, {"capacity": 5}),
("general_agent_3", "普通客服-03", GeneralSupportAgent, {"capacity": 5}),
]
for config in agents_config:
agent_id = config[0]
description = config[1]
agent_class = config[2]
kwargs = config[3] if len(config) > 3 else {}
await agent_class.register(
runtime,
agent_id,
lambda desc=description, cls=agent_class, kw=kwargs: cls(desc, **kw)
)
# 注册结果收集器
collector = ResultCollector("结果收集器")
await collector.register(runtime, "result_collector", lambda: collector)
# 订阅结果主题
from autogen_core import TypeSubscription
await runtime.add_subscription(
TypeSubscription(topic_type="service_results"),
collector
)
# 启动系统
print("="*60)
print("启动智能客服路由系统...")
print("="*60)
runtime.start()
# 模拟各种客户查询
test_queries = [
CustomerServiceQuery("Q001", "CUST001", "vip", "我的账户余额不对，请立刻处理！", "billing"),
CustomerServiceQuery("Q002", "CUST002", "standard", "如何修改密码？", "general"),
CustomerServiceQuery("Q003", "CUST003", "premium", "API接口返回500错误，系统宕机了！", "technical"),
CustomerServiceQuery("Q004", "CUST004", "standard", "订单状态查询", "general"),
CustomerServiceQuery("Q005", "CUST005", "vip", "需要紧急协助，产品无法使用！", "technical"),
CustomerServiceQuery("Q006", "CUST006", "standard", "数据库连接失败", "technical"),
CustomerServiceQuery("Q007", "CUST007", "premium", "咨询优惠活动", "general"),
CustomerServiceQuery("Q008", "CUST008", "standard", "投诉客服态度不好", "complaint"),
]
# 发布查询到相应主题
for query in test_queries:
print(f"\n📨 收到查询 [{query.query_id}]: {query.customer_id} ({query.customer_tier}) - {query.question}")
print(f" 分类: {query.category}, 优先级: {query.priority.value}")
# 根据优先级发布到不同主题
topic = TopicId(type=query.priority.value, source="customer_service")
await runtime.publish_message(query, topic_id=topic)
# 给系统一些处理时间
await asyncio.sleep(0.5)
# 等待所有查询处理完成
print("\n⏳ 等待所有查询处理完成...")
await asyncio.sleep(3)
# 停止系统并显示统计
await runtime.stop_when_idle()
collector.print_summary()
# 显示详细结果
print("\n" + "="*60)
print("详细处理结果:")
print("="*60)
for result in collector.results:
status = "✅ 已解决" if result.resolved else "⏳ 待跟进"
print(f"{result.query_id}: {result.agent_id} -> {result.response[:50]}... ({status})")
if __name__ == "__main__":
asyncio.run(main())
3.2 路由系统的智能决策逻辑
运行上面的代码，你会看到一个完整的优先级路由系统在工作。系统根据客户等级、问题类型和内容关键词自动分配优先级：
============================================================
启动智能客服路由系统...
============================================================
📨 收到查询 [Q001]: CUST001 (vip) - 我的账户余额不对，请立刻处理！
分类: billing, 优先级: vip
🎖️ [VIP专属客服-01] 处理VIP客户 CUST001 的查询
📊 [结果收集器] 收到结果: vip_agent_1 处理了 Q001 (耗时: 0.30s, 解决: True)
📨 收到查询 [Q003]: CUST003 (premium) - API接口返回500错误，系统宕机了！
分类: technical, 优先级: technical
🔧 [API技术专家] 处理技术问题: API接口返回500错误，系统宕机了！...
🚨 [紧急支持中心] 处理紧急问题: API接口返回500错误，系统宕机...
这里有几个关键点需要注意：
- 自动优先级分配：
CustomerServiceQuery类的__post_init__方法根据客户等级和问题内容自动设置优先级 - 多级路由：VIP客户即使咨询普通问题也会被路由到VIP通道
- 负载均衡：
GeneralSupportAgent有容量限制，防止单个智能体过载 - 结果聚合：所有处理结果都发送到统一的结果主题，由
ResultCollector集中收集和分析
3.3 路由规则的动态配置
在实际系统中，路由规则可能需要动态调整。下面是一个更灵活的路由配置方案：
class DynamicRouter:
"""动态路由器 - 根据规则引擎决定消息路由"""
def __init__(self, runtime):
self.runtime = runtime
self.routing_rules = self._load_routing_rules()
def _load_routing_rules(self):
"""从配置文件或数据库加载路由规则"""
return [
{
"condition": lambda msg: msg.customer_tier == "vip",
"topic": QueryPriority.VIP.value,
"weight": 100 # 优先级权重
},
{
"condition": lambda msg: any(word in msg.question
for word in ["紧急", "立刻", "马上", "宕机"]),
"topic": QueryPriority.URGENT.value,
"weight": 90
},
{
"condition": lambda msg: msg.category == "technical",
"topic": QueryPriority.TECHNICAL.value,
"weight": 80
},
{
"condition": lambda msg: True, # 默认规则
"topic": QueryPriority.NORMAL.value,
"weight": 10
}
]
async def route_message(self, message: CustomerServiceQuery):
"""根据规则路由消息"""
# 按权重排序规则
sorted_rules = sorted(self.routing_rules,
key=lambda x: x["weight"],
reverse=True)
# 找到第一个匹配的规则
for rule in sorted_rules:
if rule["condition"](message):
topic = TopicId(type=rule["topic"], source="customer_service")
await self.runtime.publish_message(message, topic_id=topic)
print(f"📤 路由消息到 {rule['topic']} 主题")
return
# 如果没有匹配的规则，使用默认
default_topic = TopicId(type=QueryPriority.NORMAL.value,
source="customer_service")
await self.runtime.publish_message(message, topic_id=default_topic)
# 使用动态路由器
async def process_with_dynamic_router():
runtime = SingleThreadedAgentRuntime()
router = DynamicRouter(runtime)
# 注册各种智能体...
# 处理消息时使用路由器
query = CustomerServiceQuery("Q009", "CUST009", "standard",
"系统突然宕机了！", "technical")
await router.route_message(query)
这种动态路由系统的优势在于：
- 规则可配置：路由逻辑可以通过配置文件或管理界面调整
- 权重系统：支持复杂的优先级判断
- 易于扩展：添加新规则不影响现有逻辑
- 实时生效：规则变更立即生效，无需重启系统
路由模式为客服系统提供了强大的消息分发能力，但有时候我们需要更直接的通信方式。接下来，我们将探讨第三种模式：点对点直接通信。
4. 模式三：点对点直接通信 - 专家转接与复杂问题处理
在某些客服场景中，问题需要特定的专家处理，或者需要在多个智能体之间进行复杂的协作。这时候，广播和路由模式可能不够灵活。AutoGen的直接消息传递模式允许智能体之间进行点对点的精确通信。
4.1 直接通信模式的应用场景
考虑这样一个场景：客户咨询一个复杂的售后问题，涉及订单查询、退款计算、物流跟踪等多个环节。通用客服无法处理，需要转接给专门的售后专家，专家在处理过程中可能需要查询订单数据库、计算退款金额、联系物流系统等。
下面是一个完整的专家转接系统实现：
import asyncio
from dataclasses import dataclass
from typing import Dict, List, Optional, Any
from enum import Enum
from autogen_core import (
AgentId, RoutedAgent, SingleThreadedAgentRuntime,
message_handler
)
class IssueType(Enum):
"""问题类型枚举"""
BILLING = "billing" # 账单问题
TECHNICAL = "technical" # 技术问题
LOGISTICS = "logistics" # 物流问题
REFUND = "refund" # 退款问题
COMPLAINT = "complaint" # 投诉问题
@dataclass
class CustomerIssue:
"""客户问题数据类"""
issue_id: str
customer_id: str
description: str
issue_type: IssueType
priority: int = 1 # 1-5, 5为最高
created_at: float = None
def __post_init__(self):
if self.created_at is None:
self.created_at = asyncio.get_event_loop().time()
@dataclass
class ExpertResponse:
"""专家响应数据类"""
issue_id: str
expert_id: str
response: str
requires_followup: bool = False
followup_time: Optional[float] = None
attachments: List[Dict] = None
@dataclass
class ResolutionSummary:
"""问题解决总结"""
issue_id: str
customer_id: str
resolution: str
resolution_time: float
experts_involved: List[str]
customer_satisfaction: Optional[int] = None # 1-5分
class CustomerServiceAgent(RoutedAgent):
"""前台客服 - 接收客户问题并进行初步分类"""
def __init__(self, description: str, experts: Dict[IssueType, List[AgentId]]):
super().__init__(description)
self.experts = experts # 各类型问题对应的专家列表
self.issue_counter = 0
self.escalated_issues = {} # 已转接的问题跟踪
@message_handler
async def handle_customer_query(self, message: CustomerIssue, ctx) -> ExpertResponse:
"""处理客户查询，必要时转接给专家"""
print(f"👤 [{self.id}] 收到客户问题: {message.issue_id} - {message.description[:50]}...")
# 简单问题直接处理
if message.priority <= 2 and len(message.description) < 100:
response = await self._handle_simple_issue(message)
return ExpertResponse(
issue_id=message.issue_id,
expert_id=self.id,
response=response,
requires_followup=False
)
# 复杂问题转接给专家
print(f"🔄 [{self.id}] 问题 {message.issue_id} 需要专家协助，正在转接...")
return await self._escalate_to_expert(message)
async def _handle_simple_issue(self, issue: CustomerIssue) -> str:
"""处理简单问题"""
await asyncio.sleep(0.5) # 模拟处理时间
responses = {
IssueType.BILLING: "您的账单问题已记录，财务部门将在1-2个工作日内处理。",
IssueType.TECHNICAL: "请尝试清除缓存后重新登录，如果问题依旧请联系技术支持。",
IssueType.LOGISTICS: "物流信息已查询，预计明天送达，请保持手机畅通。",
IssueType.REFUND: "退款申请已提交，通常需要3-5个工作日到账。",
IssueType.COMPLAINT: "您的投诉已记录，客服主管将尽快与您联系。"
}
return responses.get(issue.issue_type, "您的问题已收到，我们将尽快处理。")
async def _escalate_to_expert(self, issue: CustomerIssue) -> ExpertResponse:
"""转接问题给专家"""
# 选择可用的专家
available_experts = self.experts.get(issue.issue_type, [])
if not available_experts:
return ExpertResponse(
issue_id=issue.issue_id,
expert_id=self.id,
response="抱歉，当前没有相关专家在线，您的问题已加入队列。",
requires_followup=True,
followup_time=asyncio.get_event_loop().time() + 3600 # 1小时后跟进
)
# 选择第一个可用专家（实际中可能有更复杂的负载均衡逻辑）
expert_id = available_experts[0]
# 记录转接
self.escalated_issues[issue.issue_id] = {
"expert": expert_id,
"time": asyncio.get_event_loop().time(),
"status": "escalated"
}
# 发送给专家并等待响应
try:
print(f"📤 [{self.id}] 转接问题 {issue.issue_id} 给专家 {expert_id}")
response = await self.send_message(issue, recipient=expert_id)
if isinstance(response, ExpertResponse):
self.escalated_issues[issue.issue_id]["status"] = "handled"
return response
else:
raise ValueError(f"专家返回了意外类型的响应: {type(response)}")
except Exception as e:
print(f"❌ [{self.id}] 转接失败: {e}")
# 尝试其他专家
if len(available_experts) > 1:
return await self._try_alternative_expert(issue, available_experts[1:])
return ExpertResponse(
issue_id=issue.issue_id,
expert_id=self.id,
response="专家暂时无法处理，已为您创建工单。",
requires_followup=True
)
async def _try_alternative_expert(self, issue: CustomerIssue, alternative_experts: List[AgentId]) -> ExpertResponse:
"""尝试其他专家"""
for expert_id in alternative_experts:
try:
print(f"🔄 [{self.id}] 尝试转接给备选专家 {expert_id}")
response = await self.send_message(issue, recipient=expert_id)
if isinstance(response, ExpertResponse):
self.escalated_issues[issue.issue_id] = {
"expert": expert_id,
"time": asyncio.get_event_loop().time(),
"status": "handled"
}
return response
except Exception as e:
print(f"❌ [{self.id}] 备选专家 {expert_id} 也失败了: {e}")
continue
# 所有专家都失败
return ExpertResponse(
issue_id=issue.issue_id,
expert_id=self.id,
response="所有相关专家都暂时无法处理，问题已升级。",
requires_followup=True
)
class BillingExpert(RoutedAgent):
"""账单问题专家"""
def __init__(self, description: str):
super().__init__(description)
self.resolved_issues = []
@message_handler
async def handle_billing_issue(self, message: CustomerIssue, ctx) -> ExpertResponse:
print(f"💰 [{self.id}] 处理账单问题: {message.issue_id}")
# 模拟复杂的账单处理逻辑
await asyncio.sleep(1.5)
# 根据问题描述生成响应
if "扣费" in message.description or "多扣" in message.description:
resolution = "经核查，发现系统重复扣费，已安排退款至原支付渠道，预计3-5个工作日到账。"
requires_followup = True
followup_time = asyncio.get_event_loop().time() + 86400 # 24小时后跟进
elif "发票" in message.description:
resolution = "电子发票已发送至您的注册邮箱，纸质发票将在7个工作日内寄出。"
requires_followup = False
followup_time = None
else:
resolution = "您的账单问题已详细记录，财务专员将在24小时内联系您核实处理。"
requires_followup = True
followup_time = asyncio.get_event_loop().time() + 28800 # 8小时后跟进
self.resolved_issues.append(message.issue_id)
return ExpertResponse(
issue_id=message.issue_id,
expert_id=self.id,
response=resolution,
requires_followup=requires_followup,
followup_time=followup_time,
attachments=[{"type": "case_number", "value": f"BILL-{message.issue_id}"}]
)
class TechnicalExpert(RoutedAgent):
"""技术问题专家"""
def __init__(self, description: str, specialties: List[str]):
super().__init__(description)
self.specialties = specialties
self.knowledge_base = self._load_knowledge_base()
def _load_knowledge_base(self) -> Dict[str, str]:
"""加载知识库（简化示例）"""
return {
"登录问题": "请检查网络连接，清除浏览器缓存，或尝试使用忘记密码功能重置。",
"API错误": "请提供具体的错误代码和时间戳，我们的技术团队将立即排查。",
"性能问题": "建议先检查本地网络状况，如果问题持续，请提供系统配置信息。",
"兼容性问题": "请告知您使用的浏览器版本和操作系统，我们将进行兼容性测试。"
}
@message_handler
async def handle_technical_issue(self, message: CustomerIssue, ctx) -> ExpertResponse:
print(f"🔧 [{self.id}] 处理技术问题: {message.issue_id}")
# 分析问题类型
issue_category = self._categorize_issue(message.description)
# 查找解决方案
if issue_category in self.knowledge_base:
resolution = self.knowledge_base[issue_category]
requires_followup = False
else:
resolution = "您的问题比较复杂，需要进一步分析。技术团队已收到通知，将尽快与您联系。"
requires_followup = True
await asyncio.sleep(2.0) # 技术问题需要更多分析时间
return ExpertResponse(
issue_id=message.issue_id,
expert_id=self.id,
response=resolution,
requires_followup=requires_followup,
followup_time=asyncio.get_event_loop().time() + 7200 if requires_followup else None,
attachments=[{"type": "ticket", "value": f"TECH-{message.issue_id}"}]
)
def _categorize_issue(self, description: str) -> str:
"""分类技术问题"""
keywords = {
"登录": "登录问题",
"密码": "登录问题",
"无法访问": "登录问题",
"API": "API错误",
"接口": "API错误",
"错误代码": "API错误",
"慢": "性能问题",
"卡顿": "性能问题",
"延迟": "性能问题",
"浏览器": "兼容性问题",
"兼容": "兼容性问题",
"版本": "兼容性问题"
}
for keyword, category in keywords.items():
if keyword in description:
return category
return "其他技术问题"
class LogisticsExpert(RoutedAgent):
"""物流问题专家"""
def __init__(self, description: str):
super().__init__(description)
self.tracking_system = self._init_tracking_system()
def _init_tracking_system(self) -> Dict[str, Dict]:
"""初始化物流跟踪系统（模拟）"""
return {
"ORD123456": {"status": "已发货", "location": "上海中转中心", "eta": "2024-01-16"},
"ORD789012": {"status": "运输中", "location": "北京分拨中心", "eta": "2024-01-17"},
"ORD345678": {"status": "已签收", "location": "收货地址", "eta": "2024-01-15"},
}
@message_handler
async def handle_logistics_issue(self, message: CustomerIssue, ctx) -> ExpertResponse:
print(f"🚚 [{self.id}] 处理物流问题: {message.issue_id}")
await asyncio.sleep(1.0)
# 提取订单号（简化处理）
import re
order_match = re.search(r'[A-Z]{3}\d{6}', message.description)
if order_match:
order_number = order_match.group()
if order_number in self.tracking_system:
tracking_info = self.tracking_system[order_number]
resolution = f"订单 {order_number} 状态: {tracking_info['status']}, "
resolution += f"当前位置: {tracking_info['location']}, "
resolution += f"预计送达: {tracking_info['eta']}"
requires_followup = tracking_info['status'] != '已签收'
else:
resolution = f"未找到订单 {order_number} 的物流信息，请确认订单号是否正确。"
requires_followup = True
else:
resolution = "请在问题描述中提供订单号，以便查询物流信息。"
requires_followup = True
return ExpertResponse(
issue_id=message.issue_id,
expert_id=self.id,
response=resolution,
requires_followup=requires_followup
)
class ResolutionCoordinator(RoutedAgent):
"""解决协调器 - 跟踪所有问题的解决状态"""
def __init__(self, description: str):
super().__init__(description)
self.resolutions = {} # issue_id -> ResolutionSummary
self.pending_followups = [] # 需要跟进的问题
@message_handler
async def record_resolution(self, message: ResolutionSummary, ctx) -> None:
"""记录问题解决总结"""
self.resolutions[message.issue_id] = message
print(f"📋 [{self.id}] 记录解决总结: {message.issue_id}")
print(f" 解决方案: {message.resolution[:50]}...")
print(f" 参与专家: {', '.join(message.experts_involved)}")
print(f" 解决时间: {message.resolution_time:.1f}秒")
if message.customer_satisfaction:
print(f" 客户满意度: {'⭐' * message.customer_satisfaction}")
async def check_followups(self):
"""检查需要跟进的问题"""
current_time = asyncio.get_event_loop().time()
overdue = []
for issue_id, resolution in self.resolutions.items():
# 这里可以添加跟进逻辑
pass
return overdue
def get_statistics(self) -> Dict[str, Any]:
"""获取统计信息"""
total = len(self.resolutions)
if total == 0:
return {"total": 0, "avg_time": 0, "expert_distribution": {}}
avg_time = sum(r.resolution_time for r in self.resolutions.values()) / total
# 专家参与统计
expert_distribution = {}
for resolution in self.resolutions.values():
for expert in resolution.experts_involved:
expert_distribution[expert] = expert_distribution.get(expert, 0) + 1
return {
"total_issues": total,
"average_resolution_time": avg_time,
"expert_distribution": expert_distribution,
"pending_followups": len(self.pending_followups)
}
async def main():
"""主函数 - 演示点对点专家转接系统"""
runtime = SingleThreadedAgentRuntime()
# 创建专家智能体
billing_expert = BillingExpert("账单专家-张经理")
tech_expert = TechnicalExpert("技术专家-李工程师", specialties=["API", "性能", "兼容性"])
logistics_expert = LogisticsExpert("物流专家-王专员")
# 注册专家
expert_ids = {}
for expert in [billing_expert, tech_expert, logistics_expert]:
expert_id = AgentId(type=expert.__class__.__name__, name=expert.id)
await expert.register(runtime, expert_id.name, lambda e=expert: e)
expert_ids[expert.__class__.__name__] = expert_id
# 创建前台客服，配置专家映射
experts_mapping = {
IssueType.BILLING: [expert_ids["BillingExpert"]],
IssueType.TECHNICAL: [expert_ids["TechnicalExpert"]],
IssueType.LOGISTICS: [expert_ids["LogisticsExpert"]],
IssueType.REFUND: [expert_ids["BillingExpert"]], # 退款也由账单专家处理
IssueType.COMPLAINT: [expert_ids["BillingExpert"], expert_ids["TechnicalExpert"]] # 投诉可能需要多个专家
}
front_desk = CustomerServiceAgent("前台客服-总机", experts_mapping)
await front_desk.register(runtime, "front_desk", lambda: front_desk)
# 创建协调器
coordinator = ResolutionCoordinator("解决协调器")
await coordinator.register(runtime, "coordinator", lambda: coordinator)
# 启动系统
print("="*60)
print("启动专家转接客服系统...")
print("="*60)
runtime.start()
# 模拟客户问题
test_issues = [
CustomerIssue("ISS001", "CUST1001",
"我的订单ORD123456物流信息查不到，已经三天没更新了",
IssueType.LOGISTICS, priority=3),
CustomerIssue("ISS002", "CUST1002",
"本月账单多扣了200元，要求立即退款",
IssueType.BILLING, priority=5),
CustomerIssue("ISS003", "CUST1003",
"API接口返回500错误，无法正常使用",
IssueType.TECHNICAL, priority=4),
CustomerIssue("ISS004", "CUST1004",
"想咨询一下发票开具流程",
IssueType.BILLING, priority=2),
CustomerIssue("ISS005", "CUST1005",
"系统登录总是失败，提示密码错误",
IssueType.TECHNICAL, priority=3),
CustomerIssue("ISS006", "CUST1006",
"对客服态度不满意，要求投诉",
IssueType.COMPLAINT, priority=5),
]
# 处理问题并收集结果
resolutions = []
for issue in test_issues:
print(f"\n📞 客户 {issue.customer_id} 咨询: {issue.description}")
print(f" 问题类型: {issue.issue_type.value}, 优先级: {issue.priority}")
# 发送给前台客服
try:
response = await runtime.send_message(
issue,
recipient=AgentId(type="CustomerServiceAgent", name="front_desk")
)
if isinstance(response, ExpertResponse):
print(f"✅ 处理完成: {response.response[:60]}...")
# 记录解决总结
resolution_time = asyncio.get_event_loop().time() - issue.created_at
summary = ResolutionSummary(
issue_id=issue.issue_id,
customer_id=issue.customer_id,
resolution=response.response,
resolution_time=resolution_time,
experts_involved=[response.expert_id]
)
# 发送给协调器
await runtime.send_message(summary,
recipient=AgentId(type="ResolutionCoordinator",
name="coordinator"))
resolutions.append(summary)
except Exception as e:
print(f"❌ 处理失败: {e}")
# 等待所有处理完成
await asyncio.sleep(2)
# 显示统计信息
stats = coordinator.get_statistics()
print("\n" + "="*60)
print("问题解决统计:")
print("="*60)
print(f"总处理问题数: {stats['total_issues']}")
print(f"平均解决时间: {stats['average_resolution_time']:.2f}秒")
print(f"待跟进问题: {stats['pending_followups']}")
print("\n专家处理分布:")
for expert, count in stats['expert_distribution'].items():
print(f" {expert}: {count} 个问题")
# 停止系统
await runtime.stop_when_idle()
# 显示详细解决情况
print("\n" + "="*60)
print("详细解决情况:")
print("="*60)
for resolution in resolutions:
status = "✅ 已解决" if resolution.customer_satisfaction else "⏳ 处理中"
print(f"{resolution.issue_id}: {resolution.experts_involved[0]} -> "
f"{resolution.resolution[:50]}... ({status})")
if __name__ == "__main__":
asyncio.run(main())
4.2 直接通信模式的高级特性
上面的示例展示了直接通信模式的基本用法，但在实际生产环境中，我们可能需要更复杂的特性。让我们看看如何扩展这个系统：
class AdvancedExpertSystem:
"""增强的专家系统，支持更多高级特性"""
def __init__(self, runtime):
self.runtime = runtime
self.expert_pool = {} # 专家池
self.expert_availability = {} # 专家可用状态
self.issue_queue = asyncio.Queue() # 问题队列
self.assignment_history = {} # 分配历史
async def register_expert(self, expert: RoutedAgent,
capabilities: List[str],
max_concurrent: int = 3):
"""注册专家到系统"""
expert_id = expert.id
self.expert_pool[expert_id] = {
"agent": expert,
"capabilities": capabilities,
"max_concurrent": max_concurrent,
"current_tasks": 0,
"success_rate": 1.0,
"avg_response_time": 0.0
}
self.expert_availability[expert_id] = True
print(f"✅ 专家注册: {expert_id} (能力: {capabilities})")
async def assign_issue(self, issue: CustomerIssue) -> str:
"""智能分配问题给专家"""
# 1. 过滤有能力的专家
capable_experts = [
expert_id for expert_id, info in self.expert_pool.items()
if issue.issue_type.value in info["capabilities"]
]
if not capable_experts:
return await self._handle_no_expert(issue)
# 2. 过滤可用的专家
available_experts = [
expert_id for expert_id in capable_experts
if (self.expert_availability[expert_id] and
self.expert_pool[expert_id]["current_tasks"] <
self.expert_pool[expert_id]["max_concurrent"])
]
# 3. 如果没有可用专家，加入队列
if not available_experts:
await self.issue_queue.put(issue)
return f"问题 {issue.issue_id} 已加入队列，等待专家处理"
# 4. 选择最佳专家（基于历史表现）
best_expert = self._select_best_expert(available_experts, issue)
# 5. 分配任务
return await self._dispatch_to_expert(best_expert, issue)
def _select_best_expert(self, available_experts: List[str],
issue: CustomerIssue) -> str:
"""选择最佳专家"""
# 简单的选择策略：基于成功率和响应时间
scores = {}
for expert_id in available_experts:
info = self.expert_pool[expert_id]
# 成功率权重
success_score = info["success_rate"] * 0.6
# 响应时间权重（越快越好）
if info["avg_response_time"] > 0:
time_score = 1.0 / (info["avg_response_time"] + 1) * 0.4
else:
time_score = 0.4
# 优先级加成
priority_bonus = issue.priority * 0.1
scores[expert_id] = success_score + time_score + priority_bonus
# 返回分数最高的专家
return max(scores.items(), key=lambda x: x[1])[0]
async def _dispatch_to_expert(self, expert_id: str, issue: CustomerIssue) -> str:
"""分发问题给专家"""
expert_info = self.expert_pool[expert_id]
expert_info["current_tasks"] += 1
try:
# 记录开始时间
start_time = asyncio.get_event_loop().time()
# 发送消息给专家
response = await self.runtime.send_message(
issue,
recipient=AgentId(type=expert_info["agent"].__class__.__name__,
name=expert_id)
)
# 计算响应时间
response_time = asyncio.get_event_loop().time() - start_time
# 更新专家统计数据
self._update_expert_stats(expert_id, response_time, success=True)
return f"问题已分配给 {expert_id}，响应时间: {response_time:.2f}秒"
except Exception as e:
self._update_expert_stats(expert_id, 0, success=False)
return f"分配失败: {e}"
finally:
expert_info["current_tasks"] -= 1
def _update_expert_stats(self, expert_id: str, response_time: float,
success: bool):
"""更新专家统计信息"""
info = self.expert_pool[expert_id]
# 更新平均响应时间（指数移动平均）
alpha = 0.3 # 平滑因子
if info["avg_response_time"] == 0:
info["avg_response_time"] = response_time
else:
info["avg_response_time"] = (alpha * response_time +
(1 - alpha) * info["avg_response_time"])
# 更新成功率
if success:
info["success_rate"] = min(1.0, info["success_rate"] + 0.05)
else:
info["success_rate"] = max(0.0, info["success_rate"] - 0.1)
async def _handle_no_expert(self, issue: CustomerIssue) -> str:
"""处理没有专家的情况"""
# 可以在这里实现降级策略：
# 1. 转给通用客服
# 2. 加入特殊队列
# 3. 通知管理员
return f"⚠️ 没有找到处理 {issue.issue_type.value} 问题的专家，问题已升级"
async def monitor_system(self):
"""监控系统状态"""
while True:
await asyncio.sleep(10) # 每10秒检查一次
# 检查专家状态
for expert_id, info in self.expert_pool.items():
if info["current_tasks"] >= info["max_concurrent"]:
self.expert_availability[expert_id] = False
print(f"⚠️ 专家 {expert_id} 已达最大并发限制")
else:
self.expert_availability[expert_id] = True
# 处理队列中的问题
while not self.issue_queue.empty():
try:
issue = await asyncio.wait_for(self.issue_queue.get(), timeout=0.1)
result = await self.assign_issue(issue)
print(f"🔄 处理队列问题 {issue.issue_id}: {result}")
except asyncio.TimeoutError:
break
# 打印系统状态
self.print_status()
def print_status(self):
"""打印系统状态"""
print("\n" + "="*60)
print("专家系统状态监控:")
print("="*60)
for expert_id, info in self.expert_pool.items():
status = "🟢 空闲" if info["current_tasks"] == 0 else \
"🟡 忙碌" if info["current_tasks"] < info["max_concurrent"] else \
"🔴 满载"
print(f"{expert_id}: {status} ({info['current_tasks']}/{info['max_concurrent']}) | "
f"成功率: {info['success_rate']:.1%} | "
f"平均响应: {info['avg_response_time']:.2f}s")
print(f"队列长度: {self.issue_queue.qsize()}")
这个增强版的专家系统引入了几个重要概念：
- 专家能力管理：每个专家注册时声明自己的能力范围
- 负载均衡：跟踪每个专家的当前任务数，防止过载
- 智能选择：基于历史表现（成功率、响应时间）选择最佳专家
- 队列管理：当没有可用专家时，问题进入队列等待
- 系统监控：定期检查系统状态，自动调整专家可用性
4.3 直接通信模式的最佳实践
在实际项目中应用直接通信模式时，我总结了几个关键的最佳实践：
实践一：超时与重试机制
async def send_with_retry(agent, message, recipient, max_retries=3, timeout=30):
"""带重试的消息发送"""
for attempt in range(max_retries):
try:
# 设置超时
response = await asyncio.wait_for(
agent.send_message(message, recipient=recipient),
timeout=timeout
)
return response
except asyncio.TimeoutError:
print(f"⚠️ 第 {attempt + 1} 次尝试超时")
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt) # 指数退避
except Exception as e:
print(f"❌ 发送失败: {e}")
if attempt < max_retries - 1:
await asyncio.sleep(1)
raise Exception(f"发送失败，已重试 {max_retries} 次")
实践二：消息确认与回执
@dataclass
class AcknowledgedMessage:
"""带确认的消息"""
message_id: str
original_message: Any
sender: str
timestamp: float
requires_ack: bool = True
@dataclass
class MessageAck:
"""消息确认"""
message_id: str
receiver: str
received_at: float
status: str # "received", "processing", "completed", "failed"
class ReliableMessagingAgent(RoutedAgent):
"""可靠消息传递代理"""
def __init__(self, description: str):
super().__init__(description)
self.pending_acks = {} # 等待确认的消息
self.message_counter = 0
async def send_reliable(self, message: Any, recipient: AgentId,
timeout: float = 30) -> Any:
"""可靠发送消息"""
# 生成消息ID
self.message_counter += 1
msg_id = f"{self.id}_{self.message_counter}_{time.time()}"
# 创建带确认的消息
ack_msg = AcknowledgedMessage(
message_id=msg_id,
original_message=message,
sender=self.id,
timestamp=time.time(),
requires_ack=True
)
# 发送并等待确认
try:
# 发送消息
response = await asyncio.wait_for(
self.send_message(ack_msg, recipient=recipient),
timeout=timeout
)
# 检查确认
if isinstance(response, MessageAck):
if response.status == "received":
print(f"✅ 消息 {msg_id} 已被接收")
return await self._wait_for_completion(msg_id, timeout)
else:
raise Exception(f"消息确认失败: {response.status}")
else:
# 直接返回响应（对方可能不支持确认协议）
return response
except asyncio.TimeoutError:
print(f"⏰ 消息 {msg_id} 发送超时")
raise
async def _wait_for_completion(self, msg_id: str, timeout: float):
"""等待消息处理完成"""
# 这里可以实现更复杂的等待逻辑
await asyncio.sleep(0.1) # 简化处理
return f"消息 {msg_id} 处理完成"
@message_handler
async def handle_ack_message(self, message: AcknowledgedMessage, ctx):
"""处理带确认的消息"""
# 发送接收确认
ack = MessageAck(
message_id=message.message_id,
receiver=self.id,
received_at=time.time(),
status="received"
)
await self.send_message(ack,
recipient=AgentId(type="", name=message.sender))
# 处理原始消息
result = await self._process_message(message.original_message)
# 发送完成确认
completion_ack = MessageAck(
message_id=message.message_id,
receiver=self.id,
received_at=time.time(),
status="completed"
)
await self.send_message(completion_ack,
recipient=AgentId(type="", name=message.sender))
return result
实践三：消息优先级与队列管理
from enum import IntEnum
from dataclasses import dataclass, field
from typing import List
import heapq
class MessagePriority(IntEnum):
"""消息优先级"""
LOW = 1
NORMAL = 2
HIGH = 3
URGENT = 4
CRITICAL = 5
@dataclass(order=True)
class PrioritizedMessage:
"""带优先级的消息"""
priority: MessagePriority
timestamp: float
message: Any = field(compare=False)
recipient: AgentId = field(compare=False)
class PriorityMessageQueue:
"""优先级消息队列"""
def __init__(self):
self.queue = []
self.message_counter = 0
def push(self, message: Any, recipient: AgentId,
priority: MessagePriority = MessagePriority.NORMAL):
"""推送消息到队列"""
item = PrioritizedMessage(
priority=priority,
timestamp=time.time(),
message=message,
recipient=recipient
)
heapq.heappush(self.queue, item)
self.message_counter += 1
def pop(self) -> Optional[PrioritizedMessage]:
"""弹出优先级最高的消息"""
if self.queue:
return heapq.heappop(self.queue)
return None
def peek(self) -> Optional[PrioritizedMessage]:
"""查看但不移除最高优先级消息"""
if self.queue:
return self.queue[0]
return None
def size(self) -> int:
"""队列大小"""
return len(self.queue)
def clear_low_priority(self, threshold: MessagePriority = MessagePriority.NORMAL):
"""清除低于阈值的消息"""
self.queue = [item for item in self.queue if item.priority >= threshold]
heapq.heapify(self.queue)
这些最佳实践确保了直接通信模式在复杂生产环境中的可靠性和效率。通过超时重试、消息确认和优先级队列，我们可以构建出健壮的智能体通信系统。
5. 三种模式的综合应用与性能优化
在实际的智能客服系统中，我们很少只使用单一的消息处理模式。更多时候，我们需要根据不同的场景组合使用这三种模式。让我们看一个综合应用的例子，并探讨性能优化的策略。
5.1 混合模式：电商客服系统的完整实现
下面是一个电商客服系统的完整示例，它综合运用了三种消息处理模式：
import asyncio
import time
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Dict, List, Optional, Any
from collections import defaultdict
import random
from autogen_core import (
AgentId, RoutedAgent, SingleThreadedAgentRuntime,
DefaultTopicId, TopicId, default_subscription,
type_subscription, message_handler
)
class CustomerTier(Enum):
"""客户等级"""
REGULAR = "regular" # 普通客户
SILVER = "silver" # 白银会员
GOLD = "gold" # 黄金会员
PLATINUM = "platinum" # 白金会员
DIAMOND = "diamond" # 钻石会员
class QueryCategory(Enum):
"""查询分类"""
ORDER = "order" # 订单相关
PRODUCT = "product" # 产品相关
PAYMENT = "payment" # 支付相关
DELIVERY = "delivery" # 物流相关
RETURN = "
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
15
0- 0
已为社区贡献7条内容
所有评论(0)