Туториал: от простого агента до мультиагентной системы

from openai import OpenAI
import json

client = OpenAI()

TOOLS = [
    {
        "type": "function",
        "name": "get_ticket_status",
        "description": "Вернуть текущий статус тикета поддержки.",
        "parameters": {
            "type": "object",
            "properties": {
                "ticket_id": {"type": "string"}
            },
            "required": ["ticket_id"],
            "additionalProperties": False,
        },
        "strict": True,
    },
    {
        "type": "function",
        "name": "search_policy",
        "description": "Найти внутреннюю policy по указанной теме.",
        "parameters": {
            "type": "object",
            "properties": {
                "topic": {"type": "string"}
            },
            "required": ["topic"],
            "additionalProperties": False,
        },
        "strict": True,
    },
]


def get_ticket_status(ticket_id: str) -> str:
    return f"{ticket_id}: waiting for security review"


def search_policy(topic: str) -> str:
    if "ticket" in topic.lower():
        return "Можно сообщать клиенту статус, но не внутренние комментарии и персональные данные сотрудников."
    return "Policy not found."


TOOL_IMPL = {
    "get_ticket_status": get_ticket_status,
    "search_policy": search_policy,
}


def run_once(user_input: str) -> str:
    response = client.responses.create(
        model="gpt-5.4",
        input=user_input,
        tools=TOOLS,
    )

    outputs = []
    for item in response.output:
        if item.type == "function_call":
            tool_fn = TOOL_IMPL[item.name]
            args = json.loads(item.arguments)
            result = tool_fn(**args)
            outputs.append({
                "type": "function_call_output",
                "call_id": item.call_id,
                "output": result,
            })

    if not outputs:
        return response.output_text

    followup = client.responses.create(
        model="gpt-5.4",
        previous_response_id=response.id,
        input=outputs,
    )
    return followup.output_text


print(run_once("Проверь статус тикета INC-4421 и кратко ответь клиенту."))

from openai import OpenAI

client = OpenAI()

first = client.responses.create(
    model="gpt-5.4",
    input="Пользователь работает над запуском продукта Альфа. Запомни это в рамках текущего потока.",
)

second = client.responses.create(
    model="gpt-5.4",
    previous_response_id=first.id,
    input="Подготовь короткий план задач на неделю.",
)

print(second.output_text)

def generate_answer(task: str) -> str:
    response = client.responses.create(
        model="gpt-5.4",
        input=f"Реши задачу:\n{task}",
    )
    return response.output_text


def evaluate_answer(task: str, answer: str) -> dict:
    response = client.responses.create(
        model="gpt-5.4-mini",
        input=(
            "Проверь ответ по трём критериям: factuality, policy compliance, completeness.\n"
            f"Задача:\n{task}\n\nОтвет:\n{answer}\n\n"
            "Верни JSON вида {pass: bool, feedback: string}."
        ),
        text={"format": {"type": "json_object"}},
    )
    return json.loads(response.output_text)


def generate_with_gate(task: str, max_rounds: int = 2) -> str:
    answer = generate_answer(task)
    for _ in range(max_rounds):
        verdict = evaluate_answer(task, answer)
        if verdict["pass"]:
            return answer
        answer = generate_answer(
            task + "\n\nИсправь предыдущий ответ с учётом feedback:\n" + verdict["feedback"]
        )
    return answer

from agents import Agent, Runner, function_tool


@function_tool
def get_ticket_status(ticket_id: str) -> str:
    return f"{ticket_id}: waiting for security review"


billing_agent = Agent(
    name="Billing specialist",
    handoff_description="Отвечает на вопросы об оплате, счетах и возвратах.",
    instructions="Ты специалист по billing support.",
)


technical_agent = Agent(
    name="Technical specialist",
    handoff_description="Отвечает на технические вопросы по продукту и тикетам.",
    instructions="Ты технический специалист поддержки.",
    tools=[get_ticket_status],
)


triage_agent = Agent(
    name="Triage agent",
    instructions="Маршрутизируй пользователя к правильному специалисту.",
    handoffs=[billing_agent, technical_agent],
)


result = Runner.run_sync(
    triage_agent,
    "Проверь статус тикета INC-4421 и объясни клиенту, что происходит."
)

print(result.final_output)
print(result.last_agent.name)

agent_app/
  app.py
  tools.py
  workflows.py
  evaluators.py
  agents_runtime.py
  traces/