Coze 实战：从工作流到可发布的智能体

"""单节点隔离调试：把工作流里的一个 LLM 节点搬到本地跑。

工作流跑错时最忌讳「改一处、整条重跑一遍」：含循环的工作流跑一次要几分钟。
更快的做法：节点提示词是纯文本、输入就是几个变量，都能搬到本地，
本地跑一次只花几秒，定稿后再贴回平台。只需一个 OpenAI 风格的兼容端点。
"""
from __future__ import annotations

import json
import os
import sys
import urllib.request
from pathlib import Path

# 任何 OpenAI 风格的兼容端点都可以；调提示词时用哪个模型不重要，能快速对比就行。
BASE_URL = os.environ.get("LLM_BASE_URL", "http://127.0.0.1:8000/v1")
MODEL = os.environ.get("LLM_MODEL", "qwen2.5-7b-instruct")


def load_prompt(path: str) -> str:
    """从文件读系统提示词。

    把提示词放进文件而不是写在代码字符串里，有三个好处：
    能进版本管理、能 diff 两版差异、能直接复制回平台节点。
    """
    return Path(path).read_text(encoding="utf-8")


def render(template: str, variables: dict) -> str:
    """把 {{变量名}} 占位符替换成真实值，和平台节点里的写法保持一致。

    刻意用最笨的字符串替换而不是模板引擎：
    目的是让本地渲染结果与平台完全一致，多一层语法糖就多一层不一致的风险。
    """
    rendered = template
    for key, value in variables.items():
        rendered = rendered.replace("{{%s}}" % key, str(value))

    # 渲染完还剩 {{ 说明有变量没喂，必须当场报错——线上最常见的故障就是
    # 占位符原样进提示词被当字面量，结果「怪但不报错」。
    if "{{" in rendered:
        leftover = rendered[rendered.index("{{"): rendered.index("{{") + 40]
        raise ValueError("仍有未替换的占位符：%s…… 检查变量名是否逐字一致" % leftover)
    return rendered


def call_llm(system_prompt: str, user_prompt: str, *, temperature: float = 0.1) -> str:
    """最小的 chat completions 调用。

    temperature 默认压到很低：调试提示词时要的是可复现，
    每次结果都不一样就没法判断「是提示词改好了还是这次运气好」。
    """
    payload = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "temperature": temperature,
    }
    headers = {"Content-Type": "application/json"}
    api_key = os.environ.get("LLM_API_KEY", "")
    if api_key:
        headers["Authorization"] = "Bearer %s" % api_key

    request = urllib.request.Request(
        BASE_URL.rstrip("/") + "/chat/completions",
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=120) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    return data["choices"][0]["message"]["content"]


def check_json_output(text: str, required_keys: list) -> dict:
    """节点声明输出 JSON 时，本地先把这份契约验一遍。

    平台上的 LLM 节点选了 JSON 输出格式，仍然可能返回带 ```json 包裹的文本。
    下游节点按对象取值就会拿到空值，而且不报错——这是最难查的一类问题。
    """
    cleaned = text.strip()
    if cleaned.startswith("```"):
        # 去掉代码围栏：首行可能是 ```json，末行是 ```
        lines = [line for line in cleaned.splitlines() if not line.strip().startswith("```")]
        cleaned = "\n".join(lines)

    parsed = json.loads(cleaned)
    missing = [key for key in required_keys if key not in parsed]
    if missing:
        raise ValueError("输出缺少字段：%s；把字段名写进提示词并给一个示例" % missing)
    return parsed


def main() -> int:
    if len(sys.argv) < 3:
        print("用法：python coze_node_debug.py <提示词文件> <变量JSON> [必需字段,逗号分隔]")
        print("例：python coze_node_debug.py ./split.txt '{\"input\":\"张三 …\"}' skill,project_exp")
        return 2

    system_prompt = load_prompt(sys.argv[1])
    variables = json.loads(sys.argv[2])
    required = sys.argv[3].split(",") if len(sys.argv) > 3 else []

    user_prompt = render("输入的内容为:\n{{input}}", variables) if "input" in variables else ""
    print("===== 系统提示词（%d 字）=====" % len(system_prompt))
    print("===== 用户提示词 =====")
    print(user_prompt)

    output = call_llm(system_prompt, user_prompt)
    print("===== 模型输出 =====")
    print(output)

    if required:
        parsed = check_json_output(output, required)
        print("===== 字段校验通过 =====")
        print(json.dumps(parsed, ensure_ascii=False, indent=2)[:800])
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""集中式配置：所有凭证、ID、端点都从环境变量读入，源码里不出现任何明文。

用法（bash）：
    export COZE_API_TOKEN='...'          # 访问令牌，调试用 PAT，线上换 SAT / OAuth
    export COZE_API_BASE='https://api.coze.cn'
    export COZE_BOT_ID='...'             # 已发布到 API/SDK 渠道的智能体 ID
    export COZE_WORKFLOW_ID='...'        # 已发布的工作流 ID
    export COZE_USER_ID='heima-demo'     # 业务侧自定义的用户标识

