OpenAI SDK 实战：命令行智能助手

一个系列学到收官，最常见的遗憾是"知识都懂，串不起来"：流式会写、工具调用会写、会话裁剪也会写，可它们散在十个脚本里，从没在同一个程序里合过力。本篇是《OpenAI Python SDK 从入门到实战》系列第 11 篇，也是第 1 篇路线图的最后一站——实战项目：从零写一个命令行智能助手，把前十篇的知识点全部装进同一份两百余行的代码。

这个助手能做什么：终端里多轮聊天，回答逐字流出；它知道现在几点、能替你记待办、会算四则运算；关掉终端再打开，对话接得上，账单随时可查。全文按"需求与架构 → 配置 → 工具 → 主循环 → 持久化与成本 → 运行实录 → 知识点对照 → 扩展方向 → 总结"组织，每个模块都标注回扣的前序篇目。

1、需求与总体架构

先把需求钉死再写代码。这个助手面向日常终端使用，五条需求如下：

• 多轮对话：连续问答，指代靠历史解析
• 三个工具：当前时间、待办清单管理、四则计算器
• 流式输出：回答逐字上屏，调用工具时给出提示
• 会话持久化：历史与用量落盘 JSON，重开续聊
• 成本统计：/cost 随时查账，退出时报总账

前三条是体验，后两条是工程——demo 与产品的分界线恰恰在后两条。整份代码按走读顺序分块给出，全部拼进同一个 assistant.py 即可运行。整体架构如下：

解读：整张图分三层。主循环层管一问一答的节奏，输入先过内置命令检查，普通提问追加进历史后发起流式请求；工具路由层是模型与本地函数的交界，模型发起调用、本地执行、结果回传历史，再进下一轮请求，直到模型不再要工具；持久化层在每个回合结束后把历史与用量写进 session.json。三层分别对应本系列第 3、5、7 篇的主线，下文逐块走读。

2、配置与客户端

万事开头还是那三样：密钥、模型、客户端，全部来自第 1 篇的地基。代码如下：

import json
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()                          # 第 1 篇：.env 注入密钥

MODEL = "gpt-6.1-terra"                # 第 2 篇：交互主力档
SESSION_FILE = Path("session.json")
SYSTEM = ("你是终端里的个人助手，回答简洁；"
          "涉及时间、待办、计算时先调用工具再作答。")

client = OpenAI(
    timeout=30.0,    # 第 9 篇：默认 10 分钟必须压下来
    max_retries=2,   # SDK 默认值就是 2，显式写出是主动选择
)

解读三点。其一，timeout=30.0 是本系列第 9 篇的直接结论——SDK 默认超时 10 分钟，交互工具里用户等 30 秒早就走了，先压超时再谈别的事；构造参数的完整清单见官方 Python API 参考。其二，max_retries=2 保持默认：连接错误、429、5xx 由 SDK 带短指数退避自动兜住，第 9 篇的异常地图在这里以"不写代码"的方式生效。其三，系统提示放进 input 数组首位而不是用 instructions 参数，原因有二：落盘的会话文件自包含，读回来直接能发；system 与工具定义构成稳定前缀，对第 8 篇讲的提示缓存最友好。

3、三个工具的定义与实现

工具分两半写：本地函数是真正干活的，schema 声明是给模型看的菜单。先写前两个工具的本地实现：

TODOS = []                             # 待办清单，随会话一并落盘

def get_current_time():
    now = datetime.now()               # 第 5 篇练习题的答案
    return {"datetime": now.strftime("%Y-%m-%d %H:%M"),
            "weekday": "周" + "一二三四五六日"[now.weekday()]}

def manage_todo(action, content=None, todo_id=None):
    if action == "add":
        todo = {"id": len(TODOS) + 1, "content": content, "done": False}
        TODOS.append(todo)
        return {"ok": True, "todo": todo}
    if action == "list":
        return {"count": len(TODOS), "todos": TODOS}
    if action == "done":
        for t in TODOS:
            if t["id"] == todo_id:
                t["done"] = True
                return {"ok": True, "todo": t}
        return {"error": f"找不到编号 {todo_id} 的待办"}
    return {"error": f"未知动作: {action}"}

