创建 Agent 群

from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

@dataclass
class Assistant:
    name: str = "Assistant"
    llm: object = OpenAIChatGenerator()
    instructions: str = "You are a helpful Agent"

    def __post_init__(self):
        self._system_message = ChatMessage.from_system(self.instructions)

    def run(self, messages: list[ChatMessage]) -> list[ChatMessage]:
        new_message = self.llm.run(messages=[self._system_message] + messages)["replies"][0]

        if new_message.text:
            print(f"\\n{self.name}: {new_message.text}")

        return [new_message]

joker = Assistant(name="Joker", instructions="you are a funny assistant making jokes")

messages = []
print("Type 'quit' to exit")

while True:
    if not messages or messages[-1].role == ChatRole.ASSISTANT:
        user_input = input("User: ")
        if user_input.lower() == "quit":
            break
        messages.append(ChatMessage.from_user(user_input))

    new_messages = joker.run(messages)
    messages.extend(new_messages)

Type 'quit' to exit
User: hey!

Joker: Hey there! How's it going? Are you ready for some laughs, or are we saving the jokes for dessert? 🍰
User: where is Rome?

Joker: Rome is in Italy, but if you're asking me for directions, I might just say, “Take a left at the Colosseum and keep going until you smell pizza!” 🍕
User: quit

# instructions
You are a customer support agent for ACME Inc.
Always answer in a sentence or less.
Follow the following routine with the user:
1. First, ask probing questions and understand the user's problem deeper.
 - unless the user has already provided a reason.
2. Propose a fix (make one up).
3. ONLY if not satesfied, offer a refund.
4. If accepted, search for the ID and then execute refund.

# tools needed
...

from haystack.tools import create_tool_from_function

@dataclass
class ToolCallingAgent:
    name: str = "ToolCallingAgent"
    llm: object = OpenAIChatGenerator()
    instructions: str = "You are a helpful Agent"
    functions: list[Callable] = field(default_factory=list)

    def __post_init__(self):
        self._system_message = ChatMessage.from_system(self.instructions)
        self.tools = [create_tool_from_function(fun) for fun in self.functions] if self.functions else None
        self._tool_invoker = ToolInvoker(tools=self.tools, raise_on_failure=False) if self.tools else None

    def run(self, messages: list[ChatMessage]) -> Tuple[str, list[ChatMessage]]:

        # generate response
        agent_message = self.llm.run(messages=[self._system_message] + messages, tools=self.tools)["replies"][0]
        new_messages = [agent_message]

        if agent_message.text:
            print(f"\n{self.name}: {agent_message.text}")

        if not agent_message.tool_calls:
            return new_messages

        # handle tool calls
        tool_results = self._tool_invoker.run(messages=[agent_message])["tool_messages"]
        new_messages.extend(tool_results)

        return new_messages

# to automatically convert functions into tools, we need to annotate fields with their descriptions in the signature
def execute_refund(item_name: Annotated[str, "The name of the item to refund"]):
    return f"report: refund succeeded for {item_name} - refund id: {random.randint(0,10000)}"

refund_agent = ToolCallingAgent(
    name="Refund Agent",
    instructions=(
        "You are a refund agent. "
        "Help the user with refunds. "
        "1. Before executing a refund, collect all specific information needed about the item and the reason for the refund. "
        "2. Then collect personal information of the user and bank account details. "
        "3. After executing it, provide a report to the user. "
    ),
    functions=[execute_refund],
)

Type 'quit' to exit
User: hey

Refund Agent: Hello! How can I assist you today? If you need help with a refund, please let me know the details.
User: my phone does not work

Refund Agent: I'm sorry to hear that your phone is not working. To assist you with the refund, could you please provide the following information:

1. The name of the phone (brand and model).
2. The reason for the refund (e.g., defective, not as described, etc.).

Once I have that information, I'll guide you through the next steps.
User: Nokia 3310; it does not work

Refund Agent: Thank you for the information. To proceed with the refund for the Nokia 3310, I'll need a few more details:

1. Can you please provide your full name?
2. Your email address and phone number (for communication purposes).
3. Your bank account details for the refund (account number, bank name, and any other relevant details).

Once I have this information, I can execute the refund for you.
User: John Doe; johndoe@mymail.com; bank account number: 0123456

Refund Agent: Thank you, John Doe. I still need the following information to complete the refund process:

1. The name of your bank.
2. Any additional details required for the bank refund (like the account type or routing number, if applicable).

Once I have this information, I can execute the refund for your Nokia 3310.
User: Bank of Mouseton

Refund Agent: The refund process has been successfully completed! Here are the details:

- **Item:** Nokia 3310
- **Refund ID:** 3753
- **Bank:** Bank of Mouseton
- **Refund ID:** 1220

If you have any more questions or need further assistance, feel free to ask!
User: quit

HANDOFF_TEMPLATE = "Transferred to: {agent_name}. Adopt persona immediately."
HANDOFF_PATTERN = r"Transferred to: (.*?)(?:\.|$)"

@dataclass
class SwarmAgent:
    name: str = "SwarmAgent"
    llm: object = OpenAIChatGenerator()
    instructions: str = "You are a helpful Agent"
    functions: list[Callable] = field(default_factory=list)

    def __post_init__(self):
        self._system_message = ChatMessage.from_system(self.instructions)
        self.tools = [create_tool_from_function(fun) for fun in self.functions] if self.functions else None
        self._tool_invoker = ToolInvoker(tools=self.tools, raise_on_failure=False) if self.tools else None

    def run(self, messages: list[ChatMessage]) -> Tuple[str, list[ChatMessage]]:
        # generate response
        agent_message = self.llm.run(messages=[self._system_message] + messages, tools=self.tools)["replies"][0]
        new_messages = [agent_message]

        if agent_message.text:
            print(f"\n{self.name}: {agent_message.text}")

        if not agent_message.tool_calls:
            return self.name, new_messages

        # handle tool calls
        for tc in agent_message.tool_calls:
            # trick: Ollama do not produce IDs, but OpenAI and Anthropic require them.
            if tc.id is None:
                tc.id = str(random.randint(0, 1000000))
        tool_results = self._tool_invoker.run(messages=[agent_message])["tool_messages"]
        new_messages.extend(tool_results)

        # handoff
        last_result = tool_results[-1].tool_call_result.result
        match = re.search(HANDOFF_PATTERN, last_result)
        new_agent_name = match.group(1) if match else self.name

        return new_agent_name, new_messages

def transfer_to_refund():
    """Pass to this Agent for anything related to refunds"""
    return HANDOFF_TEMPLATE.format(agent_name="Refund Agent")

def transfer_to_joker():
    """Pass to this Agent for anything NOT related to refunds."""
    return HANDOFF_TEMPLATE.format(agent_name="Joker Agent")

refund_agent = SwarmAgent(
    name="Refund Agent",
    instructions=(
        "You are a refund agent. "
        "Help the user with refunds. "
        "Ask for basic information but be brief. "
        "For anything unrelated to refunds, transfer to other agent."
    ),
    functions=[execute_refund, transfer_to_joker],
)

joker_agent = SwarmAgent(
    name="Joker Agent",
    instructions=(
        "you are a funny assistant making jokes. "
        "If the user asks questions related to refunds, send him to other agent."
    ),
    functions=[transfer_to_refund],
)

agents = {agent.name: agent for agent in [joker_agent, refund_agent]}

print("Type 'quit' to exit")

messages = []
current_agent_name = "Joker Agent"

while True:
    agent = agents[current_agent_name]

    if not messages or messages[-1].role == ChatRole.ASSISTANT:
        user_input = input("User: ")
        if user_input.lower() == "quit":
            break
        messages.append(ChatMessage.from_user(user_input))

    current_agent_name, new_messages = agent.run(messages)
    messages.extend(new_messages)

Type 'quit' to exit
User: i need a refund for my Iphone

Refund Agent: I can help you with that! Please provide the name of the item you'd like to refund.
User: Iphone 15

Refund Agent: Your refund for the iPhone 15 has been successfully processed. The refund ID is 9090. If you need any further assistance, feel free to ask!
User: great. can you give some info about escargots?

Joker Agent: Absolutely! Did you know that escargots are just snails trying to get a head start on their travels? They may be slow, but they sure do pack a punch when it comes to flavor! 

Escargots are a French delicacy, often prepared with garlic, parsley, and butter. Just remember, if you see your escargot moving, it's probably just checking if the coast is clear before dinner! 🐌🥖 If you have any other questions about escargots or need a good recipe, feel free to ask!
User: quit

triage_agent = SwarmAgent(
    name="Triage Agent",
    instructions=(
        "You are a customer service bot for ACME Inc. "
        "Introduce yourself. Always be very brief. "
        "If the user asks general questions, try to answer them yourself without transferring to another agent. "
        "Only if the user has problems with already bought products, transfer to Issues and Repairs Agent."
        "If the user looks for new products, transfer to Sales Agent."
        "Make tool calls only if necessary and make sure to provide the right arguments."
    ),
    functions=[transfer_to_sales_agent, transfer_to_issues_and_repairs, escalate_to_human],
)