设计要点：缺必填项立刻抛错并点名（不等 HTTP 401）；token 只在内存里、打印必掩码；
端点路径也做成可配置项，平台改版时只改一处。
"""
from __future__ import annotations

import os
from dataclasses import dataclass, field

# 国际站与国内站是两套独立环境，域名、模型与插件生态都不同；换环境只改这一个变量。
DEFAULT_API_BASE = "https://api.coze.cn"

# 三条最常用的 OpenAPI 路径，以平台文档为准，集中定义便于替换。
PATH_WORKFLOW_RUN = os.environ.get("COZE_PATH_WORKFLOW_RUN", "/v1/workflow/run")
PATH_WORKFLOW_STREAM = os.environ.get("COZE_PATH_WORKFLOW_STREAM", "/v1/workflow/stream_run")
PATH_FILE_UPLOAD = os.environ.get("COZE_PATH_FILE_UPLOAD", "/v1/files/upload")


class MissingConfig(RuntimeError):
    """配置缺失。单独定义，便于上层区分「没配好」和「调用失败」。"""


def _require(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise MissingConfig(
            "环境变量 %s 未设置。先 export %s='...' 再运行，"
            "不要把令牌写进源码或提交进 Git。" % (name, name)
        )
    return value


def mask(secret: str) -> str:
    """把令牌掩码成可以安全打印的形式：保留头 4 尾 4，中间一律星号。"""
    if len(secret) <= 8:
        return "*" * len(secret)
    return "%s%s%s" % (secret[:4], "*" * (len(secret) - 8), secret[-4:])


@dataclass
class CozeSettings:
    """一份运行期配置。所有脚本都从 CozeSettings.from_env() 起步。"""

    api_token: str = field(repr=False)
    api_base: str = DEFAULT_API_BASE
    bot_id: str = ""
    workflow_id: str = ""
    user_id: str = "heima-demo-user"
    timeout: float = 600.0          # 录音分析这类长任务单次可能跑几分钟
    connect_timeout: float = 10.0

    @classmethod
    def from_env(cls, *, need_bot: bool = False, need_workflow: bool = False) -> "CozeSettings":
        """从环境变量装配配置。need_bot / need_workflow 控制哪些 ID 是这次必须的。"""
        settings = cls(
            api_token=_require("COZE_API_TOKEN"),
            api_base=os.environ.get("COZE_API_BASE", DEFAULT_API_BASE).rstrip("/"),
            bot_id=os.environ.get("COZE_BOT_ID", "").strip(),
            workflow_id=os.environ.get("COZE_WORKFLOW_ID", "").strip(),
            user_id=os.environ.get("COZE_USER_ID", "heima-demo-user").strip(),
        )
        if need_bot and not settings.bot_id:
            raise MissingConfig("环境变量 COZE_BOT_ID 未设置：需要一个已发布到 API/SDK 渠道的智能体 ID。")
        if need_workflow and not settings.workflow_id:
            raise MissingConfig("环境变量 COZE_WORKFLOW_ID 未设置：需要一个已发布的工作流 ID。")
        return settings

    @property
    def auth_header(self) -> dict:
        return {"Authorization": "Bearer %s" % self.api_token}

    def describe(self) -> str:
        """启动自检打印这一行：确认读到了配置，且不泄漏令牌。"""
        return (
            "api_base=%s  token=%s  bot_id=%s  workflow_id=%s  user_id=%s"
            % (
                self.api_base,
                mask(self.api_token),
                self.bot_id or "(未设置)",
                self.workflow_id or "(未设置)",
                self.user_id,
            )
        )


if __name__ == "__main__":
    # 直接运行本文件，用来验证环境变量配好了没有。
    print(CozeSettings.from_env().describe())

"""第一个要跑的脚本：连通性自检。

在搭任何业务之前先确认四件事：
1. 令牌读到了（而且没有写死在源码里）；
2. 端点通（网络、域名、代理都没挡）；
3. 令牌有权限（能列出工作空间说明鉴权通过）；
4. 拿得到 logid（后面排查任何问题都靠它回平台后台定位）。

任一不成立，后面写多少业务代码都是白写。先跑通它，再动工作流。
"""
from __future__ import annotations

import sys

from coze_client import CozeError, build_sdk_client
from coze_settings import CozeSettings, MissingConfig


def ping_with_sdk(settings: CozeSettings) -> int:
    """用 SDK 列出当前令牌能看见的工作空间。

    工作空间是资源隔离的基础单元：不同空间里的智能体、工作流、知识库、
    数据库互不可见。能列出空间，说明令牌关联的空间范围是对的；
    列出来是空的，多半是令牌建的时候没有勾选这个空间。
    """
    coze = build_sdk_client(settings)
    if coze is None:
        print("未安装 SDK（pip install cozepy），改用 HTTP 路径自检。")
        return 2

    workspaces = coze.workspaces.list()
    count = 0
    for workspace in workspaces:
        count += 1
        # 迭代器会自动翻页，不用自己拼 page_num / page_size。
        print("  - %s" % workspace.model_dump_json(indent=2))

    # logid 是这次请求的唯一标识。失败时把它贴给平台，比截图有用得多。
    logid = getattr(getattr(workspaces, "response", None), "logid", "")
    print("共 %d 个工作空间，logid=%s" % (count, logid))
    if count == 0:
        print("提示：令牌没有关联任何空间，去令牌管理里补上空间与接口权限。")
    return 0 if count else 1


def main() -> int:
    try:
        settings = CozeSettings.from_env()
    except MissingConfig as exc:
        print("配置缺失：%s" % exc)
        return 3

    print("配置就绪：%s" % settings.describe())
    try:
        return ping_with_sdk(settings)
    except CozeError as exc:
        print("调用失败：%s" % exc)
        return 4
    except Exception as exc:  # noqa: BLE001 - 自检脚本要把任何异常都讲清楚
        print("未预期的异常：%s: %s" % (type(exc).__name__, exc))
        print("排查顺序：令牌是否过期 → 端点域名是否用错环境 → 出网是否被代理拦截。")
        return 5


if __name__ == "__main__":
    sys.exit(main())

"""上传文件 + 多模态流式对话：从自己的程序里调用一个已发布的智能体。

链路：本地文件 → 上传得到 file_id → 组装多模态消息 → 流式发起对话 → 按事件消费。

关键认知：上传和对话是两步（上传只换到 file_id，要在消息体里带上它）；
流式返回的是一串事件而不是一段文本，正文要自己拼；
智能体必须已发布到 API/SDK 渠道，否则 bot_id 调不通。
"""
from __future__ import annotations

import os
import sys
from pathlib import Path
from typing import Optional

from coze_client import build_sdk_client
from coze_settings import CozeSettings, MissingConfig

# 平台对单个上传文件有大小上限，具体数值以平台文档为准。
# 这里读环境变量，给一个保守默认值，避免把会变的数字写死在代码里。
MAX_UPLOAD_BYTES = int(os.environ.get("COZE_MAX_UPLOAD_BYTES", str(64 * 1024 * 1024)))

# 文本类与音频类走不同的消息构造方法，靠扩展名分流。
AUDIO_EXT = {".mp3", ".wav", ".ogg"}
DOC_EXT = {".pdf", ".doc", ".docx", ".txt", ".md"}


def upload_file(coze, file_path: str):
    """上传本地文件，返回文件对象（含 id）。

    先本地校验再发请求：不存在、超限这两类错本地就能判，没必要浪费一次往返。
    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError("文件不存在：%s" % file_path)
    size = path.stat().st_size
    if size == 0:
        raise ValueError("文件是空的：%s" % file_path)
    if size > MAX_UPLOAD_BYTES:
        raise ValueError(
            "文件 %d 字节，超过本地设定的上限 %d 字节。"
            "平台真实上限以官方文档为准，调大 COZE_MAX_UPLOAD_BYTES 即可放行。"
            % (size, MAX_UPLOAD_BYTES)
        )

    uploaded = coze.files.upload(file=path)
    print("上传成功：%s → file_id=%s（%d 字节）" % (path.name, uploaded.id, size))
    return uploaded


def build_message(file_id: str, file_path: str, instruction: str):
    """组装一条多模态用户消息：一个文件 + 一句指令。

    指令不能省：编排提示词靠它判断该走哪条工作流——只丢文件过去，
    等于把行李交给门市顾问却不说要去哪儿。
    """
    from cozepy import Message, MessageObjectString  # type: ignore

    ext = Path(file_path).suffix.lower()
    if ext in AUDIO_EXT:
        file_part = MessageObjectString.build_audio(file_id=file_id)
    elif ext in DOC_EXT:
        file_part = MessageObjectString.build_file(file_id=file_id)
    else:
        raise ValueError("不支持的扩展名 %s；音频用 %s，文档用 %s" % (ext, AUDIO_EXT, DOC_EXT))

    return Message.build_user_question_objects([file_part, MessageObjectString.build_text(instruction)])


