智能体构建实用指南

1  weather_agent = Agent(
2      name="Weather agent",
3      instructions="You are a helpful agent who can talk to users about the weather.",
4      tools=[get_weather],
5  )

1   from agents import Agent, WebSearchTool, function_tool
2
3   @function_tool
4   def save_results(output):
5       db.insert({"output": output, "timestamp": datetime.time()})
6       return "File saved"
7
8   search_agent = Agent(
9       name="Search agent",
10      instructions="Help the user search the internet and save results if asked.",
11      tools=[WebSearchTool(), save_results],
12  )

"You are an expert in writing instructions for an LLM agent. Convert the
following help center document into a clear set of instructions, written
in a numbered list. The document will be a policy followed by an LLM.
Ensure that there is no ambiguity, and that the instructions are written
as directions for an agent. The help center document to convert is the
following {{help_center_doc}}"

Agents.run(agent, [UserMessage("What's the capital of the USA?")])

""" You are a call center agent. You are interacting with
{{user_first_name}} who has been a member for {{user_tenure}}.
The user's most common complaints are about {{user_complaint_categories}}.
Greet the user, thank them for being a loyal customer, and answer
any questions the user may have! """

from agents import Agent, Runner

manager_agent = Agent(
    name="manager_agent",
    instructions=(
        "You are a translation agent. You use the tools given to you to translate."
        "If asked for multiple translations, you call the relevant tools."
    ),
    tools=[
        spanish_agent.as_tool(
            tool_name="translate_to_spanish",
            tool_description="Translate the user's message to Spanish",
        ),
        french_agent.as_tool(
            tool_name="translate_to_french",
            tool_description="Translate the user's message to French",
        ),
        italian_agent.as_tool(
            tool_name="translate_to_italian",
            tool_description="Translate the user's message to Italian",
        ),
    ],
)

async def main():
    msg = input("Translate 'hello' to Spanish, French and Italian for me!")

    orchestrator_output = await Runner.run(manager_agent, msg)

    for message in orchestrator_output.new_messages:
        print(f"  - Translation step: {message.content}")

from agents import Agent, Runner

technical_support_agent = Agent(
    name="Technical Support Agent",
    instructions="You provide expert assistance with resolving technical issues,
                  system outages, or product troubleshooting.",
    tools=[search_knowledge_base]
)

sales_assistant_agent = Agent(
    name="Sales Assistant Agent",
    instructions="You help enterprise clients browse the product catalog,
                  recommend suitable solutions, and facilitate purchase transactions.",
    tools=[initiate_purchase_order]
)

order_management_agent = Agent(
    name="Order Management Agent",
    instructions="You assist clients with inquiries regarding order tracking,
                  delivery schedules, and processing returns or refunds.",
    tools=[track_order_status, initiate_refund_process]
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="You act as the first point of contact, assessing customer
                  queries and directing them promptly to the correct specialized agent.",
    handoffs=[technical_support_agent, sales_assistant_agent, order_management_agent],
)

await Runner.run(
    triage_agent,
    input("Could you please provide an update on the delivery timeline
          for our recent purchase?")
)

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
    Guardrail,
    GuardrailTripwireTriggered
)
from pydantic import BaseModel

class ChurnDetectionOutput(BaseModel):
    is_churn_risk: bool
    reasoning: str

churn_detection_agent = Agent(
    name="Churn Detection Agent",
    instructions="Identify if the user message indicates a potential customer churn risk.",
    output_type=ChurnDetectionOutput,
)

@input_guardrail
async def churn_detection_tripwire(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(churn_detection_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_churn_risk,
    )

customer_support_agent = Agent(
    name="Customer support agent",
    instructions="You are a customer support agent. You help customers with their questions.",
    input_guardrails=[
        Guardrail(guardrail_function=churn_detection_tripwire),
    ],
)

async def main():
    # This should be ok
    await Runner.run(customer_support_agent, "Hello!")
    print("Hello message passed")

    # This should trip the guardrail
    try:
        await Runner.run(agent, "I think I might cancel my subscription")
        print("Guardrail didn't trip - this is unexpected")
    except GuardrailTripwireTriggered:
        print("Churn detection guardrail tripped")