class ResolutionReport(BaseModel):
"""Structured output for resolution reports."""
resolved: bool = Field(description="Whether the issue is resolved")
solution: str = Field(description="The solution provided")
follow_up: str = Field(description="Follow-up action items")
class CustomerSupportSystem:
"""Multi-agent customer support system.
Integrates routing, multi-agent collaboration, and manual review.
Attributes:
enable_human_review: Whether to enable the manual review Hook.
"""
def __init__(self, enable_human_review: bool = False) -> None:
self.model = DashScopeChatModel(
model_name="qwen-max",
api_key=os.environ.get("DASHSCOPE_API_KEY"),
stream=True,
)
self.enable_human_review = enable_human_review
# Create the agents
self.router = self._create_router()
self.tech_agent = self._create_specialist(
"technical support",
"TechSupport",
)
self.order_agent = self._create_specialist(
"order services",
"OrderSupport",
)
self.complaint_agent = self._create_specialist(
"complaint handling",
"ComplaintHandler",
)
self.supervisor = self._create_supervisor()
# If manual review is enabled, register the hook for the supervisor
if self.enable_human_review:
self.supervisor.register_instance_hook(
hook_type="post_reply",
hook_name="human_review",
hook=human_review_post_reply_hook,
)
def _create_router(self) -> ReActAgent:
"""Create the router agent."""
return ReActAgent(
name="Router",
sys_prompt="You are an intelligent routing system that analyzes "
"customer issues and classifies them.",
model=DashScopeChatModel(
model_name="qwen-max",
api_key=os.environ.get("DASHSCOPE_API_KEY"),
stream=False,
),
formatter=DashScopeChatFormatter(),
memory=InMemoryMemory(),
toolkit=Toolkit(),
)
def _create_specialist(
self,
specialty: str,
name: str,
) -> ReActAgent:
"""Create a specialized customer support agent."""
toolkit = Toolkit()
if "order" in specialty:
toolkit.register_tool_function(query_order)
return ReActAgent(
name=name,
sys_prompt=f"You are a {specialty} specialist, professionally "
f"handling related issues.",
model=self.model,
formatter=DashScopeMultiAgentFormatter(),
memory=InMemoryMemory(),
toolkit=toolkit,
)
def _create_supervisor(self) -> ReActAgent:
"""Create the supervisor agent."""
return ReActAgent(
name="Supervisor",
sys_prompt="You are a customer service supervisor, responsible "
"for monitoring service quality and summarizing results.",
model=self.model,
formatter=DashScopeMultiAgentFormatter(),
memory=InMemoryMemory(),
toolkit=Toolkit(),
)
async def handle_customer(
self,
customer_id: str,
issue: str,
) -> str:
"""Main workflow for handling customer issues.
Flow: Route classification → Assign specialist agent →
MsgHub collaboration → Supervisor review
Args:
customer_id: The customer ID.
issue: Description of the customer's issue.
Returns:
The final response text.
"""
print(f"\n{'=' * 60}")
print(f"[New Customer Issue] {customer_id}: {issue}")
print("=" * 60)
# Step 1: Route classification
route_response = await self.router(
Msg("System", f"Analyze this customer issue: {issue}", "user"),
structured_model=RouteDecision,
)
decision = route_response.metadata
category = decision.get("category", "general")
priority = decision.get("priority", "medium")
print(f"\n[Routing Decision] Category: {category}, "
f"Priority: {priority}")
# Step 2: Assign to a specialist agent
specialist_map = {
"technical": self.tech_agent,
"order": self.order_agent,
"complaint": self.complaint_agent,
"general": self.tech_agent,
}
specialist = specialist_map.get(category, self.tech_agent)
# Step 3: Multi-agent collaborative handling (MsgHub)
async with MsgHub(participants=[specialist, self.supervisor]):
# Specialist agent handles the issue
await specialist(
Msg(
"System",
f"Please handle this customer issue: {issue}",
"user",
),
)
# Supervisor reviews and summarizes (if the human_review hook
# is enabled, manual review will be triggered automatically)
final_response = await self.supervisor(
Msg(
"System",
"Please review the handling result and provide "
"the final response.",
"user",
),
structured_model=ResolutionReport,
)
return final_response.get_text_content()