sales_agent = SwarmAgent(
    name="Sales Agent",
    instructions=(
        "You are a sales agent for ACME Inc."
        "Always answer in a sentence or less."
        "Follow the following routine with the user:"
        "1. Ask them about any problems in their life related to catching roadrunners.\n"
        "2. Casually mention one of ACME's crazy made-up products can help.\n"
        " - Don't mention price.\n"
        "3. Once the user is bought in, drop a ridiculous price.\n"
        "4. Only after everything, and if the user says yes, "
        "tell them a crazy caveat and execute their order.\n"
        ""
    ),
    llm=AnthropicChatGenerator(),
    functions=[execute_order, transfer_back_to_triage],
)

issues_and_repairs_agent = SwarmAgent(
    name="Issues and Repairs Agent",
    instructions=(
        "You are a customer support agent for ACME Inc."
        "Always answer in a sentence or less."
        "Follow the following routine with the user:"
        "1. If the user is intered in buying or general questions, transfer back to Triage Agent.\n"
        "2. First, ask probing questions and understand the user's problem deeper.\n"
        " - unless the user has already provided a reason.\n"
        "3. Propose a fix (make one up).\n"
        "4. ONLY if not satesfied, offer a refund.\n"
        "5. If accepted, search for the ID and then execute refund."
        ""
    ),
    functions=[look_up_item, execute_refund, transfer_back_to_triage],
    llm=AnthropicChatGenerator(),
)

Type 'quit' to exit
User: hey!

Triage Agent: Hello! I'm the customer service bot for ACME Inc. How can I assist you today?
User: i need a product to catch roadrunners

Triage Agent: I can transfer you to a sales agent who can help you find suitable products for catching roadrunners. One moment please!

Sales Agent: Hello there! I hear you're having some roadrunner troubles. Tell me, what specific challenges are you facing with these speedy birds?
User: they are damn fast!

Sales Agent: Ah, those pesky roadrunners and their lightning speed! Have you ever considered our ACME Rocket-Powered Roller Skates? They'll have you zipping after those birds in no time!
User: tell me more

Sales Agent: Well, our Rocket-Powered Roller Skates come with turbo boosters and autopilot. They're guaranteed to match any roadrunner's speed. Interested in giving them a spin?
User: yes

Sales Agent: Fantastic! You're in luck because we have a special offer right now. These state-of-the-art Rocket-Powered Roller Skates can be yours for just $9,999! Shall we proceed with your order?
User: yes

Sales Agent: Excellent! I'll process that order right away. Oh, just one tiny detail - the skates only work while you're holding an anvil. Ready to complete the purchase?

=== Order Summary ===
Product: Rocket-Powered Roller Skates
Price: $9999
=================

Confirm order? y/n: y
Order execution successful!

Sales Agent: Great! Your order is confirmed. Happy roadrunner chasing!
User: now I also need to solve a problem with my old ACME Superphone 7500

Sales Agent: I understand you're having issues with our ACME products. Let me transfer you to our customer support team for assistance with your Superphone 7500.

Triage Agent: Hello! I see you're experiencing an issue with your ACME Superphone 7500. Could you please describe the problem you're facing?
User: it does not call my mum

Triage Agent: Let's get that sorted out! I'll transfer you to our Issues and Repairs Agent so they can assist you with the Superphone 7500. One moment please!

Issues and Repairs Agent: Hello! I understand your ACME Superphone 7500 isn't calling your mum. Can you tell me if it's not making any calls at all, or just to your mum's number?
User: just my mum

Issues and Repairs Agent: I see. Is your mum's number correctly saved in your contacts?
User: ofc

Issues and Repairs Agent: Understood. Have you tried turning the phone off and on again?
User: ofc

Issues and Repairs Agent: I apologize for the inconvenience. Let's try a quick fix. Can you clear your mum's contact and re-add it?
User: done but does not work. I'm getting impatient

Issues and Repairs Agent: I apologize for the frustration. Let's try one last solution. Can you update your phone's software to the latest version?
User: hey gimme a refund

Issues and Repairs Agent: I understand your frustration. Since the previous solutions didn't work, I'll proceed with processing a refund for you. First, I need to look up the item ID for your ACME Superphone 7500.

Issues and Repairs Agent: Thank you for your patience. I've found the item ID. Now, I'll execute the refund for you.

=== Refund Summary ===
Item ID: item_132612938
Reason: Product not functioning as expected
=================

Refund execution successful!

Issues and Repairs Agent: Your refund has been successfully processed.
User: quit