تنسيق الوكلاء: الروتينات والتسليمات

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json
 
client = OpenAI()

system_message = (
    "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 satisfied, offer a refund.
"
    "4. If accepted, search for the ID and then execute refund."
    ""
)

def run_full_turn(system_message, messages):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": system_message}] + messages,
    )
    message = response.choices[0].message
    messages.append(message)
 
    if message.content: print("Assistant:", message.content)
 
    return message

import inspect
 
def function_to_schema(func) -> dict:
    type_map = {
        str: "string",
        int: "integer",
        float: "number",
        bool: "boolean",
        list: "array",
        dict: "object",
        type(None): "null",
    }
 
    try:
        signature = inspect.signature(func)
    except ValueError as e:
        raise ValueError(
            f"Failed to get signature for function {func.__name__}: {str(e)}"
        )
 
    parameters = {}
    for param in signature.parameters.values():
        try:
            param_type = type_map.get(param.annotation, "string")
        except KeyError as e:
            raise KeyError(
                f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
            )
        parameters[param.name] = {"type": param_type}
 
    required = [
        param.name
        for param in signature.parameters.values()
        if param.default == inspect._empty
    ]
 
    return {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": (func.__doc__ or "").strip(),
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": required,
            },
        },
    }

class Agent(BaseModel):
    name: str = "Agent"
    model: str = "gpt-4o-mini"
    instructions: str = "You are a helpful Agent"
    tools: list = []
 
def run_full_turn(agent, messages):
    current_agent = agent
    num_init_messages = len(messages)
    messages = messages.copy()
 
    while True:
        tool_schemas = [function_to_schema(tool) for tool in current_agent.tools]
        tools = {tool.__name__: tool for tool in current_agent.tools}
 
        response = client.chat.completions.create(
            model=agent.model,
            messages=[{"role": "system", "content": current_agent.instructions}]
            + messages,
            tools=tool_schemas or None,
        )
        message = response.choices[0].message
        messages.append(message)
 
        if message.content:  # print agent response
            print(f"{current_agent.name}:", message.content)
 
        if not message.tool_calls:  # if finished handling tool calls, break
            break
 
        for tool_call in message.tool_calls:
            result = execute_tool_call(tool_call, tools, current_agent.name)
 
            if type(result) is Agent:  # if agent transfer, update current agent
                current_agent = result
                result = (
                    f"Transferred to {current_agent.name}. Adopt persona immediately."
                )
 
            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)
 
    return Response(agent=current_agent, messages=messages[num_init_messages:])