解读：get_current_time 无参数，正是第 5 篇留的练习题；manage_todo 用一个 action 参数把增、查、完三个动作收进一个工具，返回值统一是字典，回传时 json.dumps 一次即可。待办编号从 1 递增，done 按 id 查找，查不到就回 error——错误也走同一条回传通道，第 5 篇讲过模型看到错误会自行调整话术。再看计算器与按名分发的路由：

def calculate(a, b, op):
    if op == "div" and b == 0:
        return {"error": "除数不能为 0"}
    result = {"add": a + b, "sub": a - b,
              "mul": a * b, "div": a / b}[op]
    return {"result": result}

def call_function(name, args):
    """第 5 篇的路由原样扩展到三个函数"""
    if name == "get_current_time":
        return get_current_time()
    if name == "manage_todo":
        return manage_todo(**args)
    if name == "calculate":
        return calculate(**args)
    raise ValueError(f"未知函数: {name}")

解读：calculate 与第 5 篇实战同款，call_function 按名分发、未知名字抛错，让问题在开发期就暴露。接着把三个函数声明成工具，schema 按 strict 要求写齐：

TOOLS = [
    {
        "type": "function",
        "name": "get_current_time",
        "description": "获取本地当前的日期时间与星期。",
        "parameters": {   # 无参函数：两个键留空即可
            "type": "object", "properties": {}, "required": [],
            "additionalProperties": False,
        },
    },
    {
        "type": "function",
        "name": "manage_todo",
        "description": "管理待办清单：add 新增、list 查看、done 完成。",
        "parameters": {
            "type": "object",
            "properties": {
                "action": {"type": "string",
                           "enum": ["add", "list", "done"]},
                "content": {"type": ["string", "null"],
                            "description": "action 为 add 时填写"},
                "todo_id": {"type": ["integer", "null"],
                            "description": "action 为 done 时填写"},
            },
            "required": ["action", "content", "todo_id"],
            "additionalProperties": False,
        },
    },
]

解读：这里的写法全是第 4、5 篇的规矩——strict 模式要求关闭额外键、字段全部列入 required，可选语义用 ["string", "null"] 这类可空类型模拟；enum 把 action 收窄到三个取值，模型编不出第四种动作；description 是模型的选型依据，"add 新增"那半句就是写给模型看的文档。第三个工具追加进去：

TOOLS.append({
    "type": "function",
    "name": "calculate",
    "description": "四则运算计算器，一次计算一步。",
    "parameters": {
        "type": "object",
        "properties": {
            "a": {"type": "number", "description": "左操作数"},
            "b": {"type": "number", "description": "右操作数"},
            "op": {"type": "string",
                   "enum": ["add", "sub", "mul", "div"]},
        },
        "required": ["a", "b", "op"],
        "additionalProperties": False,
    },
})

解读：注意 description 里"一次计算一步"这句——复合式子会被模型自己拆成连环调用，(128-36)*3.5 就是先减再乘，正好把下一节的工具循环喂饱。工具定义的完整规范可回看官方函数调用指南。

4、主循环：流式与工具循环

主循环是全文的心脏，把第 3 篇的流式消费与第 5 篇的循环骨架焊在一起：

def ask(question, input_list, usage):
    input_list.append({"role": "user", "content": question})
    for _ in range(8):                   # 第 5 篇：上限兜底防打转
        stream = client.responses.create(
            model=MODEL, input=input_list,
            tools=TOOLS, stream=True,    # 第 3 篇：一个参数开流式
        )
        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
            elif (event.type == "response.output_item.added"
                    and event.item.type == "function_call"):
                print(f"\n[调用 {event.item.name}]", flush=True)
            elif event.type == "response.completed":
                resp = event.response    # 完整响应藏在收尾事件里
        add_usage(resp.usage, usage)
        calls = [i for i in resp.output if i.type == "function_call"]
        if not calls:                    # 不再要工具：终局答案
            print()
            return resp.output_text
        input_list += resp.output        # 调用与推理条目原样带回
        for item in calls:
            result = call_function(item.name, json.loads(item.arguments))
            input_list.append({
                "type": "function_call_output",
                "call_id": item.call_id,  # 第 5 篇：call_id 配对
                "output": json.dumps(result, ensure_ascii=False),
            })

