GPTs 与 Assistant API：托管式 Agent 的现状与迁移

"""retrykit.py —— 托管式接口必备的两件事：轮询退避与失败重试。

托管式接口把「跑一次」做成了异步任务：提交 Run 之后立刻返回，
状态要自己轮询。写死 time.sleep(2) 的做法有两个毛病：
  · 任务很快就结束时，白白等满 2 秒；
  · 任务要跑一分钟时，30 次请求全打在服务端上。
指数退避 + 上限 + 总超时，是这类接口的标准写法。
"""
import random
import time


def backoff_delays(base=0.5, factor=1.8, cap=8.0, jitter=0.25):
    """生成一串逐渐变长的等待秒数：0.5 → 0.9 → 1.6 → 2.9 → … → 封顶 8 秒。

    jitter 是抖动比例，避免多个进程在同一毫秒齐刷刷地重试（惊群）。
    这是一个无限生成器，由调用方决定什么时候停。
    """
    delay = base
    while True:
        shake = delay * jitter * (random.random() * 2 - 1)
        yield max(0.05, delay + shake)
        delay = min(cap, delay * factor)


def poll_until(fetch_status, terminal, timeout=180.0, on_tick=None, **kw):
    """反复调用 fetch_status()，直到状态进入 terminal 集合或超时。

    参数
      fetch_status  无参函数，返回当前状态字符串
      terminal      终态集合，例如 {"completed", "failed", "cancelled", "expired"}
      timeout       总超时秒数；超时抛 TimeoutError，而不是无限循环
      on_tick       每轮回调，签名 on_tick(轮次, 状态, 已用秒数)
    返回
      最终状态字符串
    """
    started = time.time()
    delays = backoff_delays(**kw)
    turn = 0
    while True:
        turn += 1
        status = fetch_status()
        used = time.time() - started
        if on_tick:
            on_tick(turn, status, used)
        if status in terminal:
            return status
        if used >= timeout:
            raise TimeoutError(
                "轮询超时：%.1f 秒后状态仍为 %r（已轮询 %d 次）。"
                % (used, status, turn)
            )
        time.sleep(next(delays))


def retry(call, times=3, retry_on=(), on_fail=None, **kw):
    """把一次网络调用包成「失败自动重试」的版本。

    retry_on 传异常类型元组；网关 502、连接重置这类瞬时故障才该重试。
    参数校验错、鉴权错重试多少次都是同样的结果，不要浪费配额。
    """
    delays = backoff_delays(**kw)
    last = None
    for i in range(1, times + 1):
        try:
            return call()
        except retry_on as exc:          # noqa: B014 —— 由调用方指定类型
            last = exc
            if on_fail:
                on_fail(i, exc)
            if i == times:
                break
            time.sleep(next(delays))
    raise last


if __name__ == "__main__":
    # 无网络自检：用一个假状态机验证退避与终态判定是否正确。
    seq = ["queued", "queued", "in_progress", "in_progress", "completed"]
    box = {"i": 0}

    def fake():
        s = seq[min(box["i"], len(seq) - 1)]
        box["i"] += 1
        return s

    final = poll_until(
        fake,
        terminal={"completed", "failed"},
        timeout=30,
        base=0.05, cap=0.2,
        on_tick=lambda n, s, t: print("第 %d 轮 状态=%s 已用 %.2fs" % (n, s, t)),
    )
    print("终态:", final)

"""run_state_machine.py —— 把 Run 的状态机跑一遍（不联网，可直接执行）。

托管式接口最容易被忽略的一点：Run 不是「发出去就有结果」，
它是一个有限状态机，而且有一个分支必须你来接手：

    queued ──► in_progress ──► completed          正常结束
                   │
                   ├──► requires_action ──► （你执行本地函数并提交输出）──► in_progress
                   │
                   ├──► failed / cancelled / expired               异常结束

把 requires_action 当成终态，是新手最常见的错误：现象是
「明明状态查到了，消息列表却永远是空的」。

这份脚本用一个假后端把四条路径都演一遍，方便对照自己的轮询逻辑。

    python3 run_state_machine.py
"""
import json

from retrykit import poll_until

TERMINAL = {"completed", "failed", "cancelled", "expired"}
NEEDS_YOU = "requires_action"


class FakeRun:
    """假的平台 Run：按预设脚本吐状态，并在 requires_action 时索要工具输出。"""

    def __init__(self, script):
        self.script = list(script)
        self.i = 0
        self.submitted = []

    def status(self):
        s = self.script[min(self.i, len(self.script) - 1)]
        self.i += 1
        return s

    def required_calls(self):
        return [{"id": "call_1", "name": "lookup_fruit",
                 "arguments": json.dumps({"name": "葡萄"}, ensure_ascii=False)}]

    def submit_tool_outputs(self, outputs):
        """提交之后平台会把 Run 推回 in_progress，继续跑。"""
        self.submitted.extend(outputs)
        self.script = ["in_progress", "completed"]
        self.i = 0


def lookup_fruit(name):
    table = {"葡萄": {"cost": 2.0, "price": 4.0}}
    return table.get(name, {"found": False})


REGISTRY = {"lookup_fruit": lookup_fruit}


def drive(run, label):
    """完整的驱动循环：轮询 → 遇到 requires_action 就执行并提交 → 继续轮询。"""
    print("== %s ==" % label)
    while True:
        status = poll_until(
            run.status,
            terminal=TERMINAL | {NEEDS_YOU},
            timeout=20, base=0.02, cap=0.05,
            on_tick=lambda n, s, t: print("   轮询 %d：%s" % (n, s)))

        if status == NEEDS_YOU:
            outs = []
            for call in run.required_calls():
                fn = REGISTRY[call["name"]]
                result = fn(**json.loads(call["arguments"]))
                print("   执行 %s -> %s" % (call["name"], result))
                outs.append({"tool_call_id": call["id"],
                             "output": json.dumps(result, ensure_ascii=False)})
            run.submit_tool_outputs(outs)
            continue           # 提交完继续等，别在这里退出

        print("   终态：%s\n" % status)
        return status


if __name__ == "__main__":
    drive(FakeRun(["queued", "in_progress", "completed"]), "一路顺利")
    drive(FakeRun(["queued", "in_progress", "requires_action"]), "需要你执行函数")
    drive(FakeRun(["queued", "failed"]), "平台侧失败")
    drive(FakeRun(["queued", "in_progress", "expired"]), "超时过期")

"""minimax_min.py —— 托管式接口的最小可跑版本：四个对象走一遍。

去掉文件上传、去掉工具、去掉重试，只留下让四对象心智模型立起来的最短路径：
    建 Assistant（岗位说明书）
    建 Thread（档案袋）
    放 Message（信件）
    跑 Run（照着档案办）→ 轮询 → 读回信

先把这 40 行跑通，再去看 minimax_assistant.py 的完整版。

    export MINIMAX_API_KEY='...'
    export MINIMAX_GROUP_ID='...'
    python3 minimax_min.py
"""
import json
import time

