从单次调用到可控工作流：用 Python 约束模型执行边界

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
model = os.getenv("MODEL_NAME", "gpt-4o-mini")

ORDERS = {"ORD-1001": "已发货", "ORD-1002": "处理中"}


def get_order_status(order_id: str) -> dict:
    if not isinstance(order_id, str) or not order_id.startswith("ORD-"):
        raise ValueError("订单号格式不正确")
    if len(order_id) > 32:
        raise ValueError("订单号过长")
    return {"order_id": order_id, "status": ORDERS.get(order_id, "不存在")}


TOOLS = [{
    "type": "function",
    "name": "get_order_status",
    "description": "查询订单状态，只能用于用户明确提供订单号的只读查询。",
    "parameters": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
        "additionalProperties": False,
    },
    "strict": True,
}]

DISPATCH = {"get_order_status": get_order_status}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

MAX_STEPS = 3


def run_workflow(user_text: str) -> str:
    items = [{"role": "user", "content": user_text}]

    for step in range(1, MAX_STEPS + 1):
        response = client.responses.create(
            model=model,
            instructions=(
                "你是订单查询助手。只能查询订单，不能修改数据。"
                "如果缺少订单号，直接向用户询问；完成查询后给出简洁答案。"
            ),
            tools=TOOLS,
            input=items,
        )

        calls = [item for item in response.output
                 if item.type == "function_call"]
        if not calls:
            return response.output_text

        outputs = []
        for call in calls:
            function = DISPATCH.get(call.name)
            if function is None:
                result = {"ok": False, "error": "工具不在白名单中"}
            else:
                try:
                    arguments = json.loads(call.arguments)
                    result = {"ok": True, "data": function(**arguments)}
                except (json.JSONDecodeError, TypeError, ValueError) as exc:
                    result = {"ok": False, "error": str(exc)}

            outputs.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

        items = items + response.output + outputs

    raise RuntimeError(f"工作流超过 {MAX_STEPS} 步，已停止")


if __name__ == "__main__":
    print(run_workflow("请查询订单 ORD-1001 的状态。"))