综合小项目：用 Python 构建可测试的 AI 任务助手

前面的文章分别介绍了模型调用、对话历史、结构化输出、重试、工具和安全边界。本篇不再引入新的框架，而是把这些基础能力组合成一个小项目：用户输入一段待办描述，模型提取出结构化任务，程序校验结果，遇到高风险操作时要求人工确认。项目刻意保持为命令行程序，重点是看清一条 AI 功能从输入到输出的完整数据流。

先定义项目边界

这个助手只做两件事：把自然语言整理成任务对象，以及在用户确认后把任务保存到本地 JSON 文件。模型不能直接执行任意命令，也不能自行修改文件；写入动作由普通 Python 函数完成，并且必须经过确认。这种边界让“模型负责理解”和“程序负责执行”分开，便于测试和排错。

任务对象采用固定结构：title 是简短标题，priority 只能是 low、normal 或 high，due 是可选日期，needs_confirmation 表示是否需要人工确认。即使模型返回了额外字段，程序也不会把它们当成可执行指令。

创建最小项目

创建虚拟环境并安装本项目需要的依赖。密钥只放在环境变量中，不写入代码：

1
2
3

python -m venv .venv
source .venv/bin/activate
python -m pip install openai python-dotenv

在项目目录创建 .env，内容使用你自己的服务商配置；下面的值只是占位符：

1
2
3

OPENAI_API_KEY=替换为你的真实密钥
MODEL_NAME=替换为你可用的模型名称
OPENAI_BASE_URL=

不要把 .env 提交到 Git。若使用兼容接口，可填写服务商提供的 OPENAI_BASE_URL，程序会从环境变量读取。

编写任务助手

将下面内容保存为 task_assistant.py。代码只依赖标准库、openai 和 python-dotenv，可以直接从命令行运行：

import json
import os
import time
from datetime import date

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL") or None,
)
MODEL = os.environ["MODEL_NAME"]
TASK_FILE = "tasks.json"

SYSTEM_PROMPT = """
你是任务整理助手。把用户描述转换为 JSON 对象，不要输出 Markdown 或解释文字。
字段必须是：title（字符串）、priority（low/normal/high）、due（YYYY-MM-DD 或 null）、
needs_confirmation（布尔值）。涉及删除、支付、发送消息、修改生产数据或不明确的外部操作时，
needs_confirmation 必须为 true。日期不明确时 due 为 null。
""".strip()


def ask_model(text: str) -> str:
    """调用模型；短暂的临时失败最多重试两次。"""
    last_error = None
    for attempt in range(3):
        try:
            response = client.chat.completions.create(
                model=MODEL,
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": text},
                ],
                temperature=0,
            )
            content = response.choices[0].message.content
            if not content:
                raise ValueError("模型返回了空内容")
            return content
        except Exception as exc:
            last_error = exc
            if attempt < 2:
                time.sleep(2 ** attempt)
    raise RuntimeError(f"模型调用失败：{last_error}") from last_error


def validate_task(raw: str) -> dict:
    """解析并严格检查模型输出，拒绝未知字段和非法值。"""
    try:
        task = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError("模型没有返回合法 JSON") from exc

    expected = {"title", "priority", "due", "needs_confirmation"}
    if set(task) != expected:
        raise ValueError("任务字段不完整或包含未知字段")
    if not isinstance(task["title"], str) or not task["title"].strip():
        raise ValueError("title 必须是非空字符串")
    if task["priority"] not in {"low", "normal", "high"}:
        raise ValueError("priority 不是允许的值")
    if task["due"] is not None:
        try:
            date.fromisoformat(task["due"])
        except (TypeError, ValueError) as exc:
            raise ValueError("due 必须是 YYYY-MM-DD 或 null") from exc
    if not isinstance(task["needs_confirmation"], bool):
        raise ValueError("needs_confirmation 必须是布尔值")
    return task


def save_task(task: dict) -> None:
    """只有通过确认后才调用这个副作用函数。"""
    try:
        with open(TASK_FILE, encoding="utf-8") as file:
            tasks = json.load(file)
    except FileNotFoundError:
        tasks = []
    tasks.append(task)
    with open(TASK_FILE, "w", encoding="utf-8") as file:
        json.dump(tasks, file, ensure_ascii=False, indent=2)


def main() -> None:
    text = input("描述一个任务：").strip()
    if not text:
        print("输入不能为空")
        return
    try:
        task = validate_task(ask_model(text))
    except (RuntimeError, ValueError) as exc:
        print(f"处理失败：{exc}")
        return

    print(json.dumps(task, ensure_ascii=False, indent=2))
    if task["needs_confirmation"]:
        answer = input("该任务需要确认，仍然保存吗？[y/N] ").strip().lower()
        if answer != "y":
            print("已取消")
            return
    save_task(task)
    print(f"已保存到 {TASK_FILE}")


if __name__ == "__main__":
    main()

运行命令是：

1	python task_assistant.py

程序的可验证结果不是某一句固定的模型回复，而是三种状态：合法任务经过确认后出现在 tasks.json；用户拒绝确认时不写文件；模型返回非法 JSON 或字段不符合约束时，程序显示错误并结束。你可以分别输入普通待办、涉及删除的描述，以及一条空输入，观察这三条路径。

关键代码为什么这样组织

ask_model 只负责网络请求，不负责解析业务结果。重试使用逐次加倍的等待时间，并限制总次数，避免网络异常时无限调用和重复计费。生产代码还应根据 SDK 提供的异常类型区分超时、限流和认证失败；认证失败通常不应重试。

validate_task 是信任边界。模型输出即使看起来像 JSON，也不能直接写入文件或交给工具执行。先解析，再检查字段集合、类型、枚举值和日期格式，能挡住常见的格式错误。这里使用 set(task) 拒绝未知字段，是为了避免以后某个未审查字段意外变成执行参数。

save_task 是唯一的副作用函数，且只在确认流程之后调用。以后如果把“保存”替换成发邮件、调用支付接口或删除资源，也应保留同样的分层：模型提出结构化意图，程序校验权限和参数，必要时由人批准，最后才执行工具。

常见问题

为什么不让模型直接返回一段自然语言？ 自然语言适合展示，不适合作为程序之间的稳定接口。固定 JSON 让校验、保存和测试都有明确边界；但 JSON 本身不等于可信，所以仍然需要 validate_task。

模型把日期猜错了怎么办？ 提示词要求不明确时返回 null，程序也只校验格式，不能判断事实是否正确。涉及截止日期的业务应让用户确认具体日期，或者把日期解析交给确定性的业务规则。

重试会不会重复执行任务？ 本示例的重试只发生在模型读取阶段，保存动作不在重试循环中，因此不会重复写入同一任务。若未来重试外部工具，必须设计幂等键、查询执行状态，不能简单复制请求。

如何测试而不消耗 API 配额？ 把 ask_model 作为可替换函数，测试时传入固定 JSON 字符串，专门测试合法输出、缺字段、非法优先级和恶意额外字段。模型测试与校验测试分开，失败时才能知道是提示词、接口还是业务代码的问题。

小结

这个小项目把一次 AI 功能拆成“输入—模型理解—结构化解析—严格校验—人工确认—副作用执行”六步。它没有依赖复杂框架，却包含了真实应用中最重要的控制点：密钥隔离、有限重试、输出校验和高风险操作确认。继续扩展时，可以为任务增加数据库、历史会话和评估样例，但每次只新增一个边界，并为它写出可重复的测试。这样，AI 原型才会逐步变成可维护的程序。

搜索时光笔记