import requests

from envkit import need_env, opt_env

KEY = need_env("MINIMAX_API_KEY", "MiniMax 开放平台密钥")
GID = need_env("MINIMAX_GROUP_ID", "MiniMax 的 GroupId")
BASE = opt_env("MINIMAX_BASE", "https://api.minimax.chat")
H = {"Authorization": "Bearer %s" % KEY, "Content-Type": "application/json"}


def post(path, body=None):
    """所有写操作都是 POST + GroupId 查询参数 + JSON 体。"""
    url = "%s/v1/%s?GroupId=%s" % (BASE, path, GID)
    resp = requests.post(url, headers=H, data=json.dumps(body or {}), timeout=60)
    resp.raise_for_status()
    return resp.json()


# 1. Assistant：一次创建，可以被无数个 Thread 反复使用
assistant = post("assistants/create", {
    "model": "abab5.5-chat",
    "name": "算术助手",
    "instructions": "你是一个严谨的算术助手，回答时先写计算过程再写结果",
})
print("assistant:", assistant["id"])

# 2. Thread：一个客户一个档案袋，本身不含任何配置
thread = post("threads/create")
print("thread:", thread["id"])

# 3. Message：把问题作为一封信存进袋子，这一步模型还没动
post("threads/messages/add", {
    "thread_id": thread["id"],
    "role": "user",
    "content": "一斤苹果3元，我买了2.5斤，一共多少钱",
})

# 4. Run：喊一声开工，立刻返回，事情在平台那边异步跑
run = post("threads/run/create", {
    "thread_id": thread["id"],
    "assistant_id": assistant["id"],
})
print("run:", run["id"])

# 5. 轮询：最小版用固定间隔，生产环境务必换成指数退避（见 retrykit.py）
while True:
    url = "%s/v1/threads/run/retrieve?GroupId=%s" % (BASE, GID)
    body = json.dumps({"thread_id": thread["id"], "run_id": run["id"]})
    status = requests.request("GET", url, headers=H, data=body, timeout=60) \
                     .json().get("status", "")
    print("status:", status)
    if status in ("completed", "failed", "cancelled", "expired"):
        break
    time.sleep(2)

# 6. 读回信：助手写的内容也在同一个档案袋里
url = "%s/v1/threads/messages/list?GroupId=%s" % (BASE, GID)
out = requests.get(url, headers=H,
                   data=json.dumps({"thread_id": thread["id"]}), timeout=60).json()
print(json.dumps(out, indent=2, ensure_ascii=False))

"""responses_min.py —— Responses API 的最小调用。

OpenAI 侧接替托管式助手接口的是 Responses API。它把「一次运行」
直接做成了一个 response 对象：送进去 input，拿回来 output。
没有先建助手、再建线程、再建运行这三级台阶。

    export OPENAI_API_KEY='...'
    python3 responses_min.py
"""
from openai import OpenAI

from envkit import MissingEnv, need_env, opt_env