解读四点。其一，外层 for 是第 5 篇的循环骨架：模型不再返回 function_call 就是终局，range(8) 防打转。其二，流式与工具循环能无缝叠加，关键在收尾事件——response.completed 携带的 response 与非流式返回同构，所以判断调用、拼接 input 的代码一行不用改；第 3 篇说"有账可查的信息全在收尾事件里"，在这里兑现。其三，output_item.added 拿来实时报工具名，用户能看到"正在调用 calculate"；想连参数也实时拼装，还有第 3 篇预告过的 function_call_arguments.delta。其四，教程从简：收尾也可能是 incomplete 或 failed，生产代码要按第 9 篇给三类收尾各留分支。最后是入口：

def main():
    input_list, usage, todos = load_session()
    TODOS.extend(todos)                  # 待办随会话一并恢复
    print("提问直接输入；/cost 查账 /reset 清空 exit 退出")
    while (q := input("\n你> ").strip()) not in ("exit", "quit", ""):
        if q == "/cost":
            print(report(usage))
        elif q == "/reset":
            input_list, usage, _ = fresh_session()
            TODOS.clear()
            print("已开启新会话")
        else:
            ask(q, input_list, usage)
        save_session(input_list, usage, TODOS)
    print(f"[账单] {report(usage)}")     # 退出时报总账

解读：walrus 表达式把空行与退出命令统一收进循环条件，与第 7 篇的 CLI 脚本同款；每轮结束即落盘，进程意外退出也只丢当前回合。

5、会话持久化与成本统计

会话存取三个函数，围绕 session.json 展开：

def fresh_session():
    return ([{"role": "system", "content": SYSTEM}],
            {"in": 0, "out": 0, "cached": 0}, [])

def load_session():
    if SESSION_FILE.exists():
        d = json.loads(SESSION_FILE.read_text(encoding="utf-8"))
        items = [e if isinstance(e, dict)
                 else e.model_dump(exclude_none=True)
                 for e in d["input_list"]]
        return items, d["usage"], d.get("todos", [])
    return fresh_session()

def save_session(input_list, usage, todos):
    items = [e if isinstance(e, dict)
             else e.model_dump(exclude_none=True) for e in input_list]
    SESSION_FILE.write_text(json.dumps(
        {"input_list": items, "usage": usage, "todos": todos},
        ensure_ascii=False, indent=1), encoding="utf-8")

解读：历史里混着两种条目——对话消息是普通 dict，模型的输出条目是 SDK 对象，落盘前统一 model_dump 成字典；读回后全是 dict，与新一轮请求里新拼的条目混用，服务端都认。为什么选本地拼接而不是第 7 篇的 previous_response_id 托管？状态在自己手里：可审计、可迁移、可随时配 store=False 零留存，这正是第 7 篇的选型结论。

session.json 是完整对话明文，记得把它加进 .gitignore，别随仓库提交。

成本统计直接搬第 8 篇的计费函数：

PRICE = {  # 每百万 token：输入、输出、缓存读取
    "gpt-6.1-terra": (1.0, 5.0, 0.05),
    "gpt-6-luna": (0.1, 0.5, 0.01),
}

def add_usage(u, usage):
    usage["in"] += u.input_tokens
    usage["out"] += u.output_tokens
    usage["cached"] += u.input_tokens_details.cached_tokens

