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

前面的文章已经介绍了工具调用：模型可以提出工具请求，应用负责校验、执行并回传结果。但如果把“请求模型—执行工具—再次请求模型”直接写成没有边界的循环，程序就可能重复调用、无限运行，甚至在不该执行时执行有副作用的操作。本篇只聚焦一个核心知识点：如何把单次模型调用组织成一个有明确状态和停止条件的可控工作流。示例使用一个只读的订单查询工具，不涉及真实业务写入。

工作流不是“让模型自己循环”

单次调用的结构通常是：准备输入，调用模型，读取文本。工作流则把一次任务拆成可观察的步骤：

准备（plan）：固定系统规则和用户目标，设置允许使用的工具。
执行（act）：只执行应用白名单中的工具，并校验模型给出的参数。
观察（observe）：把脱敏后的工具结果回传给模型。
结束（finish）：模型给出最终文本，或程序因为达到上限、出错、需要确认而停止。

关键点是，模型只负责提出下一步建议，Python 程序才拥有循环控制权。工作流状态至少应包括当前输入、已执行步数和是否已经得到最终答案。将这些状态显式写在代码中，比依赖提示词中的“请不要无限调用”可靠得多。

先定义边界和工具

下面的工具使用本地字典模拟数据。真实项目可以换成数据库或内部服务，但工具函数仍应自行检查参数，不能因为请求来自模型就跳过校验。

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}

TOOLS 是给模型看的能力说明，DISPATCH 才是应用真正允许执行的集合。两者都由程序维护。尤其不要根据模型返回的名称动态导入模块、拼接 SQL，或使用 eval 执行字符串。工具是只读查询时，工作流可以自动运行；涉及付款、删除、发送等副作用时，应在执行前转入人工确认状态。

把一次工具往返封装成有限循环

下面的函数实现一个最小工作流。每轮先请求模型，遇到普通文本就结束；遇到工具调用就执行并把结果回传。MAX_STEPS 是硬上限，即使模型持续请求工具，程序也会停止。代码只允许一个工具，便于看清状态变化；以后增加工具时仍应沿用白名单分发。

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 的状态。"))

每次循环都产生一个清晰的状态迁移：items 保存到目前为止的输入和工具结果，response.output 表示模型本轮的决定，outputs 表示程序实际完成的动作。最后一次请求得到文本后返回；如果模型只返回工具调用，下一轮会让它看到结果并组织答案。工具结果用 ok 区分成功与可预期失败，模型便能告诉用户订单不存在或订单号格式有误，而不是让异常直接中断整个进程。

三个必须保留的控制点

第一是步数上限。 上限不是用来修复模型逻辑的，而是最后一道保险。生产环境还可以同时设置总耗时、单次工具调用数和响应大小上限。超限时记录任务标识和原因，向用户返回可理解的失败提示。

第二是工具白名单和权限。 模型请求的工具名、参数都属于不可信输入。分发前检查名称、参数类型、字段长度和当前用户权限；错误信息要脱敏，不能把密钥、内部地址或完整异常堆栈传回模型。用户身份应来自应用会话，而不是模型生成的参数。

第三是可恢复的状态。 示例把状态放在内存中，适合一次任务。需要跨请求恢复时，可保存步骤号、输入摘要、工具调用标识和结果，但不要无限保存完整对话。恢复任务前还要重新检查权限和工具版本，避免旧状态触发已经撤销的能力。

常见问题

为什么不把所有控制规则写进 system 提示？ 提示词能指导模型，却不能阻止程序执行未知工具，也不能可靠地限制循环次数。安全边界必须落在应用代码中。

达到步数上限时要不要继续请求模型？ 不要。上限意味着程序无法在预算内完成任务，应停止并记录原因；盲目增加次数会掩盖工具错误和提示词问题。

工具失败后是否立刻结束？ 对于可预期的业务失败，可以将结构化错误回传，让模型澄清或解释；认证失败、网络异常等基础设施错误则应按错误处理策略重试或终止，不应无限重试。

为什么不直接使用一个“万能工具”？ 工具边界越宽，参数校验、权限审计和测试越困难。按一个明确动作拆分工具，工作流才容易观察和验证。

小结

可控工作流的核心不是增加更多模型调用，而是把调用放进应用掌控的状态机：模型提出计划，白名单决定动作，程序记录观察结果，有限循环负责结束。先用只读工具验证每个状态迁移，再逐步加入超时、日志、人工审批和持久化，才能把单次问答稳妥地扩展为可维护的 AI 应用。

搜索时光笔记