Multi-Agent Customer Support System

# Install AgentScope
pip install agentscope

# Set the API Key
export DASHSCOPE_API_KEY=your_api_key

import asyncio
import os
from typing import Any, Literal

from pydantic import BaseModel, Field

from agentscope.agent import AgentBase, ReActAgent, UserAgent
from agentscope.formatter import (
    DashScopeChatFormatter,
    DashScopeMultiAgentFormatter,
)
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import MsgHub
from agentscope.tool import Toolkit, ToolResponse

def create_support_agent() -> ReActAgent:
    """Create a basic customer support agent."""
    model = DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        stream=True,
    )

    return ReActAgent(
        name="CustomerSupport",
        sys_prompt="""You are a professional customer service representative \
responsible for helping customers resolve their issues.

You should:
1. Answer customer questions politely and patiently
2. Accurately understand customer needs
3. Provide clear and helpful responses
4. For issues you cannot resolve, explain why and offer alternatives""",
        model=model,
        formatter=DashScopeChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=Toolkit(),
    )


async def demo_basic_support() -> None:
    """Demonstrate the basic customer support agent."""
    agent = create_support_agent()

    queries = [
        "When will my order arrive?",
        "I'd like to return an item. How do I do that?",
        "What benefits does your membership offer?",
    ]

    for query in queries:
        print(f"\nCustomer: {query}")
        response = await agent(Msg("Customer", query, "user"))
        print(f"Support: {response.get_text_content()}")


asyncio.run(demo_basic_support())

class RouteDecision(BaseModel):
    """Structured output for routing decisions."""

    category: Literal["technical", "order", "complaint", "general"] = Field(
        description=(
            "Issue category: technical (technical issues), "
            "order (order issues), complaint (complaints), "
            "general (general inquiries)"
        ),
    )
    confidence: float = Field(
        description="Classification confidence, between 0 and 1",
        ge=0,
        le=1,
    )
    summary: str = Field(
        description="Brief summary of the issue",
    )
    priority: Literal["low", "medium", "high"] = Field(
        description="Issue priority",
    )


def create_router_agent() -> ReActAgent:
    """Create a router agent."""
    model = DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        stream=False,  # Recommended to disable stream for structured output
    )

    return ReActAgent(
        name="Router",
        sys_prompt="""You are an intelligent routing system responsible for \
analyzing customer issues and deciding which specialized team should \
handle them.

Classification rules:
- technical: Product usage issues, technical failures, feature inquiries
- order: Order status, logistics, returns and exchanges
- complaint: Dissatisfaction, complaints, compensation requests
- general: Membership, promotions, general inquiries

Priority rules:
- high: Complaints, urgent order issues
- medium: General order and technical issues
- low: Inquiry-type questions""",
        model=model,
        formatter=DashScopeChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=Toolkit(),
    )


async def demo_routing() -> None:
    """Demonstrate intelligent routing classification."""
    router = create_router_agent()

    queries = [
        "My order shows shipped, but I haven't received it after three days",
        "The app keeps crashing. What should I do?",
        "This is terrible! I demand a full refund!",
        "Do you have student discounts?",
    ]

    for query in queries:
        print(f"\nCustomer: {query}")

        # Use structured_model to get structured output
        response = await router(
            Msg("Customer", query, "user"),
            structured_model=RouteDecision,
        )

        # Retrieve the structured result from metadata
        decision = response.metadata
        print(f"  → Category: {decision.get('category')}")
        print(f"  → Priority: {decision.get('priority')}")
        print(f"  → Confidence: {decision.get('confidence')}")
        print(f"  → Summary: {decision.get('summary')}")


asyncio.run(demo_routing())

                    MsgHub
                      ↓
    ┌─────────────────┼─────────────────┐
    ↓                 ↓                 ↓
 TechAgent       OrderAgent      ComplaintAgent
    │                 │                 │
    └────────── Message Broadcast ──────┘

def create_model() -> DashScopeChatModel:
    """Create a shared model instance."""
    return DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        stream=True,
    )


def create_specialist_agent(
    name: str,
    specialty: str,
    description: str,
) -> ReActAgent:
    """Create a specialized customer support agent.

    Args:
        name: Agent name.
        specialty: Area of expertise.
        description: Detailed description.
    """
    return ReActAgent(
        name=name,
        sys_prompt=f"""You are {name}, specializing in {specialty}.

{description}

When collaborating with other support agents:
- Listen carefully to your colleagues' opinions
- Provide insights within your area of expertise
- If the issue is outside your specialty, suggest handing it off to the \
appropriate colleague""",
        model=create_model(),
        # Use MultiAgentFormatter for multi-agent scenarios
        formatter=DashScopeMultiAgentFormatter(),
        memory=InMemoryMemory(),
        toolkit=Toolkit(),
    )