def chat_stream(coze, settings: CozeSettings, messages, *, show_reasoning: bool = False) -> str:
    """发起流式对话，按事件消费，返回拼好的正文。

    四类事件必须分开处理：增量正文（追加）、增量思考过程（默认不展示）、
    对话完成（拿 token 统计并退出）、对话失败（别当成「结果为空」）。
    """
    from cozepy import ChatEventType  # type: ignore

    answer_parts: list[str] = []
    reasoning_started = False

    stream = coze.chat.stream(
        bot_id=settings.bot_id,
        user_id=settings.user_id,
        additional_messages=messages,
        parameters={},
    )
    # 日志 ID 在建立流的那一刻就有了，先打出来：
    # 万一后面卡住或超时，至少手上有一个能回后台查的凭据。
    print("logid=%s" % getattr(stream.response, "logid", ""))

    for event in stream:
        if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
            reasoning = getattr(event.message, "reasoning_content", None)
            if reasoning:
                if show_reasoning:
                    if not reasoning_started:
                        reasoning_started = True
                        print("\n----- 思考过程 -----")
                    print(reasoning, end="", flush=True)
                continue
            piece = event.message.content or ""
            answer_parts.append(piece)
            print(piece, end="", flush=True)

        elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
            print("\n----- 完成，token 用量：%s -----" % event.chat.usage.token_count)
            break

        elif event.event == ChatEventType.CONVERSATION_CHAT_FAILED:
            print("\n----- 失败：%s -----" % event.chat.last_error)
            break

    return "".join(answer_parts)


def run(file_path: str, instruction: str, out_path: Optional[str] = None) -> str:
    settings = CozeSettings.from_env(need_bot=True)
    coze = build_sdk_client(settings)
    if coze is None:
        raise RuntimeError("未安装 SDK，先执行 pip install cozepy")

    uploaded = upload_file(coze, file_path)
    messages = [build_message(uploaded.id, file_path, instruction)]
    answer = chat_stream(coze, settings, messages)

    if out_path:
        Path(out_path).parent.mkdir(parents=True, exist_ok=True)
        Path(out_path).write_text(answer, encoding="utf-8")
        print("结果已写入 %s（%d 字）" % (out_path, len(answer)))
    return answer


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("用法：python coze_chat_stream.py <文件路径> <指令> [输出md路径]")
        print("例：python coze_chat_stream.py ./resume.pdf '帮我看看简历' ./out/resume.md")
        sys.exit(2)
    try:
        run(sys.argv[1], sys.argv[2], sys.argv[3] if len(sys.argv) > 3 else None)
    except MissingConfig as exc:
        print("配置缺失：%s" % exc)
        sys.exit(3)

"""工作流直连调用：绕过智能体，直接把一条工作流当成一个远程函数用。

何时直连工作流：输入输出结构化、分支由自己的程序决定，等价于调一个远程函数；
何时走智能体：输入是自然语言、由模型决定走哪条工作流、需要多轮上下文。

直连的两个前提：
1. 工作流必须已经发布（草稿状态调不通）；
2. 开始节点声明的参数名，就是 parameters 里的键名——多一个少一个都不行。
   这正是「只认变量不认上下文」在 API 这一层的体现。
"""
from __future__ import annotations

import json
import sys
from typing import Any, Iterator

from coze_client import CozeError, HttpClient
from coze_settings import PATH_WORKFLOW_RUN, PATH_WORKFLOW_STREAM, CozeSettings, MissingConfig


def run_workflow(http: HttpClient, workflow_id: str, parameters: dict) -> dict:
    """非流式运行：一次请求拿到结束节点的全部返回。

    只适合总耗时可控的工作流；像整段录音分析那类长任务容易顶到网关超时，
    应该改用流式或异步。
    """
    payload = {"workflow_id": workflow_id, "parameters": parameters}
    data = http.post_json(PATH_WORKFLOW_RUN, payload)

    # 结束节点「返回变量」模式下业务结果是一段 JSON 字符串，要再解析一层；
    # 忽略这一层就会拿着字符串当字典用。
    raw = data.get("data")
    if isinstance(raw, str):
        try:
            data["parsed"] = json.loads(raw)
        except json.JSONDecodeError:
            # 结束节点选的是「返回文本」模式，本来就不是 JSON，原样保留。
            data["parsed"] = raw
    else:
        data["parsed"] = raw
    return data


def stream_workflow(http: HttpClient, workflow_id: str, parameters: dict) -> Iterator[tuple]:
    """流式运行：边跑边拿中间输出。工作流里输出节点的文字会先于最终结果到达。"""
    payload = {"workflow_id": workflow_id, "parameters": parameters}
    yield from http.post_stream(PATH_WORKFLOW_STREAM, payload)


def pretty(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, indent=2)


def main() -> int:
    if len(sys.argv) < 2:
        print("用法：python coze_workflow_run.py '<开始节点参数的 JSON>' [--stream]")
        print("例：python coze_workflow_run.py '{\"input\":\"成都\"}'")
        return 2

    try:
        settings = CozeSettings.from_env(need_workflow=True)
    except MissingConfig as exc:
        print("配置缺失：%s" % exc)
        return 3

    try:
        parameters = json.loads(sys.argv[1])
    except json.JSONDecodeError as exc:
        print("第一个参数必须是合法 JSON：%s" % exc)
        return 2
    if not isinstance(parameters, dict):
        print("参数必须是一个对象，键名要和开始节点声明的变量名逐字一致。")
        return 2

    http = HttpClient(settings)
    try:
        if "--stream" in sys.argv:
            for event_name, chunk in stream_workflow(http, settings.workflow_id, parameters):
                print("[%s] %s" % (event_name or "message", pretty(chunk)))
        else:
            result = run_workflow(http, settings.workflow_id, parameters)
            print("logid=%s" % result.get("_logid", ""))
            print(pretty(result.get("parsed")))
    except CozeError as exc:
        print("调用失败：%s" % exc)
        print("先自查：工作流发布了吗？参数名和开始节点一致吗？令牌有这个空间的权限吗？")
        return 4
    return 0


if __name__ == "__main__":
    sys.exit(main())