def report(usage, model=MODEL):
    p_in, p_out, p_cached = PRICE[model]
    fresh = usage["in"] - usage["cached"]
    dollars = (fresh * p_in + usage["cached"] * p_cached
               + usage["out"] * p_out) / 1e6
    return (f"输入 {usage['in']}（缓存命中 {usage['cached']}）"
            f"输出 {usage['out']} tokens，约 ${dollars:.4f}")

解读：usage 三件套加 cached_tokens 是第 7 篇讲透的口径；计费逻辑与第 8 篇一字不差——未命中输入按常规价、缓存读取按 0.05 倍、输出按输出价，单价以官方定价页为准。价格表只列两款：助手跑在 Terra 上，练习时换 Luna 对照账单刚好够用。

6、运行实录

两次启动的运行实录如下（第二次启动摘录）：

$ python assistant.py
提问直接输入；/cost 查账 /reset 清空 exit 退出

你> 现在几点了？顺便把"复习 SDK 第 11 篇"加进待办
[调用 get_current_time]
[调用 manage_todo]
现在是 2026-10-31 21:40（周六）。已把"复习 SDK 第 11 篇"记入待办，编号 1。

你> 算一下 (128-36)*3.5 再减 15
[调用 calculate]
[调用 calculate]
[调用 calculate]
分步算完：128-36=92，92*3.5=322，322-15=307，结果是 307。

你> exit
[账单] 输入 2360（缓存命中 1780）输出 312 tokens，约 $0.0022

$ python assistant.py
你> 我上轮让你记的待办是什么？
[调用 manage_todo]
你记的待办是"复习 SDK 第 11 篇"，编号 1，还未完成。

解读四处看点。第一问里模型并行发起两个调用——时间与待办互不依赖，第 5 篇说的并行调用在这里现形。第二问里三个 calculate 串行连环，后一步要用前一步的结果，工具循环整整转了三圈。重启之后，历史与待办都从盘上读回，"上轮"的指代照常解析，持久化闭环。运行体感也与第 3 篇的分析对得上：提问后不到一秒文字就开始流出，工具提示行先于正文出现，等待感被压到最低。账单里的缓存命中 1780 同样有讲究：system 与工具定义是稳定前缀，第二轮起自动命中，第 8 篇的隐式缓存零成本生效——一次完整会话总成本两美厘量级，放心用。

7、知识点对照表

这份助手没有引入任何新 API，全部零件来自前十篇。对应关系如下：

表里最后两行是本篇未直接动用的能力，恰好构成下一节的扩展方向——十篇没有一篇白学。

8、可扩展方向

拿这份脚本继续练手，三个方向各有对应篇目：

• 换模型档：把 MODEL 改成 gpt-6-luna 跑同样对话，按第 8 篇的定档流程对比质量与账单
• 加记忆摘要：历史过长时按第 7 篇把旧轮交给 gpt-6-luna 压成摘要，钉住单次输入
• 接向量存储：按第 10 篇把笔记归集进 vector store，再声明一个 file_search 工具

三个方向分别动单价、动上下文、动知识面，第 1 篇路线图到这里闭合成环。

9、总结

一个文件、五个模块、两百余行代码，命令行智能助手从需求落到了成品：配置层压好超时与重试，三个工具按 strict 规范声明，主循环把流式与工具调用焊成一体，会话与账单随每轮落盘。写它没用任何新 API，用到的每个机制——事件流、call_id、可空类型、cached_tokens——都是前十篇逐步攒下的零件，收官只是把它们装回一台完整的机器。系列到这里正好收束：从第一次调用到完整的助手，Responses API 的主线始终没变——拼好 input、读懂 output、管住 usage，其余一切都是这个循环的变奏。对于想继续深入的读者，建议把这份脚本当模板改成自己的日常工具，从加第四个工具开始动手。

留一道练习：给助手加天气工具 get_weather，先用北京、上海、广州三城市的假数据字典顶上，description 写明仅支持这三城；然后问"深圳今天多少度"，观察模型拿到 error 结果后如何向用户转告——第 5 篇讲的错误通道在这里复现。想加难度，再把假数据换成真实天气 API 的返回结构。