AI Agent 與工作流自動化

import json
import os
from openai import OpenAI

client = OpenAI(base_url="https://www.tutujin.com/v1", api_key=os.environ["TUTUJIN_API_KEY"])
audit_log, approval_queue = [], []

TICKET = {"name": "ticket", "strict": True, "schema": {
    "type": "object",
    "properties": {
        "category": {"type": "string", "enum": ["refund", "shipping", "invoice", "other"]},
        "order_id": {"type": ["string", "null"]},
        "urgent": {"type": "boolean"},
    },
    "required": ["category", "order_id", "urgent"],
    "additionalProperties": False,
}}


def classify(text):
    """Small model: free text in, fixed fields out. Code decides the workflow."""
    resp = client.chat.completions.create(
        model="gpt-5.4-nano",
        messages=[{"role": "system", "content": "Extract the ticket fields."}, {"role": "user", "content": text}],
        # models without json_schema support: {"type": "json_object"}, then check the fields yourself
        response_format={"type": "json_schema", "json_schema": TICKET},
    )
    return json.loads(resp.choices[0].message.content)


TOOLS = [
    {"type": "function", "function": {
        "name": "get_order", "description": "Look up an order. Read-only.",
        "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]},
    }},
    {"type": "function", "function": {
        "name": "refund", "description": "Refund an order. Runs only after a person approves it.",
        "parameters": {"type": "object", "properties": {
            "order_id": {"type": "string"}, "amount": {"type": "number"}, "reason": {"type": "string"}},
            "required": ["order_id", "amount", "reason"]},
    }},
]


def run_tool(name, args):
    if name == "get_order":
        return {"order_id": args["order_id"], "status": "delivered", "paid": 129.0}  # call your order API here
    if name == "refund":
        approval_queue.append(args)  # side effect: a person approves, then your code executes it
        return {"status": "waiting_for_approval"}
    return {"error": "unknown tool"}


def run_agent(task, max_steps=6):
    messages = [
        {"role": "system", "content": "You handle after-sales requests with the tools. Never promise a refund before it is approved."},
        {"role": "user", "content": task},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(model="gpt-5.4-mini", messages=messages, tools=TOOLS)
        msg = resp.choices[0].message
        audit_log.append({"step": step, "model": resp.model, "usage": resp.usage.model_dump() if resp.usage else None})
        if not msg.tool_calls:
            return msg.content
        messages.append({"role": "assistant", "content": msg.content, "tool_calls": [c.model_dump() for c in msg.tool_calls]})
        for call in msg.tool_calls:
            try:
                args = json.loads(call.function.arguments or "{}")
                result = run_tool(call.function.name, args)
            except (json.JSONDecodeError, KeyError) as e:
                args, result = call.function.arguments, {"error": "bad arguments: " + str(e)}
            audit_log.append({"step": step, "tool": call.function.name, "args": args, "result": result})
            messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)})
    return "Step limit reached, handing over to a person."


text = "Order 10086 arrived broken. I want my money back."
if classify(text)["category"] == "refund":
    print(run_agent(text))
    print("waiting for approval:", approval_queue)