{
  "workflow": "ai_resume_handler",
  "description": "接收一份简历文件，分块后并行评估各部分，最后整合成一份可读的修改建议",
  "note": "讲结构用的骨架描述，不是平台导入格式。价值在于：把工作流写成「节点 + 边 + 变量契约」三件事，换任何编排平台都成立。",
  "start": { "outputs": [ { "name": "resume", "type": "file", "required": true } ] },
  "nodes": [
    { "id": "extract_text", "type": "plugin", "title": "简历文字提取",
      "inputs": { "url": "{{start.resume}}" },
      "outputs": ["pdf_content", "data", "content"] },

    { "id": "forbid_words", "type": "database", "title": "违禁词查询",
      "op": "select", "table": "forbidden_words", "fields": ["word"],
      "outputs": ["outputList"] },

    { "id": "now", "type": "plugin", "title": "当前时间", "outputs": ["message"] },

    { "id": "split", "type": "llm", "title": "简历内容分块", "model_hint": "长上下文模型",
      "inputs": { "input": "{{extract_text.pdf_content}}", "content": "{{extract_text.content}}" },
      "output_format": "json",
      "prompt_ref": "node_prompts.yaml#split_resume",
      "outputs": ["name", "base_info", "skill", "education", "work_exp", "project_exp", "self", "context"] },

    { "id": "eval_base_info", "type": "llm", "parallel_group": "sections",
      "inputs": { "base_info": "{{split.base_info}}", "context": "{{split.context}}", "date": "{{now.message}}" } },
    { "id": "eval_skill", "type": "llm", "parallel_group": "sections", "model_hint": "推理模型",
      "inputs": { "skill": "{{split.skill}}", "project_exp": "{{split.project_exp}}" } },
    { "id": "eval_education", "type": "llm", "parallel_group": "sections", "skills": ["联网搜索"],
      "inputs": { "education": "{{split.education}}", "date": "{{now.message}}" } },
    { "id": "eval_work", "type": "llm", "parallel_group": "sections", "model_hint": "推理模型",
      "inputs": { "work_exp": "{{split.work_exp}}", "context": "{{split.context}}", "date": "{{now.message}}" } },
    { "id": "eval_self", "type": "llm", "parallel_group": "sections",
      "inputs": { "self": "{{split.self}}", "context": "{{split.context}}" } },

    { "id": "loop_projects", "type": "loop", "loop_type": "array", "array": "{{split.project_exp}}",
      "body": [
        { "id": "eval_project", "type": "llm", "model_hint": "推理模型",
          "inputs": { "project_exp": "{{loop.item}}", "context": "{{split.context}}",
                      "date": "{{now.message}}", "forbid_list": "{{forbid_words.outputList}}" } },
        { "id": "query_rewrite", "type": "llm", "title": "检索 query 优化",
          "inputs": { "project_exp": "{{loop.item}}" }, "outputs": ["query"] },
        { "id": "template_search", "type": "knowledge", "dataset": "resume_templates",
          "inputs": { "query": "{{query_rewrite.query}}" },
          "params": { "top_k": 20, "min_score": 0.85, "rewrite": true, "rerank": true } },
        { "id": "dup_check", "type": "llm", "title": "重复度评估",
          "inputs": { "query": "{{query_rewrite.query}}", "context": "{{template_search.outputList}}" } }
      ],
      "outputs": ["project_results"] },

    { "id": "merge", "type": "llm", "title": "评估结果整合", "model_hint": "推理模型",
      "inputs": { "sections": "以上五个并行分支的 output", "projects": "{{loop_projects.project_results}}" },
      "outputs": ["report"] }
  ],
  "end": { "mode": "text", "answer": "{{merge.report}}", "stream": true },
  "edges": [
    ["start", "extract_text"], ["start", "forbid_words"], ["start", "now"],
    ["extract_text", "split"],
    ["split", "eval_base_info"], ["split", "eval_skill"], ["split", "eval_education"],
    ["split", "eval_work"], ["split", "eval_self"], ["split", "loop_projects"],
    ["forbid_words", "loop_projects"], ["now", "loop_projects"],
    ["eval_base_info", "merge"], ["eval_skill", "merge"], ["eval_education", "merge"],
    ["eval_work", "merge"], ["eval_self", "merge"], ["loop_projects", "merge"],
    ["merge", "end"]
  ],
  "design_notes": [
    "extract_text：传文档时内容落在 pdf_content，传网页地址时落在 data，同一时刻只有一个有值，下游必须同时接住。",
    "forbid_words 用数据库而不是写进提示词：违禁词会持续增补，改一行即生效，不必重新发布工作流和智能体。",
    "now 节点不可省：判断工作年限与毕业时间是否自洽，必须有一个可信的「今天」，模型自己并不知道今天几号。",
    "split 用长上下文模型：原文加分块结果同时占窗口，短窗口会截断，表现为后半段字段莫名为空。",
    "loop_projects：一次把 5 个项目全塞给模型，评估质量明显下滑；拆成 5 次单项目评估，每次上下文短而干净。",
    "query_rewrite：整段原文拿去检索噪声太大，先压成结构化摘要，召回相关性明显提升。",
    "template_search 参数：查重要「宁可多召回再筛」，top_k 放大，同时抬高阈值，避免只沾边的片段也算雷同。",
    "end 用返回文本适合直接展示；若要被别的工作流或程序调用，改成返回变量、输出结构化 JSON。"
  ],
  "invariants": [
    "所有边合起来必须是有向无环图：任何边都不能指回祖先，否则死循环。",
    "每个节点的 inputs 只能引用祖先节点的 outputs；引用不到就说明这条边没连。",
    "并行分支之间不得互相引用变量：它们没有先后关系，读到的值不确定。",
    "整合节点是所有并行分支的汇聚点，必须等所有分支完成才开始执行。"
  ]
}

# 节点提示词模板：每个 LLM 节点一段，带变量契约
#
# 三条约定：
# 1. {{变量名}} 必须与节点输入面板里声明的名字逐字一致。不一致时平台不报错，
#    占位符原样进提示词被当成普通文字——最常见也最难发现的故障。
# 2. 要输出多个字段时，字段名必须在提示词里显式点名。
# 3. 规则型提示词把「规则」和「注意」分开写：规则是硬约束，注意是边界情况。

split_resume:
  purpose: "把一份简历原文拆成结构化的若干块，供后续各评估节点并行使用"
  model_hint: "长上下文模型；简历原文 + 拆分结果会同时占用窗口"
  output_format: json
  inputs:
    input: "文档解析插件的 pdf_content 字段"
    content: "文档解析插件的 content 字段，两者只有一个有值"
  outputs: [name, base_info, skill, education, work_exp, project_exp, self, context]
  system: |
    你是一个大模型和算法工程师简历内容提取助手，能够根据简历中的内容提取出来对应的实体。
    你将接受简历内容作为输入，并按照下列要求进行拆分，并输出为 json 格式：

    姓名：字段名 {{name}}
    基本信息：字段名 {{base_info}}，包含姓名、求职岗位、期望薪资、工作年限、手机号、邮箱等，不含学历与技能
    个人技能：字段名 {{skill}}，个人掌握的 IT 相关技能
    教育经历：字段名 {{education}}，包含学历、在校时间、专业，以及位于简历第几部分
    工作经历：字段名 {{work_exp}}，包含公司、时间、角色、从事内容，以及位于简历第几部分
    项目经验：字段名 {{project_exp}}，每个项目作为数组中的一个元素，数组长度与项目数一致
    个人爱好与评价：字段名 {{self}}
    上下文：字段名 {{context}}，以上所有内容的概括，供后续分块校验时作为参考

    注意：以上要求需要严格遵循。除上下文以外，其他内容不要省略和总结，
    直接拆出原文即可，不要出现错误的划分。
  user: |
    输入的内容为:
    {{input}}
    如果为空，则：
    {{content}}