# Create the specialized agent team
tech_agent = create_specialist_agent(
    "TechSupport",
    "technical support",
    "You are an expert in product technical issues, troubleshooting, "
    "and feature usage guidance.",
)

order_agent = create_specialist_agent(
    "OrderSupport",
    "order services",
    "You handle order inquiries, logistics tracking, and returns/exchanges.",
)

complaint_agent = create_specialist_agent(
    "ComplaintHandler",
    "complaint handling",
    "You handle customer complaints, soothe emotions, and provide solutions.",
)


async def demo_msghub() -> None:
    """Demonstrate MsgHub multi-agent collaboration.

    All participants share context within the same MsgHub.
    Each agent's reply is automatically broadcast to other participants.
    """
    customer_issue = (
        "The app on my phone won't open, and the order shows it hasn't "
        "shipped yet. This is really disappointing!"
    )

    # Use MsgHub to enable multi-agent collaboration
    async with MsgHub(
        participants=[tech_agent, order_agent, complaint_agent],
    ):
        # Technical support analyzes the technical issue first
        await tech_agent(
            Msg(
                "Coordinator",
                f"Customer issue: {customer_issue}\n"
                "Please have technical support analyze the technical problem.",
                "user",
            ),
        )

        # Order support checks the order status (can see tech support's reply)
        await order_agent(
            Msg("Coordinator", "Please have order support check the "
                "order status.", "user"),
        )

        # Complaint handler synthesizes information and provides a solution
        # (can see the discussion from the previous two agents)
        await complaint_agent(
            Msg(
                "Coordinator",
                "Please synthesize all the above information and provide "
                "the customer with a satisfactory solution.",
                "user",
            ),
        )


asyncio.run(demo_msghub())

Orchestrator Agent
    ├── 🔧 create_tech_worker(task)   → TechWorker Agent
    ├── 🔧 create_order_worker(task)  → OrderWorker Agent
    └── 🔧 create_complaint_worker(task) → ComplaintWorker Agent

def query_order(order_id: str) -> ToolResponse:
    """Query order status.

    Args:
        order_id (``str``):
            The order ID.
    """
    # Simulate order data
    orders = {
        "12345": {"status": "shipped", "eta": "2024-01-20"},
        "67890": {"status": "processing", "eta": "2024-01-22"},
    }
    order = orders.get(order_id, {"status": "not_found"})
    return ToolResponse(
        content=[TextBlock(type="text", text=f"Order {order_id}: {order}")],
    )