def main():
    client = OpenAI(
        api_key=need_env("OPENAI_API_KEY", "OpenAI 密钥"),
        base_url=opt_env("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )

    # 一次调用 = 旧模型里的「建 Thread + 加 Message + 跑 Run + 读回信」
    resp = client.responses.create(
        model="gpt-4.1-mini",
        instructions="你是一个严谨的算术助手，先写计算过程再写结果",
        input="一斤苹果3元，我买了2.5斤，一共多少钱",
    )

    # output_text 是 SDK 给的便捷属性，把所有文本型 output item 拼起来。
    # 真正的结构在 resp.output 里，是一个 item 列表。
    print("回复：", resp.output_text)
    print("id：", resp.id)

    # 把 item 摊开看一眼，能直观感受到「Run step 变成了 item」这句话：
    for item in resp.output:
        print("item type =", getattr(item, "type", "?"))


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""envkit.py —— 读取环境变量的公共小工具。

托管式接口的鉴权信息（API-KEY、GroupId、base_url）全部从环境变量读，
源码里永远不出现明文。缺失时抛出一句人能看懂的话，而不是让程序在
几十行之后因为 401 才炸掉。

用法：
    from envkit import need_env, opt_env

    API_KEY = need_env("MINIMAX_API_KEY", "MiniMax 开放平台的 API-KEY")
    BASE    = opt_env("MINIMAX_BASE", "https://api.minimax.chat")
"""
import os


class MissingEnv(RuntimeError):
    """环境变量缺失。单独定义一个异常类型，方便上层统一兜底提示。"""


def need_env(name, what=""):
    """取一个必须存在的环境变量；没有就抛出带指引的异常。

    注意用 os.environ.get 而不是 os.environ[name]：
    前者能让我们自己控制报错文案，后者只会抛一个光秃秃的 KeyError。
    """
    val = os.environ.get(name)
    if not val:
        hint = ("（%s）" % what) if what else ""
        raise MissingEnv(
            "环境变量 %s 未设置%s。\n"
            "  临时设置：export %s='你的取值'\n"
            "  或写进项目根目录的 .env，再用 python-dotenv 加载。"
            % (name, hint, name)
        )
    return val.strip()


def opt_env(name, default):
    """取一个可选的环境变量，缺失时用默认值。"""
    val = os.environ.get(name)
    return val.strip() if val else default


def mask(secret):
    """把密钥打码后再打印。日志、异常、截图里都不该出现完整密钥。"""
    if not secret:
        return "(空)"
    if len(secret) <= 8:
        return "*" * len(secret)
    return secret[:4] + "*" * (len(secret) - 8) + secret[-4:]


if __name__ == "__main__":
    # 自检：把当前环境里几个常用变量的存在情况打出来，值一律打码。
    for key in ("MINIMAX_API_KEY", "MINIMAX_GROUP_ID",
                "DASHSCOPE_API_KEY", "OPENAI_API_KEY"):
        raw = os.environ.get(key)
        print("%-20s %s" % (key, mask(raw) if raw else "未设置"))

"""minimax_assistant.py —— 托管式 Agent 接口的完整七步：水果店收银助手。

MiniMax 开放平台提供的这套接口，对象模型与 Assistant / Thread / Message / Run
完全一致：助手配置存在平台上，会话档案袋存在平台上，跑一次叫 Run，
状态要自己轮询。这份脚本把七步一次跑完：

    上传文件 → 创建 Assistant → 创建 Thread → 添加 Message
             → 创建 Run → 轮询状态 → 取回消息列表

运行前准备：
    export MINIMAX_API_KEY='...'      # 平台控制台里找到密钥管理功能获取
    export MINIMAX_GROUP_ID='...'     # 同一处可以找到 GroupId
    同目录准备 fruit_price.txt，内容形如：
        香蕉成本价2元一斤，售价为3元一斤。
        橘子成本价1.5元一斤，售价为2.5元一斤
        苹果成本价3元一斤，售价为3.5元一斤
        芒果成本价5元一斤，售价为6元一斤
        葡萄成本价2元一斤，售价为4元一斤
    （虚构数据，只用来演示 retrieval 工具能不能把文件内容找回来。）
"""
import json
import os

import requests

from envkit import MissingEnv, need_env, opt_env
from retrykit import poll_until

API_KEY = None          # 延迟到 main() 里读，导入本模块不应该直接报错
GROUP_ID = None
BASE = opt_env("MINIMAX_BASE", "https://api.minimax.chat")
PRICE_FILE = opt_env("FRUIT_PRICE_FILE", "./fruit_price.txt")

TIMEOUT = 60            # 单次 HTTP 超时，秒


def _headers(json_body=True):
    """鉴权头。上传文件走 multipart，不能自己写 Content-Type，
    否则 requests 无法补上 multipart 的 boundary，服务端直接解析失败。"""
    head = {"Authorization": "Bearer %s" % API_KEY}
    if json_body:
        head["Content-Type"] = "application/json"
    return head


def _url(path):
    """所有接口都要带 GroupId 查询参数，集中拼一次，省得每个函数重复。"""
    return "%s/v1/%s?GroupId=%s" % (BASE, path, GROUP_ID)


def _check(resp, step):
    """统一的返回体检查：HTTP 码 + 业务 base_resp 双重判断。

    托管式接口常见的坑是 HTTP 200 但业务码非 0，
    只看 status_code 会把失败当成功，一路跑到最后才发现 id 是空串。
    """
    if resp.status_code != 200:
        raise RuntimeError("%s 失败：HTTP %s %s"
                           % (step, resp.status_code, resp.text[:300]))
    data = resp.json()
    base = data.get("base_resp") or {}
    if base and base.get("status_code") not in (0, None):
        raise RuntimeError("%s 失败：业务码 %s %s"
                           % (step, base.get("status_code"), base.get("status_msg")))
    return data


# ---------- 第 0 步：上传文件 ----------
def create_file(path):
    """把本地文件传给平台，拿到 file_id。

    purpose 声明这个文件给谁用；助手检索场景固定填 assistants。
    用 with 打开文件，避免句柄泄漏。
    """
    if not os.path.exists(path):
        raise FileNotFoundError("找不到 %s，先在同目录准备好这个文件" % path)
    with open(path, "rb") as fh:
        resp = requests.post(_url("files/upload"),
                             headers=_headers(json_body=False),
                             data={"purpose": "assistants"},
                             files={"file": fh},
                             timeout=TIMEOUT)
    data = _check(resp, "上传文件")
    file_id = (data.get("file") or {}).get("file_id")
    if not file_id:
        raise RuntimeError("上传成功但没拿到 file_id：%s" % json.dumps(data)[:300])
    return str(file_id)


# ---------- 第 1 步：创建 Assistant ----------
def create_assistant(file_id):
    """创建助手：模型、名字、描述、人设指令、挂载的文件、启用的工具。

    instructions 是这个岗位的说明书，它对这个助手的每一次 Run 都生效；
    tools 里的 retrieval 让助手可以在挂载文件里检索。
    """
    payload = {
        "model": "abab5.5-chat",
        "name": "水果店财务助手",
        "description": "水果店财务助手，用在水果销售过程中计算营业额",
        "instructions": "你是一个理财能手，根据每类水果的售出量以及单价，"
                        "统计其成本和收入，计算出总利润，并给出计算过程",
        "file_ids": [file_id],
        "tools": [{"type": "retrieval"}],
    }
    resp = requests.post(_url("assistants/create"), headers=_headers(),
                         data=json.dumps(payload), timeout=TIMEOUT)
    data = _check(resp, "创建助手")
    return data["id"]


# ---------- 第 2 步：创建 Thread ----------
def create_thread():
    """创建一个会话档案袋。它是空的，只有一个 id，
    之后这一位客户的所有来往信件都按顺序装进这个袋子。"""
    resp = requests.post(_url("threads/create"), headers=_headers(), timeout=TIMEOUT)
    return _check(resp, "创建线程")["id"]


# ---------- 第 3 步：往 Thread 里放 Message ----------
def add_message(thread_id, content, role="user"):
    """往档案袋里塞一张信件。这一步不会触发模型，只是存档。"""
    payload = {"thread_id": thread_id, "role": role, "content": content}
    resp = requests.post(_url("threads/messages/add"), headers=_headers(),
                         data=json.dumps(payload), timeout=TIMEOUT)
    return _check(resp, "添加消息")


# ---------- 第 4 步：创建 Run ----------
def create_run(thread_id, assistant_id):
    """喊一声「照着档案办」。请求立刻返回一个 run_id，事情还没办完。"""
    payload = {"thread_id": thread_id, "assistant_id": assistant_id}
    resp = requests.post(_url("threads/run/create"), headers=_headers(),
                         data=json.dumps(payload), timeout=TIMEOUT)
    return _check(resp, "创建运行")["id"]


# ---------- 第 5 步：轮询 Run 状态 ----------
def fetch_run_status(thread_id, run_id):
    """查一次运行状态，返回状态字符串。"""
    payload = json.dumps({"thread_id": str(thread_id), "run_id": str(run_id)})
    resp = requests.request("GET", _url("threads/run/retrieve"),
                            headers=_headers(), data=payload, timeout=TIMEOUT)
    return _check(resp, "查询运行状态").get("status", "")


def wait_run(thread_id, run_id, timeout=180):
    """指数退避轮询到终态。失败终态照样返回，交给上层判断，不要当成功。"""
    terminal = {"completed", "failed", "cancelled", "expired", "requires_action"}
    return poll_until(
        lambda: fetch_run_status(thread_id, run_id),
        terminal=terminal,
        timeout=timeout,
        on_tick=lambda n, s, t: print("  第 %d 次查询：status=%s（已等待 %.1fs）" % (n, s, t)),
    )


# ---------- 第 6 步：取回消息 ----------
def list_messages(thread_id):
    """把档案袋里现在所有的信件按顺序拉回来，助手的回复也在里面。"""
    payload = json.dumps({"thread_id": thread_id})
    resp = requests.get(_url("threads/messages/list"), headers=_headers(),
                        data=payload, timeout=TIMEOUT)
    return _check(resp, "拉取消息列表")


def main():
    global API_KEY, GROUP_ID
    API_KEY = need_env("MINIMAX_API_KEY", "MiniMax 开放平台密钥")
    GROUP_ID = need_env("MINIMAX_GROUP_ID", "MiniMax 的 GroupId")

    print("[0/6] 上传文件 …")
    file_id = create_file(PRICE_FILE)
    print("      file_id =", file_id)

    print("[1/6] 创建 Assistant …")
    assistant_id = create_assistant(file_id)
    print("      assistant_id =", assistant_id)

    print("[2/6] 创建 Thread …")
    thread_id = create_thread()
    print("      thread_id =", thread_id)

    print("[3/6] 添加 Message …")
    add_message(thread_id,
                "我卖了2斤葡萄，3斤半的香蕉，2斤苹果，"
                "计算下总成本和总收入，给出具体的计算过程")

    print("[4/6] 创建 Run …")
    run_id = create_run(thread_id, assistant_id)
    print("      run_id =", run_id)

    print("[5/6] 轮询状态 …")
    status = wait_run(thread_id, run_id)
    if status != "completed":
        raise RuntimeError("Run 以 %s 结束，没有可用回复" % status)

    print("[6/6] 拉取消息 …")
    msgs = list_messages(thread_id)
    print(json.dumps(msgs, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""dashscope_assistant.py —— 同一套四对象，换一家平台的 SDK 写法。

MiniMax 那份是自己拼 HTTP 请求，这份用阿里 DashScope 的 SDK。
对比着看能看清一件事：**四对象是抽象，不是某一家的接口细节**。
换平台变的是函数名和字段名，不变的是
    Assistant → Thread → Message → Run → Run Step → Message
这条链路。

    pip install dashscope
    export DASHSCOPE_API_KEY='...'
    python3 dashscope_assistant.py
"""
import json
import sys
from http import HTTPStatus

import dashscope

from envkit import MissingEnv, need_env

QUESTION = (
    "香蕉成本价2元一斤，售价为3元一斤；苹果成本价3元一斤，售价为3.5元一斤；"
    "葡萄成本价2元一斤，售价为4元一斤。我卖了2斤葡萄、3.5斤香蕉、2斤苹果，"
    "计算总成本和总收入，给出具体计算过程"
)


def verify(res, step):
    """SDK 把 HTTP 状态码放在返回对象上，不抛异常，所以每步都要自己查。

    漏掉这一步的典型症状：前面某一步其实失败了，
    后面拿 res.id 得到 None，报错却出现在毫不相干的地方。
    """
    if res.status_code != HTTPStatus.OK:
        print("[%s] 失败：%s" % (step, res), file=sys.stderr)
        sys.exit(res.status_code)
    return res


def create_assistant():
    """岗位说明书：模型、名字、描述、instructions、可用工具。

    tools 里填平台自己的内置工具类型；不同平台支持的工具类型不同，
    这也是托管式接口迁移时最先崩掉的地方。
    """
    return dashscope.Assistants.create(
        model="qwen-max",
        name="水果店财务助手",
        description="用在水果销售过程中计算营业额",
        instructions="你是一个理财能手，根据每类水果的售出量以及单价，"
                     "统计其成本和收入，计算出总利润",
        tools=[{"type": "quark_search"}],
    )


def main():
    # SDK 的鉴权是给模块级变量赋值；密钥仍然只从环境变量来
    dashscope.api_key = need_env("DASHSCOPE_API_KEY", "阿里云百炼的 API-KEY")

    assistant = verify(create_assistant(), "创建 Assistant")
    print("assistant:", assistant.id)

    thread = verify(dashscope.Threads.create(), "创建 Thread")
    print("thread:", thread.id)

    message = verify(dashscope.Messages.create(thread.id, content=QUESTION),
                     "添加 Message")
    print("message:", message.id)

    run = verify(dashscope.Runs.create(thread.id, assistant_id=assistant.id),
                 "创建 Run")
    print("run:", run.id, "初始状态:", run.status)

    # SDK 自带的 wait 就是一个封装好的轮询；它会在 completed 或
    # requires_action 时返回。requires_action 意味着「助手要你去执行工具」，
    # 这时候循环还没结束，得提交工具输出之后再继续等。
    run = verify(dashscope.Runs.wait(run.id, thread_id=thread.id), "等待 Run")
    print("终态:", run.status)

    steps = verify(dashscope.Steps.list(run.id, thread_id=thread.id), "列出 Run Step")
    print("run steps:", steps)

    msgs = verify(dashscope.Messages.list(thread.id), "列出 Message")
    print(json.dumps(msgs, default=lambda o: o.__dict__,
                     sort_keys=True, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""responses_conversation.py —— Responses + Conversations：多轮会话的等价写法。

托管式助手接口里，多轮对话靠 Thread：消息存在平台上，每次 Run
自动带上整袋历史。迁到 Responses 之后，这件事由 Conversations 承担：
先建一个 conversation，之后每次 create 都带上它的 id，
平台照样替你保存上下文。

对照关系：
    建 Thread            →  client.conversations.create()
    往 Thread 加 Message →  带 conversation 参数发一次 responses.create()
    跑 Run               →  responses.create() 本身
    列 Thread 的消息      →  client.conversations.items.list(conversation_id)

一个关键差别：conversation 里存的不再只是 message，
而是 item —— 工具调用、工具输出、模型回复全是 item。

    export OPENAI_API_KEY='...'
    python3 responses_conversation.py
"""
from openai import OpenAI

from envkit import MissingEnv, need_env, opt_env


def main():
    client = OpenAI(
        api_key=need_env("OPENAI_API_KEY", "OpenAI 密钥"),
        base_url=opt_env("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )

    # 1. 建一个会话容器，相当于以前的 Thread
    conv = client.conversations.create()
    print("conversation:", conv.id)

    # 2. 第一轮。instructions 相当于以前写在 Assistant 上的岗位说明书，
    #    现在跟着每次请求走 —— 这既是灵活，也是负担：
    #    它不再有平台侧的唯一版本，得由你自己保证每次都传对。
    first = client.responses.create(
        model="gpt-4.1-mini",
        conversation=conv.id,
        instructions="你是一个耐心的数学老师，回答简短，必要时列式",
        input="一斤苹果3元，我买了2.5斤，一共多少钱",
    )
    print("第一轮：", first.output_text)

    # 3. 第二轮只发新问题。历史由 conversation 带着，
    #    不需要像纯 Chat Completions 那样自己把整个 messages 数组重发一遍。
    second = client.responses.create(
        model="gpt-4.1-mini",
        conversation=conv.id,
        input="那如果再加两斤香蕉，香蕉一斤3元呢",
    )
    print("第二轮：", second.output_text)

    # 4. 把会话里的 item 列出来，看看平台替我们存了什么
    items = client.conversations.items.list(conv.id)
    for it in items.data:
        print("item:", getattr(it, "type", "?"), getattr(it, "role", ""))


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""responses_tool_loop.py —— 工具调用循环改成显式自己管。

这是迁移里改动最大的一处。

旧写法（托管式 Run）：提交 Run 之后轮询状态，平台在内部替你
反复调模型；只有轮到「需要你执行本地函数」时，Run 才会停在
requires_action 上等你提交工具输出。循环是平台的。

新写法（Responses）：模型要调工具时，直接在 output 里给你
function_call item；你执行完，把 function_call_output 作为
新的 input item 再发一次。**循环是你的 while 写的。**

代价是多写十几行，好处是这十几行完全在你手里 ——
换任何一家兼容 tool calling 的模型，这段循环都不用重写。

    export OPENAI_API_KEY='...'
    python3 responses_tool_loop.py
"""
import json

from openai import OpenAI

from envkit import MissingEnv, need_env, opt_env

# ---------- 你的真实函数 ----------
PRICE = {"香蕉": 3.0, "橘子": 2.5, "苹果": 3.5, "芒果": 6.0, "葡萄": 4.0}
COST = {"香蕉": 2.0, "橘子": 1.5, "苹果": 3.0, "芒果": 5.0, "葡萄": 2.0}


def lookup_fruit(name):
    """查一种水果的成本价与售价。查不到也要正常返回，别抛异常 ——
    异常会打断循环，而「查无此水果」是模型需要知道的事实。"""
    if name not in PRICE:
        return {"found": False, "fruit": name, "msg": "价目表里没有这一项"}
    return {"found": True, "fruit": name, "cost": COST[name], "price": PRICE[name]}


TOOLS = [{
    "type": "function",
    "name": "lookup_fruit",
    "description": "按水果名查询它的成本价与售价，单位是元每斤",
    "parameters": {
        "type": "object",
        "properties": {
            "name": {"type": "string", "description": "水果名称，例如 苹果"},
        },
        "required": ["name"],
    },
}]

REGISTRY = {"lookup_fruit": lookup_fruit}


def main():
    client = OpenAI(
        api_key=need_env("OPENAI_API_KEY", "OpenAI 密钥"),
        base_url=opt_env("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )
    conv = client.conversations.create()

    # input 是一个 item 列表；第一轮只有用户那一条
    pending = [{"role": "user",
                "content": "我卖了2斤葡萄、3.5斤香蕉、2斤苹果，总成本和总收入各是多少"}]

    for turn in range(1, 9):                       # 上限保护，防止死循环
        resp = client.responses.create(
            model="gpt-4.1-mini",
            conversation=conv.id,
            instructions="你是水果店财务助手，需要价格时调用工具，不要自己编价格",
            tools=TOOLS,
            input=pending,
        )

        # 挑出这一轮模型要求执行的函数调用
        calls = [it for it in resp.output if getattr(it, "type", "") == "function_call"]
        if not calls:
            print("第 %d 轮：模型给出最终答复" % turn)
            print(resp.output_text)
            return

        print("第 %d 轮：模型要求执行 %d 个函数" % (turn, len(calls)))
        pending = []
        for call in calls:
            fn = REGISTRY.get(call.name)
            if fn is None:
                # 模型编了一个不存在的函数名。把这件事如实告诉它，
                # 比直接抛异常更容易让它自己纠正。
                result = {"error": "未注册的函数 %s" % call.name}
            else:
                result = fn(**json.loads(call.arguments))
            print("   %s(%s) -> %s" % (call.name, call.arguments, result))
            # 回填的 item 类型是 function_call_output，靠 call_id 和请求配对
            pending.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

    print("达到轮次上限仍未收敛，检查工具描述是否含糊")


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""migrate_side_by_side.py —— 同一个需求，两种写法并排放。

需求：让一个「严谨算术助手」回答两轮追问，并把会话留在服务端。

上半部分是托管式四对象写法（伪代码形态，接口已不可用，留着是为了对照）；
下半部分是 Responses + Conversations 的可运行写法。
把两段并排读一遍，迁移要改哪几行就一目了然。

    export OPENAI_API_KEY='...'
    python3 migrate_side_by_side.py
"""
from openai import OpenAI

from envkit import MissingEnv, need_env, opt_env

INSTRUCTIONS = "你是一个严谨的算术助手，先写计算过程再写结果"
Q1 = "一斤苹果3元，我买了2.5斤，一共多少钱"
Q2 = "再加两斤香蕉，香蕉一斤3元，一共多少钱"


# ---------------------------------------------------------------- 旧写法
def old_way_pseudocode():
    """托管式四对象的形状。这段不要运行，接口已经不存在了。

        assistant = client.beta.assistants.create(       # 岗位说明书存在平台
            model="gpt-4-turbo",
            name="算术助手",
            instructions=INSTRUCTIONS,
        )
        thread = client.beta.threads.create()            # 档案袋存在平台

        client.beta.threads.messages.create(             # 第一封信
            thread_id=thread.id, role="user", content=Q1)
        run = client.beta.threads.runs.create(           # 喊一声开工
            thread_id=thread.id, assistant_id=assistant.id)
        run = wait_until_terminal(run)                   # 轮询到终态
        msgs = client.beta.threads.messages.list(thread.id)

        client.beta.threads.messages.create(             # 第二封信，同一个袋子
            thread_id=thread.id, role="user", content=Q2)
        run = client.beta.threads.runs.create(
            thread_id=thread.id, assistant_id=assistant.id)

    要点：配置（assistant）与会话（thread）是两个独立的服务端对象，
    运行（run）是异步任务，工具调用循环由平台内部完成。
    """
    return old_way_pseudocode.__doc__


# ---------------------------------------------------------------- 新写法
def new_way():
    client = OpenAI(
        api_key=need_env("OPENAI_API_KEY", "OpenAI 密钥"),
        base_url=opt_env("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )

    # Thread → Conversation：会话容器还在服务端，换了个名字和语义
    conv = client.conversations.create()

    # Message + Run → 一次 responses.create()：
    # 送进 input item，同步拿回 output item，不再有 run 状态机
    first = client.responses.create(
        model="gpt-4.1-mini",
        conversation=conv.id,
        instructions=INSTRUCTIONS,     # 配置跟着请求走，不再是平台上的对象
        input=Q1,
    )
    print("第一轮：", first.output_text)

    second = client.responses.create(
        model="gpt-4.1-mini",
        conversation=conv.id,
        instructions=INSTRUCTIONS,
        input=Q2,
    )
    print("第二轮：", second.output_text)

    # Thread 的 message 列表 → Conversation 的 item 列表
    for it in client.conversations.items.list(conv.id).data:
        print("item:", getattr(it, "type", "?"), getattr(it, "role", ""))


MAPPING = [
    ("Assistants", "Prompts", "配置可版本化、可回滚；在控制台里创建，不通过接口建"),
    ("Threads", "Conversations", "从只存 message 变成存 item"),
    ("Runs", "Responses", "异步任务变成一次请求；工具调用循环由自己写"),
    ("Run steps", "Items", "泛化成统一的 item 对象"),
]


def print_mapping():
    print("%-12s %-16s %s" % ("旧对象", "新对象", "变化"))
    for old, new, why in MAPPING:
        print("%-12s %-16s %s" % (old, new, why))


if __name__ == "__main__":
    print_mapping()
    print()
    try:
        new_way()
    except MissingEnv as exc:
        print("（未配置密钥，跳过实际调用）")
        print(exc)

"""selfhost_store.py —— 把档案袋搬回自己家：SQLite 版会话存储。

托管式接口替你保管 Thread 和 Message，代价是这些数据在别人手里，
接口下线那天一起消失。自己存其实只要三张表：

    agents     一行 = 一个 Assistant（岗位说明书：模型、指令、工具清单）
    threads    一行 = 一个会话档案袋
    messages   一行 = 袋子里的一封信，按 seq 排序

这份实现不依赖任何第三方库，标准库 sqlite3 就够；
换成 MySQL / PostgreSQL 时，只有建表语句和占位符要改。

    python3 selfhost_store.py     # 直接运行会跑一段自检
"""
import json
import os
import sqlite3
import time
import uuid

DB_PATH = os.environ.get("AGENT_DB", "./agent_state.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS agents (
  id           TEXT PRIMARY KEY,
  name         TEXT NOT NULL,
  model        TEXT NOT NULL,
  instructions TEXT NOT NULL,
  tools_json   TEXT NOT NULL DEFAULT '[]',
  created_at   REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS threads (
  id         TEXT PRIMARY KEY,
  agent_id   TEXT NOT NULL,
  title      TEXT,
  created_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS messages (
  id         TEXT PRIMARY KEY,
  thread_id  TEXT NOT NULL,
  seq        INTEGER NOT NULL,
  role       TEXT NOT NULL,          -- system / user / assistant / tool
  content    TEXT,
  extra_json TEXT,                   -- tool_calls、tool_call_id 等原样存这里
  created_at REAL NOT NULL
);

-- 按 thread 取消息是最高频的查询，这条索引不能省
CREATE INDEX IF NOT EXISTS idx_msg_thread ON messages(thread_id, seq);
"""


def _uid(prefix):
    """自己发 id。格式模仿托管式接口，迁移时对账方便。"""
    return "%s_%s" % (prefix, uuid.uuid4().hex[:24])


class Store:
    def __init__(self, path=DB_PATH):
        self.conn = sqlite3.connect(path)
        self.conn.row_factory = sqlite3.Row
        self.conn.executescript(SCHEMA)
        self.conn.commit()

    # ---------- Assistant ----------
    def create_agent(self, name, model, instructions, tools=None):
        aid = _uid("asst")
        self.conn.execute(
            "INSERT INTO agents(id,name,model,instructions,tools_json,created_at)"
            " VALUES(?,?,?,?,?,?)",
            (aid, name, model, instructions,
             json.dumps(tools or [], ensure_ascii=False), time.time()))
        self.conn.commit()
        return aid

    def get_agent(self, agent_id):
        row = self.conn.execute("SELECT * FROM agents WHERE id=?", (agent_id,)).fetchone()
        if row is None:
            raise KeyError("没有这个 agent：%s" % agent_id)
        agent = dict(row)
        agent["tools"] = json.loads(agent.pop("tools_json"))
        return agent

    # ---------- Thread ----------
    def create_thread(self, agent_id, title=None):
        self.get_agent(agent_id)          # 提前校验，避免建出孤儿 thread
        tid = _uid("thread")
        self.conn.execute(
            "INSERT INTO threads(id,agent_id,title,created_at) VALUES(?,?,?,?)",
            (tid, agent_id, title, time.time()))
        self.conn.commit()
        return tid

    # ---------- Message ----------
    def add_message(self, thread_id, role, content, extra=None):
        row = self.conn.execute(
            "SELECT COALESCE(MAX(seq),0)+1 AS nxt FROM messages WHERE thread_id=?",
            (thread_id,)).fetchone()
        mid = _uid("msg")
        self.conn.execute(
            "INSERT INTO messages(id,thread_id,seq,role,content,extra_json,created_at)"
            " VALUES(?,?,?,?,?,?,?)",
            (mid, thread_id, row["nxt"], role, content,
             json.dumps(extra, ensure_ascii=False) if extra else None, time.time()))
        self.conn.commit()
        return mid

    def history(self, thread_id, limit=40):
        """取最近 limit 条，按时间正序返回，形状直接就是 messages 数组。

        limit 是必须的：托管式接口会自动截断以适应上下文窗口，
        自己存就得自己截 —— 这正是自建要补回来的那部分工作量之一。
        """
        rows = self.conn.execute(
            "SELECT role,content,extra_json FROM messages WHERE thread_id=?"
            " ORDER BY seq DESC LIMIT ?", (thread_id, limit)).fetchall()
        out = []
        for row in reversed(rows):
            msg = {"role": row["role"]}
            if row["content"] is not None:
                msg["content"] = row["content"]
            if row["extra_json"]:
                msg.update(json.loads(row["extra_json"]))
            out.append(msg)
        return out

    def export_thread(self, thread_id):
        """整袋导出成 JSON。托管式接口最缺的就是这个动作 ——
        数据在自己库里，导出、备份、迁移随时可做。"""
        return {"thread_id": thread_id, "messages": self.history(thread_id, limit=10000)}

    def close(self):
        self.conn.close()


if __name__ == "__main__":
    store = Store(":memory:")
    agent = store.create_agent("水果店财务助手", "glm-4",
                               "你是理财能手，统计成本与收入", tools=[])
    thread = store.create_thread(agent, title="9月3日 张老板")
    store.add_message(thread, "user", "卖了2斤葡萄")
    store.add_message(thread, "assistant", "收到，葡萄售价4元一斤，收入8元")
    store.add_message(thread, "user", "再加3斤香蕉呢")
    print(json.dumps(store.export_thread(thread), indent=2, ensure_ascii=False))
    store.close()

"""selfhost_agent.py —— 自建托管式助手：会话表 + Function Call 循环。

把四对象全部搬到自己家：
    Assistant  → agents 表里的一行
    Thread     → threads 表里的一行
    Message    → messages 表里的一行
    Run        → 本文件里的 run() 函数，一次调用就是一次运行
    Run Step   → 循环里每一轮的记录，同样落进 messages 表

底座只需要一个支持 tool calling 的 Chat Completions 接口 ——
这是目前最不可能被下线的一层，几乎每一家都兼容它。

    export LLM_API_KEY='...'
    export LLM_BASE_URL='https://open.bigmodel.cn/api/paas/v4/'   # 换成你的
    export LLM_MODEL='glm-4'
    python3 selfhost_agent.py
"""
import json

from openai import OpenAI

from envkit import MissingEnv, need_env, opt_env
from selfhost_store import Store

# ---------- 工具区：这就是你的「平台能力」 ----------
PRICE = {"香蕉": 3.0, "橘子": 2.5, "苹果": 3.5, "芒果": 6.0, "葡萄": 4.0}
COST = {"香蕉": 2.0, "橘子": 1.5, "苹果": 3.0, "芒果": 5.0, "葡萄": 2.0}


def lookup_fruit(name):
    if name not in PRICE:
        return {"found": False, "fruit": name, "msg": "价目表里没有这一项"}
    return {"found": True, "fruit": name, "cost": COST[name], "price": PRICE[name]}


REGISTRY = {"lookup_fruit": lookup_fruit}

TOOLS = [{
    "type": "function",
    "function": {
        "name": "lookup_fruit",
        "description": "按水果名查询成本价与售价，单位是元每斤",
        "parameters": {
            "type": "object",
            "properties": {"name": {"type": "string", "description": "水果名称"}},
            "required": ["name"],
        },
    },
}]


def make_client():
    return OpenAI(
        api_key=need_env("LLM_API_KEY", "任意兼容 OpenAI 协议的模型服务密钥"),
        base_url=opt_env("LLM_BASE_URL", "https://open.bigmodel.cn/api/paas/v4/"),
    )


def run(store, client, thread_id, user_text, max_turns=8):
    """一次 Run：把用户消息入库 → 循环调模型与工具 → 最终答复入库。

    与托管式接口的关键差别：这里每一轮的中间状态都写进 messages 表，
    程序中途崩了也能接着往下跑，而不是丢一个悬空的 run_id。
    """
    agent_id = store.conn.execute(
        "SELECT agent_id FROM threads WHERE id=?", (thread_id,)).fetchone()["agent_id"]
    agent = store.get_agent(agent_id)

    store.add_message(thread_id, "user", user_text)

    for turn in range(1, max_turns + 1):
        # 每轮都用 instructions + 历史重新拼一份 messages
        messages = [{"role": "system", "content": agent["instructions"]}]
        messages += store.history(thread_id)

        resp = client.chat.completions.create(
            model=agent["model"], messages=messages, tools=TOOLS, tool_choice="auto")
        msg = resp.choices[0].message

        if not msg.tool_calls:
            store.add_message(thread_id, "assistant", msg.content)
            print("第 %d 轮：给出最终答复" % turn)
            return msg.content

        # 把模型这条带 tool_calls 的消息原样入库，回执才认得爹
        store.add_message(thread_id, "assistant", msg.content, extra={
            "tool_calls": [{
                "id": c.id, "type": "function",
                "function": {"name": c.function.name,
                             "arguments": c.function.arguments},
            } for c in msg.tool_calls],
        })

        for call in msg.tool_calls:
            fn = REGISTRY.get(call.function.name)
            result = (fn(**json.loads(call.function.arguments)) if fn
                      else {"error": "未注册的函数 %s" % call.function.name})
            print("   %s(%s) -> %s" % (call.function.name,
                                       call.function.arguments, result))
            store.add_message(thread_id, "tool",
                              json.dumps(result, ensure_ascii=False),
                              extra={"tool_call_id": call.id})

    return "达到轮次上限仍未收敛"


def main():
    store = Store()
    client = make_client()

    agent_id = store.create_agent(
        name="水果店财务助手",
        model=opt_env("LLM_MODEL", "glm-4"),
        instructions="你是水果店财务助手。需要价格时必须调用工具查询，"
                     "不要自己编价格；最后给出成本、收入与利润，并写出计算过程。",
        tools=["lookup_fruit"],
    )
    thread_id = store.create_thread(agent_id, title="演示会话")

    print(run(store, client, thread_id,
              "我卖了2斤葡萄、3.5斤香蕉、2斤苹果，总成本和总收入各是多少"))
    print("---- 第二轮，验证上下文确实留在自己的库里 ----")
    print(run(store, client, thread_id, "那利润是多少"))

    print(json.dumps(store.export_thread(thread_id), indent=2, ensure_ascii=False))
    store.close()


if __name__ == "__main__":
    try:
        main()
    except MissingEnv as exc:
        raise SystemExit(str(exc))

"""export_hosted_thread.py —— 退出演练：把托管在平台上的会话导回自己的库。

托管式路线可以走，但必须配一条随时能撤的通道。这份脚本做的就是这件事：
    遍历你记录下来的 thread_id → 拉平台上的消息列表
    → 归一成统一形状 → 写进本地 SQLite

关键在于**平时就把 thread_id 存在自己这边**。只存在平台控制台里的会话，
接口一关就再也列不出来了。

    export HOSTED_API_KEY='...'
    python3 export_hosted_thread.py thread_ids.txt
"""
import json
import os
import sys

import requests

from selfhost_store import Store

BASE = os.environ.get("HOSTED_BASE", "https://api.example.com")
GROUP = os.environ.get("HOSTED_GROUP_ID", "")


def _headers():
    key = os.environ.get("HOSTED_API_KEY")
    if not key:
        raise SystemExit("环境变量 HOSTED_API_KEY 未设置（托管式平台密钥）")
    return {"Authorization": "Bearer %s" % key, "Content-Type": "application/json"}


def fetch_messages(thread_id):
    """拉一个 thread 的全部消息。不同平台路径不同，改这一处即可。"""
    resp = requests.get("%s/v1/threads/messages/list" % BASE,
                        params={"GroupId": GROUP} if GROUP else None,
                        headers=_headers(),
                        data=json.dumps({"thread_id": thread_id}), timeout=60)
    resp.raise_for_status()
    return resp.json()


def normalize(raw):
    """把平台返回的消息结构压成 {role, content} 列表。

    各家的字段名不一样（data / messages / object 里套 content 数组），
    这一层就是为了让下游只面对一种形状。
    """
    items = raw.get("data") or raw.get("messages") or []
    out = []
    for it in items:
        content = it.get("content")
        if isinstance(content, list):
            # 有的平台把 content 做成分块数组，取其中的文本块拼起来
            parts = []
            for blk in content:
                if isinstance(blk, dict):
                    parts.append((blk.get("text") or {}).get("value")
                                 or blk.get("text") or "")
                else:
                    parts.append(str(blk))
            content = "\n".join(p for p in parts if p)
        out.append({"role": it.get("role", "assistant"),
                    "content": content if isinstance(content, str) else json.dumps(
                        content, ensure_ascii=False)})
    return out


def main(id_file):
    if not os.path.exists(id_file):
        raise SystemExit("找不到 %s：里面一行一个 thread_id" % id_file)

    store = Store()
    agent_id = store.create_agent("导入的历史会话", "unknown",
                                  "从托管平台导回的归档，仅供查阅")

    total = 0
    with open(id_file, encoding="utf-8") as fh:
        for line in fh:
            tid = line.strip()
            if not tid or tid.startswith("#"):
                continue
            try:
                msgs = normalize(fetch_messages(tid))
            except Exception as exc:
                print("%s 导出失败：%s" % (tid, exc))
                continue
            local = store.create_thread(agent_id, title="导入自 %s" % tid)
            for msg in msgs:
                store.add_message(local, msg["role"], msg["content"])
            total += len(msgs)
            print("%s → %s，%d 条" % (tid, local, len(msgs)))

    print("累计导入 %d 条消息，落库路径 %s"
          % (total, os.environ.get("AGENT_DB", "./agent_state.db")))
    store.close()


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "thread_ids.txt")

"""hosted_agent_skeleton.py —— 托管式 Agent 接口通用骨架（改 TODO 即可）。

适用：任何提供 Assistant / Thread / Message / Run 四对象的托管式接口
（MiniMax、阿里百炼等），不管它是 HTTP 还是 SDK，形状都是这五步。

设计要点：
  · 接口细节全部收进 HostedBackend 一个类，换平台只改这个类
  · 业务代码只认 create_agent / open_thread / say 三个方法
  · 轮询用指数退避 + 总超时，不用裸 sleep
  · 会话 id 自己也留一份，平台侧出问题时能对账

    export HOSTED_API_KEY='...'
    python3 hosted_agent_skeleton.py
"""
import json
import os
import time

import requests


class MissingEnv(RuntimeError):
    pass


def need_env(name, what=""):
    val = os.environ.get(name)
    if not val:
        raise MissingEnv("环境变量 %s 未设置（%s）" % (name, what))
    return val.strip()


class HostedBackend:
    """把一家平台的四对象接口包成统一形状。换平台只重写这个类。"""

    # TODO 1：换成目标平台的 base url 与鉴权方式
    BASE = os.environ.get("HOSTED_BASE", "https://api.example.com")

    def __init__(self):
        self.key = need_env("HOSTED_API_KEY", "托管式平台密钥")
        # TODO 2：有的平台还要 GroupId / workspace 之类的租户参数，在这里补
        self.extra = {k: v for k, v in
                      (("GroupId", os.environ.get("HOSTED_GROUP_ID")),) if v}
        self.session = requests.Session()

    def _headers(self):
        return {"Authorization": "Bearer %s" % self.key,
                "Content-Type": "application/json"}

    def _post(self, path, body=None):
        resp = self.session.post("%s/%s" % (self.BASE, path),
                                 params=self.extra, headers=self._headers(),
                                 data=json.dumps(body or {}), timeout=60)
        if resp.status_code != 200:
            raise RuntimeError("%s 返回 HTTP %s：%s"
                               % (path, resp.status_code, resp.text[:300]))
        return resp.json()

    # ---- 四对象，每个平台都有对应接口，只是路径不同 ----
    def create_agent(self, name, model, instructions, tools=None):
        # TODO 3：换成该平台的创建助手接口与字段名
        data = self._post("v1/assistants/create", {
            "name": name, "model": model,
            "instructions": instructions, "tools": tools or [],
        })
        return data["id"]

    def create_thread(self):
        # TODO 4：换成该平台的创建线程接口
        return self._post("v1/threads/create")["id"]

    def add_message(self, thread_id, content, role="user"):
        # TODO 5：换成该平台的添加消息接口
        return self._post("v1/threads/messages/add",
                          {"thread_id": thread_id, "role": role, "content": content})

    def create_run(self, thread_id, agent_id):
        # TODO 6：换成该平台的创建运行接口
        return self._post("v1/threads/run/create",
                          {"thread_id": thread_id, "assistant_id": agent_id})["id"]

    def run_status(self, thread_id, run_id):
        # TODO 7：换成该平台的查询运行状态接口
        return self._post("v1/threads/run/retrieve",
                          {"thread_id": thread_id, "run_id": run_id}).get("status", "")

    def list_messages(self, thread_id):
        # TODO 8：换成该平台的消息列表接口
        return self._post("v1/threads/messages/list", {"thread_id": thread_id})


TERMINAL_OK = {"completed", "succeeded"}
TERMINAL_BAD = {"failed", "cancelled", "expired", "requires_action"}


class HostedAgent:
    """业务层只用这三个方法，接口换代时这一层不用改。"""

    def __init__(self, backend, name, model, instructions, tools=None):
        self.be = backend
        self.agent_id = backend.create_agent(name, model, instructions, tools)
        # 本地留痕：托管式对象的 id 是唯一能对账的东西，务必自己也存一份
        self.log = {"agent_id": self.agent_id, "threads": []}

    def open_thread(self):
        tid = self.be.create_thread()
        self.log["threads"].append({"thread_id": tid, "runs": []})
        return tid

    def say(self, thread_id, text, timeout=180):
        self.be.add_message(thread_id, text)
        run_id = self.be.create_run(thread_id, self.agent_id)
        self.log["threads"][-1]["runs"].append(run_id)

        started, delay = time.time(), 0.5
        while True:
            status = self.be.run_status(thread_id, run_id)
            if status in TERMINAL_OK:
                return self.be.list_messages(thread_id)
            if status in TERMINAL_BAD:
                raise RuntimeError("Run 以 %s 结束" % status)
            if time.time() - started > timeout:
                raise TimeoutError("等待 %ss 后仍为 %s" % (timeout, status))
            time.sleep(delay)
            delay = min(8.0, delay * 1.8)      # 退避，别把服务端打满


if __name__ == "__main__":
    try:
        agent = HostedAgent(
            HostedBackend(),
            name="TODO 9：助手名字",
            model="TODO 10：平台上的模型名",
            instructions="TODO 11：岗位说明书，写清角色、边界、输出格式",
            tools=[],                          # TODO 12：平台内置工具类型
        )
        thread = agent.open_thread()
        print(json.dumps(agent.say(thread, "TODO 13：第一句话"),
                         indent=2, ensure_ascii=False))
        print("本地留痕：", json.dumps(agent.log, ensure_ascii=False))
    except MissingEnv as exc:
        raise SystemExit(str(exc))