eval_project:
  purpose: "循环体内部，每次只评估一个项目"
  model_hint: "推理模型"
  output_format: json
  inputs:
    project_exp: "循环变量 item，一个项目的全文"
    context: "简历概括"
    date: "当前时间"
    forbid_list: "违禁词数据库查询结果"
  system: |
    你是一个大模型和算法工程师简历审核大师，专门负责项目经历部分。

    规则：
    1. 项目数量要与工作年限匹配
    2. 工作经历和项目经历要分开写
    3. 每个项目至少包含项目介绍（背景）、个人负责部分、成果数据，分开写
    4. 建议在每个项目前加上技术栈
    5. 项目背景需要合理，要让人觉得这个项目是真实需求驱动的
    6. 实习阶段避免出现过重的措辞
    7. 项目周期需要合理
    8. 禁止出现：{{forbid_list}} 中的词
    9. 成果数据中不好解释的指标，需要提醒用户准备说法
    10. 项目经历不得与上下文冲突

    如果违反规则，列出需要修改的部分，并说明违反了哪一条。
  user: |
    项目经历:
    {{project_exp}}
    上下文：
    {{context}}
    当前时间：
    {{date}}

dedup_check:
  purpose: "拿检索回来的模板片段和本人项目做雷同比对"
  inputs: { query: "优化后的项目摘要", context: "知识库检索结果" }
  system: |
    你是一个大模型和算法工程师简历审核大师，专门负责项目经历部分，
    能够根据用户输入的项目内容和知识库中检索出来的内容进行对比，提示用户重合情况。

    需要严格执行：
    1. 技术点、业务流程、个人职责、优化点、成果数据一致或基本相似的，认为是雷同
    2. 输出时需要指出具体哪些部分雷同，并给出两边的原文对照
  user: |
    项目内容：
    {{query}}
    检索出来的内容：
    {{context}}

split_role:
  purpose: "把一段面试录音转出的文字划分成面试官与应试者两方，并做非面试内容拦截"
  model_hint: "推理模型；文字里已经没有声纹和语气，只能靠语义判断"
  inputs: { input: "语音识别节点的输出文本" }
  system: |
    你是一个语音转文字内容总结助手，专门负责面试内容的整理。
    1. 提取对话内容，以面试官和应试者进行区分，存放到字段 {{result}}，
       按原内容整理，不要省略，每次说话换行。

    需要注意：
    1. 中间有卡顿或表现异常的地方，用方括号括起来并注明情况
    2. 在每句开头注明是面试官还是应试者
    3. 如果判断本次内容不是面试对话，直接输出空字符
  user: |
    语音转文字内容：
    {{input}}
  guard_note: "第 3 条是入口拦截：没有它，任何一段录音都会跑完全部循环与检索，白烧算力。"

qa_extract:
  purpose: "把整段面试对话拆成一个个问答对，供后续循环逐条评估"
  inputs: { qa: "角色划分后的对话原文" }
  system: |
    你是一个面试内容提取助手，能够根据面试官和面试者的对话提取出问题和回答两部分内容，
    格式："问题：" + 换行 + "回答"，并以数组形式返回。

    需要注意：
    1. 每个问答只保留问题和对应回答，省略无关的连接词
    2. 保留回答中的停顿等标注信息，以便后续评估
    3. 不需要提取自我介绍和项目介绍
    4. 问题需要结合上文还原指代。例如先问了编码器，接着问"解码器呢"，
       要把第二个问题改写成完整的问句
  pitfall: "第 4 条必需：指代不还原，检索就会拿着'解码器呢'三个字去查，什么也查不到。"

answer_grade:
  purpose: "逐条评估回答质量，知识库查不到时回落到联网检索"
  model_hint: "推理模型；要判断两段表述是不是同一个意思"
  inputs: { input: "知识库检索结果", query: "问题", answer: "面试者的回答" }
  system: |
    你是一个大模型算法工程师的面试官，能够根据面试问题、用户回答、题库检索结果，
    以及互联网上的答案对回答进行综合评估。

    步骤：
    1. 先判断检索出来的内容是不是这个问题的答案；不是则转为联网检索；是则以它为准
    2. 对比问题和回答，判断回答质量

    需要注意：
    1. 给出 A/B/C/D 四级评价，并说明理由
    2. 回答不好的问题，要告诉用户正确答案
    3. 表达混乱的，要指出表达方式的问题
    4. 回答跑题的，要指出需要先理解问题再回答
  user: |
    问题：
    {{query}}
    用户回答：
    {{answer}}
    题库检索结果：
    {{input}}

# 智能体编排模板：把「人设 + 工作流路由 + 兜底话术」写成一份可评审的文件
#
# 为什么落成文件：平台的编排框改了就覆盖、没有历史；放进版本管理才能回答
# 「上周还好好的，谁改了什么」。定稿后原样贴回人设与回复逻辑面板即可。
#
# 用法：把所有 TODO 换成自己的业务内容；标注 KEEP 的结构不要动。

agent:
  name: "TODO：智能体名称"
  intro: "TODO：一句话说明它能干什么，会显示在列表和商店里"

  # KEEP：单 Agent 自主规划 与 Multi-Agent 的取舍写在这里，评审时一眼能看到理由
  mode: single_agent_planning
  mode_rationale: >
    场景少、彼此容易分辨，但每个场景内部的数据处理很复杂——这种形状用单 Agent
    自主规划最合适：每条工作流是一个只干一件事的下属，模型只负责「这一单交给谁」。
    场景多、且每个场景要维持自己的多轮人格时，才值得上 Multi-Agent。

  # KEEP：长期记忆按需开启。开了会记住用户偏好，也意味着上次会话会影响这次，调试时先想到它。
  long_term_memory: false

# ---------------------------------------------------------------------------
# 人设与回复逻辑：贴进平台的提示词框
# 结构固定为六段。少任何一段，模型的行为都会开始飘。
# ---------------------------------------------------------------------------
persona: |
  # 角色：
  TODO：一句话定义身份，例如「大模型算法工程师面试助手」

  ## 目标：
  TODO：说明它要替用户解决什么问题

  ## 技能：
  1. TODO：技能一，对应一条工作流
  2. TODO：技能二，对应一条工作流
  3. TODO：技能三，对应一条工作流
  4. 如果不属于以上场景，不要发挥。返回固定话术：
     TODO：兜底话术，告诉用户支持什么、该怎么做

  ## 工作流：
  1. 如果 TODO（触发条件一），执行 TODO（工作流英文名一）
  2. 如果 TODO（触发条件二），执行 TODO（工作流英文名二）
  3. 如果 TODO（触发条件三）；若缺少必要输入，不要执行任何工作流，
     直接提示用户补齐再来

  ## 输出格式：
  TODO：无特定格式 / Markdown / 固定小节

  ## 限制：
  - 多给用户鼓励而不是打击
  - 对用户屏蔽调用工作流等内部细节
  - TODO：本业务特有的限制