async def create_tech_worker(task_description: str) -> ToolResponse:
    """Create a technical support Worker agent to handle technical issues.

    Args:
        task_description (``str``):
            Description of the technical issue to handle.
    """
    worker = ReActAgent(
        name="TechWorker",
        sys_prompt=(
            "You are a technical support specialist, skilled in product "
            "technical issues, troubleshooting, and feature usage guidance. "
            "Please answer the user's technical questions professionally "
            "and accurately."
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=Toolkit(),
    )
    res = await worker(Msg("user", task_description, "user"))
    return ToolResponse(content=res.get_content_blocks("text"))


async def create_order_worker(task_description: str) -> ToolResponse:
    """Create an order service Worker agent to handle order-related issues.

    Args:
        task_description (``str``):
            Description of the order issue to handle.
    """
    # Equip the order Worker with the order query tool
    toolkit = Toolkit()
    toolkit.register_tool_function(query_order)

    worker = ReActAgent(
        name="OrderWorker",
        sys_prompt="You are an order service specialist, responsible for "
        "order inquiries, logistics tracking, and returns/exchanges.",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=toolkit,
    )
    res = await worker(Msg("user", task_description, "user"))
    return ToolResponse(content=res.get_content_blocks("text"))


async def create_complaint_worker(task_description: str) -> ToolResponse:
    """Create a complaint handling Worker agent to handle customer complaints.

    Args:
        task_description (``str``):
            Description of the complaint to handle.
    """
    worker = ReActAgent(
        name="ComplaintWorker",
        sys_prompt=(
            "You are a complaint handling specialist, responsible for "
            "dealing with customer complaints, soothing emotions, and "
            "providing reasonable solutions. Your goal is to ensure "
            "customer satisfaction."
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        memory=InMemoryMemory(),
        toolkit=Toolkit(),
    )
    res = await worker(Msg("user", task_description, "user"))
    return ToolResponse(content=res.get_content_blocks("text"))

async def demo_handoff() -> None:
    """Demonstrate the Handoff (Orchestrator-Workers) pattern.

    The Orchestrator autonomously decides which specialized Worker to
    delegate to via tool calls.
    """
    # Register the specialized Worker creation functions as tools
    toolkit = Toolkit()
    toolkit.register_tool_function(create_tech_worker)
    toolkit.register_tool_function(create_order_worker)
    toolkit.register_tool_function(create_complaint_worker)

    orchestrator = ReActAgent(
        name="Orchestrator",
        sys_prompt="""You are the Orchestrator of the customer support system.

Your responsibilities are:
1. Analyze customer issues and determine which specialized support is needed
2. Delegate sub-tasks to the appropriate Workers (via tool calls)
3. Summarize the results from each Worker and provide the final response

For complex issues, you can delegate to multiple Workers in parallel.""",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
            stream=False,
            generate_kwargs={"parallel_tool_calls": True},
        ),
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        parallel_tool_calls=True,
    )

    # A complex customer issue involving multiple aspects
    customer_issue = (
        "The app on my phone won't open, and the order shows it hasn't "
        "shipped yet. This is really disappointing!"
    )

    print(f"Customer: {customer_issue}")
    response = await orchestrator(
        Msg("Customer", customer_issue, "user"),
    )
    print(f"\nFinal Response:\n{response.get_text_content()}")


asyncio.run(demo_handoff())

Agent.__call__()
  ├── pre_reply hooks   ← Review/modify input before agent replies
  ├── reply()            ← Agent core logic
  └── post_reply hooks  ← Review/modify output after agent replies

ReActAgent.reply()
  ├── pre_reasoning / post_reasoning  ← Before/after reasoning
  └── pre_acting / post_acting        ← Before/after action execution

async def human_review_post_reply_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
    output: Msg,
) -> Msg | None:
    """Post-reply hook: perform manual review after the agent replies.

    If the manual review does not pass, the review feedback is appended
    to the original input and the agent is called again to generate a
    new response.

    Args:
        self: The agent instance.
        kwargs: The argument dictionary of the reply function.
        output: The response message generated by the agent.

    Returns:
        Returns None if the review passes (keeping the original reply),
        otherwise returns the revised reply.
    """
    print("\n" + "=" * 50)
    print("[Manual Review] Agent response:")
    print(f"  {output.get_text_content()}")
    print("=" * 50)

    # Use UserAgent to get human input
    human = UserAgent(name="Reviewer")
    review_msg = await human(
        Msg(
            "System",
            "Please review the above response. Type 'ok' to approve, "
            "or enter your revision feedback:",
            "user",
        ),
    )

    review_text = review_msg.get_text_content().strip()

    if review_text.lower() == "ok":
        print("[Review Approved] Response confirmed.")
        return None  # Return None to keep the original output

    # Review rejected: append feedback to the original message and regenerate
    print(f"[Review Rejected] Feedback: {review_text}")
    original_msg = kwargs.get("msg")
    if original_msg is not None:
        # Temporarily remove the hook to avoid infinite loops
        self.clear_instance_hooks("post_reply")

        revised_msg = Msg(
            original_msg.name,
            f"{original_msg.get_text_content()}\n\n"
            f"[Review Feedback] Please revise based on the following "
            f"feedback: {review_text}",
            original_msg.role,
        )
        revised_output = await self.reply(revised_msg)

        # Re-register the hook
        self.register_instance_hook(
            hook_type="post_reply",
            hook_name="human_review",
            hook=human_review_post_reply_hook,
        )

        return revised_output

    return None


async def demo_human_review_hook() -> None:
    """Demonstrate using a post_reply Hook for manual review."""
    agent = create_support_agent()

    # Register the manual review hook
    agent.register_instance_hook(
        hook_type="post_reply",
        hook_name="human_review",
        hook=human_review_post_reply_hook,
    )

    response = await agent(
        Msg(
            "Customer",
            "I bought your product and the quality is terrible. "
            "I demand a refund!",
            "user",
        ),
    )
    print(f"\nFinal Response: {response.get_text_content()}")


asyncio.run(demo_human_review_hook())

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()

async def main() -> None:
    """Run the complete multi-agent customer support system."""
    # Set enable_human_review=True to enable manual review
    system = CustomerSupportSystem(enable_human_review=False)

    # Simulate multiple customer issues
    customer_issues = [
        ("C001", "Your app keeps crashing. I can't use it at all!"),
        ("C002", "Has my order 12345 shipped? When will it arrive?"),
        (
            "C003",
            "I strongly protest! The product quality is terrible. "
            "I demand a refund!",
        ),
    ]

    for customer_id, issue in customer_issues:
        response = await system.handle_customer(customer_id, issue)
        print(f"\n[Final Response]\n{response}")
        print("\n" + "-" * 60)


asyncio.run(main())