Vivix 接入时，如果需要同时支撑“持续对话/流式输出”和“一次性提交一批任务”这两类场景，通常应拆成两条链路分别验证，而不是共用一套参数和一套检查清单。判断入口很简单：看单次会话的持续时间、是否需要跨轮次保留上下文、单次提交的条目规模，以及结果是增量返回还是一次性回收。如果把实时会话和批量请求放在同一套超时、重试和日志里验证，常见的结果就是实时侧断线后找不回上下文、批量侧重试把同一批任务提交两遍。

业务需要跨轮次上下文、持续接收增量输出，就先按实时会话链路验证连接保持、心跳和断线重连；业务是一次性提交大批条目、只关心最终结果，就按批量链路验证分批提交、结果回收和完成判定。两条链路的失败重试与幂等要分开做，最后用同一批样本比对输出是否一致。边界在于：心跳间隔、单批条数、超时和重试次数都需结合 Vivix 侧的实际配置与本机网络环境确认，示例值只能当起点，不能直接照搬。

按会话时长和提交方式把需求分成两类

划分标准建议落在三个可观察的维度上：单次交互持续多久、一次提交涉及多少条目、是否必须连续保留上下文。三项里只要“需要连续上下文”成立，基本就归到实时会话链路；如果一次提交几百上千条、彼此独立、只等最终汇总结果，则归到批量链路。

• 单次时长：几十秒内结束、以轮次为单位的，偏实时；单次提交后要等几分钟甚至更久才有结果的，偏批量。
• 并发量：同时保持多条会话（每条都要独立连接和上下文）的，偏实时；并发体现为“同时在跑的批次数量”的，偏批量。
• 连续上下文：后续轮次必须引用前面内容的，归实时；每条任务自带完整输入、前后不依赖的，归批量。

归类结论写进接入文档时，最好连“用哪套超时、用哪套重试、日志里打哪个 trace 字段”一起确定，避免后面两套代码互相抄参数。判断不清的混合场景（例如批量任务里每条又要多轮追问），建议拆成两个内部模块，各自按对应链路验证，不要做成一个既要长连接又要分批提交的中间层。

实时会话链路验证连接保持与中断重连

实时链路要验证两件事：连接在空闲期是否被对端或中间设备掐断，以及断开后重连能不能接着上一轮的上下文继续。心跳配置通常放在客户端初始化处，示例仅示意字段，具体键名需按实际接入方式替换：

# 客户端连接参数示例（字段名按实际 SDK 或网关配置替换）
connect_timeout_ms: 10000
heartbeat_interval_ms: 20000      # 小于中间设备的空闲断链阈值
heartbeat_timeout_ms: 60000       # 连续丢失心跳后判定为断开
reconnect_max_attempts: 5
reconnect_backoff_ms: 1000        # 起始退避，逐次放大
session_resume: true              # 若接入侧支持续接会话

验证方式：在心跳间隔内不做任何业务请求，观察日志里是否出现心跳超时或连接被重置；再用防火墙或本地代理制造一次短时中断，看客户端是否进入重连流程并恢复。重连骨架的关键是把“会话标识”和“已确认的输出位置”持久化，而不是重连后从头再问一遍：

def run_realtime(session_id, store):
    state = store.load(session_id) or {"cursor": 0, "turns": []}
    for attempt in range(MAX_ATTEMPTS):
        try:
            conn = connect(session_id=session_id, resume_from=state["cursor"])
            for chunk in conn.stream():
                state["cursor"] = chunk.seq          # 记录已消费位置
                state["turns"].append(chunk.text)
                store.save(session_id, state)
            return state
        except (Disconnected, HeartbeatTimeout) as e:
            log.warning("reconnect session=%s cursor=%s err=%s",
                        session_id, state["cursor"], e)
            sleep(backoff(attempt))
    raise RealtimeAborted(session_id, state["cursor"])

恢复后的状态记录至少包含：会话标识、最后确认的序号或游标、已收到的内容摘要、重连次数与最后一次错误类型。检查是否“没丢内容”的办法是对比重连前后的游标是否连续；如果接入侧不支持按游标续接，就要接受“该轮需要重放”的边界，并在业务层把重放的那一轮标记出来，避免用户看到重复片段。

批量链路验证分批提交与结果回收

批量链路的失败大多不是请求发不出去，而是一批太大被截断、或者提交成功但结果没收全。分批参数建议显式配置，不要依赖默认值：

batch_size: 100            # 单批条目数，按单条体积和超时反推
batch_timeout_ms: 120000   # 单批提交到可查询结果的总时限
max_inflight_batches: 2    # 同时在跑的批次数，避免互相挤占
poll_interval_ms: 3000     # 结果回收轮询间隔
poll_max_wait_ms: 600000   # 单批回收等待上限

结果回收一般有两种方式：提交后拿回任务标识再轮询查询，或由接入侧推送完成事件。轮询方式更容易在本地排查，骨架如下：

for batch in split(items, batch_size):
    job = submit(batch)
    deadline = now() + poll_max_wait_ms
    while now() < deadline:
        r = query(job.id)
        if r.status in ("done", "failed"):
            record(job.id, r.status, r.items)
            break
        sleep(poll_interval_ms)
    else:
        record(job.id, "timeout", [])   # 超时也要落表，便于补跑

完成判定不能只看“请求返回成功”，建议同时满足三条：状态字段为终态、回收到的条目数与提交条目数一致、每条都能按业务主键对上号。任意一条不满足，就按未完成处理并进入补跑，而不是直接标记完成。日志里按批次标识 + 条目序号打点，后面比对和补跑都靠这两列定位。

分别补上失败重试与幂等处理

重试策略两条链路要分开定。实时链路的重试单位是“连接”，重点是重连次数与退避；批量链路的重试单位是“批次或单条”，重点是别把已经成功的那部分再提交一次。通用做法是指数退避加随机抖动，并给总时长设上限：

delay = min(base_ms * (2 ** attempt), cap_ms)
delay = delay * (0.5 + random.random() * 0.5)   # 抖动，避免同步重试

重试次数建议只对可恢复错误（超时、连接重置、限流类返回）生效；参数错误、校验失败这类直接落失败，重试没有意义。幂等键的生成建议由业务主键拼出来，稳定且可复算，例如 hash(business_id + ":" + batch_no + ":" + item_index)；存放位置有两处需要同时落：请求侧放进请求头或提交字段随请求带上，本地侧写入任务表并建唯一索引。补跑前先按幂等键查本地表，已成功则跳过，未完成才重新提交。验证方式是故意让某一批在提交后超时，再触发补跑，检查对端结果与本地记录是否都只有一条。

用同一批样本比对两条链路的输出一致性

两条链路分别跑通之后，还要确认链路差异没有导致结果差异。样本量不必大，但要有覆盖度：建议取几十条，包含短输入、接近长度上限的输入、含特殊字符或空字段的输入，以及一条刻意会失败的异常样本。这批样本在实时链路里按单轮提问跑一遍，在批量链路里按同样顺序分批提交跑一遍。

比对字段建议至少包括：结果正文（做空白归一化后比较）、状态值、条目数、按业务主键对齐后的顺序是否一致。不一致时的排查方向按顺序来：先看批量侧是否存在截断导致条目缺失，再看超时或重试是否让某条被处理了两次，然后看实时侧是否因断线重连丢了部分增量内容，最后才去比对参数差异（超时、单批条数、是否有额外截断长度）。把每次比对的样本标识、两条链路的结果摘要和差异字段写进同一份记录，后续换配置时可以直接复跑这批评测，不用重新设计用例。