# ---------------------------------------------------------------------------
# 工作流清单：名称与描述是路由的唯一依据
# ---------------------------------------------------------------------------
workflows:
  - name: "TODO_workflow_one"          # 英文 + 数字 + 下划线；这个名字会出现在提示词里
    description: >
      TODO：写清「什么时候该调它」，而不是「它内部怎么实现」。
      这段描述是模型选择工作流的依据，写成实现细节等于没写。
    inputs:
      - { name: "TODO_param", type: "string", required: true }
    published: true                     # KEEP：只有发布过的工作流才能被智能体调用

  - name: "TODO_workflow_two"
    description: "TODO"
    inputs:
      - { name: "TODO_param", type: "file", required: true }
    published: true

# ---------------------------------------------------------------------------
# 开场白与预置问题：降低用户的启动成本
# ---------------------------------------------------------------------------
opening:
  greeting: |
    TODO：一句自我介绍 + 支持哪几件事 + 操作三步走
    操作流程：
    1. TODO
    2. TODO
    3. 等待结果返回
  suggestions:
    - "TODO：预置问题一（上传文件后点击）"
    - "TODO：预置问题二（上传文件后点击）"
    - "TODO：预置问题三（上传文件后点击）"

# ---------------------------------------------------------------------------
# 发布：渠道决定了它能被谁用、怎么用
# ---------------------------------------------------------------------------
publish:
  channels:
    - store                # 平台内的商店，给人直接聊
    - api_sdk              # KEEP：要在自己的程序里调用，这一项必须勾上，否则 bot_id 调不通
  checklist:
    - "所有引用的工作流都已单独试运行通过"
    - "所有引用的工作流都已发布，不是草稿状态"
    - "数据库的线上环境数据已导入，不能只有测试环境有数据"
    - "知识库内容已完成处理，不是还在解析中"
    - "兜底话术实测过一次：随便说句闲聊，看它是不是老老实实拒绝"
    - "发布记录写清这一版改了什么，出问题时才知道回滚到哪一版"

# ---------------------------------------------------------------------------
# 上线后的观测项：不看这些，出了问题只能靠用户投诉才知道
# ---------------------------------------------------------------------------
observability:
  - "调用量与失败率：失败率突然抬头，先查是不是某条工作流发布了新版本"
  - "单次耗时分布：含循环的工作流耗时长尾很长，要给前端配合适的等待提示"
  - "token 消耗趋势：循环 + 长上下文模型是消耗大户，突增往往意味着有人传了超长输入"
  - "兜底话术触发率：太高说明路由描述写得不好，用户的正常请求被挡在门外了"

"""批处理：把已发布的智能体当成一个命令行工具，扫一个目录跑完。

交互逻辑留在平台上（工作流、提示词、知识库随时改），调度逻辑留在代码里
（遍历、限速、重试、落盘、断点续跑）。三类任务共用同一套骨架，区别只有
收哪些扩展名、发哪句指令，所以用一张表集中差异，不复制三份函数。
"""
from __future__ import annotations

import argparse
import sys
import time
from pathlib import Path

from coze_chat_stream import build_message, chat_stream, upload_file
from coze_client import build_sdk_client
from coze_retry import RateLimiter, call_with_retry
from coze_settings import CozeSettings, MissingConfig

# 任务名 → (可处理的扩展名, 发给智能体的指令, 输出子目录)
TASKS = {
    "resume": ({".pdf", ".doc", ".docx"}, "帮我看看简历", "resume_evaluated"),
    "audio": ({".mp3", ".wav", ".ogg"}, "帮我分析一下面试录音", "audio_evaluated"),
    "question": ({".pdf", ".doc", ".docx"}, "帮我生成一下面试题", "question_generated"),
}


def collect_files(root: Path, extensions: set) -> list:
    """递归收集待处理文件；顺序稳定，续跑才接得上。"""
    return [p for p in sorted(root.rglob("*")) if p.is_file() and p.suffix.lower() in extensions]


def handle_one(coze, settings: CozeSettings, path: Path, instruction: str, out_dir: Path) -> Path:
    """单文件：上传 → 对话 → 落盘，返回结果路径。"""
    uploaded = upload_file(coze, str(path))
    messages = [build_message(uploaded.id, str(path), instruction)]
    answer = chat_stream(coze, settings, messages)

    if not answer.strip():
        # 空结果不能当成功：多半是编排提示词没命中这个场景，或工作流中途失败。
        raise RuntimeError("返回内容为空，检查智能体编排提示词是否覆盖了这个指令")

    out_dir.mkdir(parents=True, exist_ok=True)
    out_file = out_dir / (path.stem + ".md")
    out_file.write_text(answer, encoding="utf-8")
    return out_file


def run_batch(task: str, directory: str, *, interval: float, attempts: int, resume: bool) -> int:
    if task not in TASKS:
        print("未知任务 %s，可选：%s" % (task, "、".join(TASKS)))
        return 2
    extensions, instruction, out_name = TASKS[task]

    root = Path(directory)
    if not root.is_dir():
        print("不是一个目录：%s" % directory)
        return 2

    try:
        settings = CozeSettings.from_env(need_bot=True)
    except MissingConfig as exc:
        print("配置缺失：%s" % exc)
        return 3
    coze = build_sdk_client(settings)
    if coze is None:
        print("未安装 SDK，先执行 pip install cozepy")
        return 3

    files = collect_files(root, extensions)
    out_dir = root / out_name
    if resume:
        done = {p.stem for p in out_dir.glob("*.md")} if out_dir.exists() else set()
        skipped = [f for f in files if f.stem in done]
        files = [f for f in files if f.stem not in done]
        print("续跑：跳过已完成 %d 个" % len(skipped))

    print("任务=%s  待处理 %d 个文件  输出到 %s" % (task, len(files), out_dir))
    limiter = RateLimiter(interval)
    started = time.time()
    ok, failed = 0, []

    for index, path in enumerate(files, start=1):
        print("\n[%d/%d] %s" % (index, len(files), path))
        limiter.wait()
        try:
            out_file = call_with_retry(
                lambda p=path: handle_one(coze, settings, p, instruction, out_dir),
                attempts=attempts,
            )
            ok += 1
            print("完成 → %s" % out_file)
        except Exception as exc:  # noqa: BLE001 - 单个文件失败不能中断整批
            failed.append((str(path), "%s: %s" % (type(exc).__name__, exc)))
            print("失败：%s" % exc)

    elapsed = time.time() - started
    print("\n" + "=" * 52)
    print("总计 %d　成功 %d　失败 %d　耗时 %.1fs" % (len(files), ok, len(failed), elapsed))
    for name, reason in failed:
        print("  失败：%s —— %s" % (name, reason))
    print("=" * 52)
    # 失败清单单独落盘，方便只重跑这一批
    if failed:
        report = out_dir / "_failed.txt"
        out_dir.mkdir(parents=True, exist_ok=True)
        report.write_text("\n".join("%s\t%s" % item for item in failed), encoding="utf-8")
        print("失败清单：%s" % report)
    return 0 if not failed else 1


