工具调用循环

import json

from openai import OpenAI

client = OpenAI()

def get_weather(city: str) -> str:
    return "晴"

def get_clothes(weather: str) -> str:
    return "短袖"

TOOLS = {"get_weather": get_weather, "get_clothes": get_clothes}
SCHEMAS = [...]  # 工具定义，写法见 Function Calling 一章

def run_agent(task: str, max_steps: int = 10) -> str:
    messages = [
        {"role": "system", "content": "你是生活助手，需要时调用工具"},
        {"role": "user", "content": task},
    ]
    for _ in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL, messages=messages, tools=SCHEMAS,
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))

        if not message.tool_calls:
            return message.content

        for call in message.tool_calls:
            function = TOOLS[call.function.name]
            result = function(**json.loads(call.function.arguments))
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })
    raise RuntimeError("超过最大步数")

print(run_agent("今天北京该穿什么？"))  # => 今天晴，穿短袖就行

from collections import Counter

class Budget:
    def __init__(self, max_steps: int, max_tokens: int):
        self.max_steps = max_steps
        self.max_tokens = max_tokens
        self.steps = 0
        self.tokens = 0
        self.calls: Counter[str] = Counter()

    def charge(self, tokens: int) -> None:
        self.steps += 1
        self.tokens += tokens
        if self.steps > self.max_steps:
            raise RuntimeError("超过最大步数")
        if self.tokens > self.max_tokens:
            raise RuntimeError("超过 token 预算")

    def check_repeat(self, name: str, arguments: str, limit: int = 3) -> None:
        self.calls[f"{name}:{arguments}"] += 1
        if self.calls[f"{name}:{arguments}"] > limit:
            raise RuntimeError(f"重复调用 {name}")

budget = Budget(max_steps=5, max_tokens=1000)
budget.charge(400)
budget.charge(400)

try:
    budget.charge(400)
except RuntimeError as e:
    print(e)  # => 超过 token 预算

import json

TOOLS = {"divide": lambda a, b: a / b}

def execute(name: str, arguments: str) -> str:
    if name not in TOOLS:
        return f"错误：没有 {name} 这个工具，可用的有 {list(TOOLS)}"
    try:
        return str(TOOLS[name](**json.loads(arguments)))
    except json.JSONDecodeError:
        return "错误：参数不是合法的 JSON"
    except (TypeError, ValueError, ZeroDivisionError) as e:
        return f"错误：{e}"

execute("divide", '{"a": 6, "b": 3}')  # => '2.0'
execute("divide", '{"a": 6, "b": 0}')  # => '错误：division by zero'
execute("search", "{}")  # => "错误：没有 search 这个工具，可用的有 ['divide']"

from collections.abc import Callable

DANGEROUS = {"send_email", "delete_file"}

def execute_with_approval(
    name: str,
    arguments: dict,
    approve: Callable[[str, dict], bool],
) -> str:
    if name in DANGEROUS and not approve(name, arguments):
        return "用户拒绝了这个操作"
    return f"{name} 已执行"

reject = lambda name, arguments: False
accept = lambda name, arguments: True

execute_with_approval("read_file", {}, reject)  # => 'read_file 已执行'
execute_with_approval("send_email", {}, reject)  # => '用户拒绝了这个操作'
execute_with_approval("send_email", {}, accept)  # => 'send_email 已执行'

async def run_agent(task: str):
    messages = [{"role": "user", "content": task}]
    for _ in range(10):
        response = await client.chat.completions.create(
            model=MODEL, messages=messages, tools=SCHEMAS,
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))

        if not message.tool_calls:
            yield {"type": "answer", "text": message.content}
            return

        for call in message.tool_calls:
            yield {"type": "tool", "name": call.function.name}
            result = await execute(call.function.name, call.function.arguments)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})