def main() -> int:
    parser = argparse.ArgumentParser(description="批量调用智能体处理一个目录")
    parser.add_argument("task", choices=sorted(TASKS), help="任务")
    parser.add_argument("directory", help="待处理目录")
    parser.add_argument("--interval", type=float, default=1.0, help="最小间隔秒")
    parser.add_argument("--attempts", type=int, default=3, help="最多尝试次数")
    parser.add_argument("--no-resume", action="store_true", help="重跑")
    args = parser.parse_args()
    return run_batch(
        args.task,
        args.directory,
        interval=args.interval,
        attempts=args.attempts,
        resume=not args.no_resume,
    )


if __name__ == "__main__":
    sys.exit(main())

"""错误处理与重试：把「偶发抖动」和「配置写错了」分开对待。

批量跑 100 份简历时，最耗人的不是跑不通，而是「跑了 40 份挂了」。原因分两类，
处理方式完全相反：可重试（网络抖动、5xx、限流、超时）等一会儿再来一次大概率就好；
不可重试（令牌过期、工作流没发布、参数名写错、格式不支持）重试一万次还是错，
只会烧额度、刷日志。核心不是「重试几次」，而是「哪些错该重试」。
"""
from __future__ import annotations

import random
import time
from typing import Callable, Iterable, TypeVar

from coze_client import CozeError

T = TypeVar("T")

# 这些 HTTP 状态码代表服务端侧的临时问题，值得退避后重来。
RETRIABLE_HTTP = {408, 425, 429, 500, 502, 503, 504}

# 这些异常类型代表本地环境或调用方式的问题，重试没有意义。
FATAL_EXC = (FileNotFoundError, ValueError, TypeError, KeyError)


def is_retriable(exc: BaseException) -> bool:
    """判断一个异常值不值得重试。"""
    if isinstance(exc, FATAL_EXC):
        return False
    if isinstance(exc, CozeError):
        code = exc.code
        if isinstance(code, int):
            # 4xx 里只有限流和请求超时值得重试，其余都是「你写错了」。
            if code in RETRIABLE_HTTP:
                return True
            if 400 <= code < 500:
                return False
        # 业务错误码拿不准时，保守地不重试：宁可报错让人看一眼，
        # 也不要闷头重试把配额烧掉。
        return False
    if isinstance(exc, TimeoutError):
        return True
    # 网络层异常（连接重置、DNS 抖动）通常是 OSError 的子类。
    return isinstance(exc, OSError)


def backoff_delays(attempts: int, base: float = 1.0, cap: float = 30.0) -> Iterable[float]:
    """指数退避 + 抖动。

    指数退避：1s、2s、4s、8s……给下游一个恢复的窗口。
    抖动：如果 100 个并发任务同时失败、又同时在第 1 秒重试，
    就会形成一波整齐的二次冲击。加一点随机偏移把它们打散。
    """
    for i in range(attempts):
        delay = min(cap, base * (2 ** i))
        yield delay * (0.5 + random.random() * 0.5)


def call_with_retry(
    func: Callable[[], T],
    *,
    attempts: int = 3,
    base: float = 1.0,
    on_retry: Callable[[int, BaseException, float], None] | None = None,
) -> T:
    """执行 func，失败时按策略重试。成功返回结果，最终失败抛出最后一个异常。

    attempts 是总次数不是额外次数：attempts=3 表示最多跑 3 次。
    """
    last_exc: BaseException | None = None
    delays = list(backoff_delays(attempts, base=base))

    for attempt in range(1, attempts + 1):
        try:
            return func()
        except BaseException as exc:  # noqa: BLE001 - 这里就是要统一分类
            last_exc = exc
            if attempt >= attempts or not is_retriable(exc):
                raise
            delay = delays[attempt - 1]
            if on_retry:
                on_retry(attempt, exc, delay)
            else:
                print("第 %d 次失败（%s），%.1fs 后重试" % (attempt, exc, delay))
            time.sleep(delay)

    assert last_exc is not None
    raise last_exc


class RateLimiter:
    """最简单的令牌间隔限速器：保证两次调用之间至少隔 interval 秒。

    批处理时主动限速，比被平台限流后再退避更划算：
    被限流会浪费一次请求、一次等待，还可能影响同空间的其他任务。
    具体该设多少，取决于账号的并发与调用限制——去平台的用量页面查，
    别照抄别人的数字。
    """

    def __init__(self, interval: float = 0.0):
        self.interval = interval
        self._last = 0.0

    def wait(self) -> None:
        if self.interval <= 0:
            return
        now = time.monotonic()
        gap = now - self._last
        if gap < self.interval:
            time.sleep(self.interval - gap)
        self._last = time.monotonic()


if __name__ == "__main__":
    # 一个自包含的演示：前两次抛可重试错误，第三次成功。
    state = {"n": 0}

    def flaky() -> str:
        state["n"] += 1
        if state["n"] < 3:
            raise CozeError("网关抖动", code=502)
        return "第 %d 次成功" % state["n"]

    print(call_with_retry(flaky, attempts=4, base=0.05))

    def broken() -> str:
        raise CozeError("工作流未发布", code=400)

    try:
        call_with_retry(broken, attempts=4, base=0.05)
    except CozeError as exc:
        print("不可重试，立即失败：%s" % exc)

"""把平台当后端：一个最小的 FastAPI 封装，前端只认自己的域名。

不让前端直连平台：令牌会暴露、没法鉴权与配额、没法换供应商。
正确形状：浏览器 → 自己的后端（带登录态）→ 平台 OpenAPI。

运行：
    pip install fastapi uvicorn cozepy
    uvicorn coze_backend:app --host 127.0.0.1 --port 8000
"""
from __future__ import annotations

import json
import tempfile
from pathlib import Path
from typing import Iterator

from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, UploadFile
from fastapi.responses import StreamingResponse

from coze_chat_stream import build_message, upload_file
from coze_client import build_sdk_client
from coze_settings import CozeSettings

app = FastAPI(title="面试助手网关", version="1.0")

# 进程启动时装配一次，失败就直接起不来——比运行到一半才发现配置错要好。
SETTINGS = CozeSettings.from_env(need_bot=True)
COZE = build_sdk_client(SETTINGS)

# 允许的上传类型。白名单而不是黑名单：没想到的格式默认拒绝。
ALLOWED_SUFFIX = {".pdf", ".doc", ".docx", ".mp3", ".wav", ".ogg"}
MAX_BYTES = 32 * 1024 * 1024


def current_user(authorization: str = Header(default="")) -> str:
    """登录态 → 用户标识。平台侧 user_id 用自己的业务 ID，日志、配额、审计才对得上。"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="未登录")
    token = authorization[len("Bearer "):].strip()
    if not token:
        raise HTTPException(status_code=401, detail="未登录")
    # TODO: 换成真实的会话校验，返回业务用户 ID
    return "user-" + token[:8]


def save_upload(upload: UploadFile) -> Path:
    """把上传流落到临时文件，顺便做大小与类型校验。"""
    suffix = Path(upload.filename or "").suffix.lower()
    if suffix not in ALLOWED_SUFFIX:
        raise HTTPException(status_code=415, detail="不支持的文件类型 %s" % suffix)

    tmp = Path(tempfile.mkdtemp()) / (Path(upload.filename or "upload").name)
    size = 0
    with tmp.open("wb") as fh:
        while True:
            chunk = upload.file.read(1024 * 256)
            if not chunk:
                break
            size += len(chunk)
            if size > MAX_BYTES:
                raise HTTPException(status_code=413, detail="文件过大")
            fh.write(chunk)
    return tmp


def sse(payload: dict) -> str:
    """按 SSE 格式打包一条消息。"""
    return "data: %s\n\n" % json.dumps(payload, ensure_ascii=False)


def relay(file_path: Path, instruction: str, user_id: str) -> Iterator[str]:
    """把平台的事件流翻译成自己的事件流，中间加一层自己的协议。"""
    from cozepy import ChatEventType  # type: ignore

    try:
        uploaded = upload_file(COZE, str(file_path))
        messages = [build_message(uploaded.id, str(file_path), instruction)]
        stream = COZE.chat.stream(
            bot_id=SETTINGS.bot_id,
            user_id=user_id,
            additional_messages=messages,
            parameters={},
        )
        yield sse({"type": "start", "logid": getattr(stream.response, "logid", "")})

        for event in stream:
            if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
                if getattr(event.message, "reasoning_content", None):
                    continue          # 思考过程不外发给终端用户
                yield sse({"type": "delta", "text": event.message.content or ""})
            elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
                yield sse({"type": "done", "tokens": event.chat.usage.token_count})
                break
            elif event.event == ChatEventType.CONVERSATION_CHAT_FAILED:
                yield sse({"type": "error", "message": str(event.chat.last_error)})
                break
    except Exception as exc:  # noqa: BLE001 - 流里不能抛异常，只能作为事件发出
        yield sse({"type": "error", "message": "%s: %s" % (type(exc).__name__, exc)})
    finally:
        # 临时文件用完即删，避免隐私数据堆在磁盘上。
        try:
            file_path.unlink(missing_ok=True)
        except OSError:
            pass


@app.post("/api/analyze")
def analyze(
    file: UploadFile = File(...),
    instruction: str = Form("帮我看看简历"),
    user_id: str = Depends(current_user),
):
    """上传一个文件并流式拿回分析结果。"""
    saved = save_upload(file)
    return StreamingResponse(
        relay(saved, instruction, user_id),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.get("/api/health")
def health() -> dict:
    """健康检查。只暴露「配好了没有」，绝不回显令牌本身。"""
    return {"ok": COZE is not None, "bot_configured": bool(SETTINGS.bot_id)}

"""会话、消息、对话、上下文段落：四个对象的关系，以及多轮怎么接。

平台把一次交互拆成四个对象，名字很像，职责完全不同：

- 会话 Conversation：一段问答交互的容器，装着消息，自动处理截断。
- 消息 Message：用户或智能体创建的一条内容，可以是文本、图片或文件。
- 对话 Chat：在某个会话里对智能体的一次调用，会往会话里追加消息。
- 上下文段落 Section：会话内部的独立段落。用户点「清除上下文」时系统开一个
  新 Section，新对话不再受之前消息影响。

一句话记：会话是账本，消息是流水，对话是一次记账动作，段落是账本里的分页。

多轮对话怎么接：不要自己在本地拼历史消息塞进请求里——
把 conversation_id 传过去，平台会把这个会话的历史当上下文送给模型；自己拼既容易漏，也容易撑爆。
"""
from __future__ import annotations

import sys

from coze_client import build_sdk_client
from coze_settings import CozeSettings, MissingConfig


def create_conversation(coze):
    """新建会话。一个用户的一个业务场景对应一个会话。"""
    conversation = coze.conversations.create()
    print("会话已创建：conversation_id=%s" % conversation.id)
    return conversation


def ask_in_conversation(coze, settings: CozeSettings, conversation_id: str, text: str) -> str:
    """在指定会话里发起一次流式对话，返回拼好的正文。

    与 coze_chat_stream.py 的区别只有一个参数：多传了 conversation_id。
    多了这个参数，模型就看得见这个会话里之前说过的话。
    """
    from cozepy import ChatEventType, Message  # type: ignore

    parts: list[str] = []
    stream = coze.chat.stream(
        bot_id=settings.bot_id,
        user_id=settings.user_id,
        conversation_id=conversation_id,
        additional_messages=[Message.build_user_question_text(text)],
    )
    for event in stream:
        if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
            if getattr(event.message, "reasoning_content", None):
                continue
            piece = event.message.content or ""
            parts.append(piece)
            print(piece, end="", flush=True)
        elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
            print("\n[token 用量 %s]" % event.chat.usage.token_count)
            break
        elif event.event == ChatEventType.CONVERSATION_CHAT_FAILED:
            print("\n[失败 %s]" % event.chat.last_error)
            break
    return "".join(parts)


def demo_multi_turn() -> int:
    """两轮对话的最小演示：第二轮只说「那它呢」，靠会话上下文还原指代。"""
    try:
        settings = CozeSettings.from_env(need_bot=True)
    except MissingConfig as exc:
        print("配置缺失：%s" % exc)
        return 3

    coze = build_sdk_client(settings)
    if coze is None:
        print("未安装 SDK，先执行 pip install cozepy")
        return 3

    conversation = create_conversation(coze)

    print("\n第一轮 >>> 帮我生成面试题需要准备什么？")
    ask_in_conversation(coze, settings, conversation.id, "帮我生成面试题需要准备什么？")

    # 第二轮故意说得很省略。如果不传 conversation_id，模型无从知道「它」指什么；
    # 传了，平台会把第一轮的问答一起送进去，指代就能解析。
    print("\n第二轮 >>> 那录音分析呢？")
    ask_in_conversation(coze, settings, conversation.id, "那录音分析呢？")
    return 0


if __name__ == "__main__":
    sys.exit(demo_multi_turn())