☰
实时AI的隐形战场:文本工作流与RelayRouter路由解析
2026/10/28 16:41:22 网站建设 项目流程

Gemini Live Avatar 上线那阵子,我朋友圈里刷屏的不是技术评测,而是"实时 AI 时代来了"的感叹。大家只觉得那个能跟你面对面聊天、表情跟得上语气的虚拟形象足够震撼,我却更关注它底下那条看不见的文本工作流,以及 RelayRouter 这类路由组件在其中的位置。今天这篇文章不打算复刻 Avatar,而是想借这个热点,把"实时 AI 不只是把聊天变成视频"这件事讲透:视频化只是门面,真正的战场在文本管道。如果你正在做 AI 应用,或者想把单模型 Demo 升级成多模型生产系统,这篇值得收藏。

1. 大家都在聊视频,而我关注的是那根看不见的管道

1.1 Gemini Live Avatar 到底做对了什么

Gemini Live Avatar 是 Google 在 Gemini Live 基础上加了一路实时视频形象的对话形态。用户对着手机说话,屏幕上会出现一个虚拟人的特写,它会听你说话、看着你、在合适的时机点头,然后以接近真人对话的节奏出声回答。从产品体验看,它确实把"和 AI 对话"从冷冰冰的文字聊天、语音聊天,拉到了视频通话的维度,让人天然觉得"这才是真正的实时 AI"。

但如果你拆开它的技术链路,会发现 Avatar 只是最外层的那层壳。一次对话的完整路径大致是这样:麦克风采集音频 → 语音识别转成文本 → 文本进入大模型推理(结合历史上下文、工具调用)→ 模型流式吐出回复文本 → 文本交给语音合成 → 合成音频驱动虚拟人渲染表情和口型。仔细看,中间一大段全是文本在处理:识别出来的是文本,推理进去的是文本,吐出来的也是文本。用户感知到的"实时",本质是这条文本管道跑得足够快、足够稳,而不是那张脸渲染得够不够生动。

Gemini Live Avatar 做对的事情,是把这条链路整合得足够顺:语音接入延迟低,模型推理速度快,流式返回节奏接近真人对话。它证明了一个事实——实时 AI 是系统工程,不是某个单点能力。视频头像做得再逼真,如果文本链路慢了 2 秒、答非所问、上下文接不上,体验照样崩。所以研究实时 AI,不能只看前台界面,得盯着那条看不见的管道。

1.2 别被"视频化"带偏:文本工作流才是主战场

很多人看到 Gemini Live Avatar 的第一反应是:实时 AI = 把聊天变成视频。这个判断对产品方向有一定参考价值,但对工程架构是一种误导。视频化解决的是"呈现层"的问题,它让 AI 更像人,但并没有降低文本处理上的难度,反而把压力推到了极限。异步聊天场景里,用户愿意等几十秒;实时对话场景里,用户期望的是 1 秒内给出感知,这种差异直接改变了整个架构设计的前提。

我见过不少团队,看到这类产品后先买视频渲染方案、做一个会动的数字人,接上大模型才发现真正的坑全在文本链路:多轮上下文怎么维护、请求到底打到哪个模型、模型响应慢了要不要切换、流式输出怎么转发给前端、突然报错怎么兜底。这些问题,业界其实已经有一批被反复验证的解法,核心就是在客户端和大模型之间加一个智能化的路由层。我习惯把这类组件统称为 RelayRouter。名字不是重点,重点是你得理解它的位置和职责。

所以这篇内容真正想聊的,是实时 AI 背后那套文本工作流:它长什么样,RelayRouter 站在哪里,能解决哪些问题、不能解决哪些问题。适合正在做 AI 应用、想把单模型聊天接成多模型生产级系统的工程师参考。

2. 文本工作流:实时 AI 的地基

2.1 先定义清楚:什么是文本工作流

"文本工作流"听起来玄,其实就是一条从输入文本到输出文本的处理流水线。拆开看,典型环节包括这么几个:

  1. 输入处理:把用户的提问做清洗、拼接系统提示词、识别意图和关键实体。
  2. 上下文装配:从会话存储或向量库里捞历史记录、相关知识片段,组装成最终送进模型的 prompt。
  3. 模型调用决策:选哪个模型、用多大温度、允许多少预算、要不要并行调用。
  4. 生成与流式返回:拿到模型输出,按流式协议传给前端,必要时做输出格式修正。
  5. 工具与函数调用:模型需要查数据、调接口时,工作流要暂停生成、发起外部调用、把结果喂回模型继续生成。
  6. 事后处理:记录 token 用量、埋点监控、缓存重复结果。

很多教程只讲第 3 步"调用模型",好像拿着 SDK 发一个请求就完事。但真实应用里,第 2、3、5 步才是决定体验和成本的关键。一个文本工作流做得好的系统,可以做到:用户问一句天气,系统自动接上定位工具、查完再回答;用户追问一句"明天呢",系统能把上下文接上,不重复追问。这些能力跟界面是视频还是文字毫无关系,全是文本管道里的功夫。

2.2 实时文本流的三道坎

实时场景下,文本工作流比普通聊天难在哪?我总结为三道坎。

第一道坎是延迟预算。用户对"实时"的感知有硬指标:首字延迟(TTFT,Time To First Token)最好在 1 秒以内,超过 2 秒人就会觉得卡;后续吐字速度要稳定在每秒 30~50 个 token,忽快忽慢的流式输出同样让人难受。这意味着工作流的每个环节都要控延迟——上下文装配不能慢慢查库,模型调用不能排队太久,流式转发不能中途积压。任何一个环节成为瓶颈,用户端表现出来的都是同一个词:卡。

第二道坎是上下文连续性。实时对话天然是多轮、长会话的,上一句话的背景要到下一句仍然有效,会话记忆不能丢。这中间有一个很实际的矛盾:上下文越长,模型响应越慢、成本越高;上下文太短,又答不对。工作流必须做取舍,比如自动摘要压缩历史,或者把关键信息抽出来放到系统提示词里。这是纯文本层面的处理,和画面表现无关。

第三道坎是协调多模型。生产级系统几乎不会只接一个大模型。常见的做法是:简单的闲聊用便宜的小模型,复杂的逻辑推理交给旗舰模型,摘要用专门的模型,另外还要穿插向量检索、数据库、业务 API 等外部动作。谁来安排这些模型和工具的调用顺序?谁负责在模型失败时切换备用方案?这就是路由层的职责。

2.3 为什么应用不能直连模型

不搞技术的人常问:我直接在前端调大模型 API 不行吗?单机 Demo 当然可以,要上生产,问题马上冒出来。

安全上,模型 API 的密钥不该出现在客户端,所有调用要收敛到服务端;成本上,不同模型的定价差距很大,同一个问题问旗舰模型和问小模型,费用可能差几十倍,需要按规则分流域;稳定性上,模型服务会有限流和偶发故障,应用要能自动重试和降级,而不是把裸报错抛给用户;一致性上,模型升级、提示词修改、策略调整,应该在一个地方统一管理,而不是散落在各个客户端代码里。

这些需求叠在一起,结论很明确:客户端和模型之间必须有一层专门做"路由和策略"的中间件。它不是一个可有可无的网关,而是文本工作流的控制中枢。理解了这一点,我们再来看 RelayRouter 就顺理成章了。

3. RelayRouter 的本质与定位

3.1 RelayRouter 是什么:一个称职的中间层

先说明一下,RelayRouter 不是某个商业产品的固定名字,我在这里用它指代一类组件:位于调用方与多种模型/服务之间,负责把请求转送到正确模型、并把流式响应送回来的路由转发层。你可以把它理解成 AI 世界的交换机——它不生产内容,但每个请求都要经过它,由它决定往哪条路走。

一个合格的 RelayRouter 至少具备四类能力。第一是模型路由:根据请求的特征(意图、会话历史、业务标签、预算)选择最合适的模型。比如用户问"今天几号",直接走轻量模型就够了,不用动用旗舰。第二是容错与负载均衡:同一个模型配多个实例或密钥时,把请求分摊开;某个模型超时或报错时,自动切到备用模型,用户几乎无感。第三是策略执行:统一的限流、鉴权、内容安全过滤、token 预算控制,这些策略如果散落在业务代码里,早晚要出乱子。第四是可观测性:每个请求从进系统到流式结束,中间调了哪个模型、用了多少 token、每环节耗时多少、失败在哪一步,都要有日志和链路追踪。

3.2 在文本工作流中的精确位置

文本工作流一般可以画成一条链:应用层 → 预处理/上下文装配 → 路由决策 → 模型调用 → 后处理 → 流式返回。RelayRouter 蹲在"路由决策"这个点上,前后各有两个邻居。

它前面的邻居是"上下文装配",负责把历史对话、知识片段、业务参数拼装成 prompt。它后面的邻居是"模型和执行器",也就是各种大模型、向量库、外部 API。RelayRouter 的工作,就是把前面拼好的请求,按规则打到后面合适的执行器上。

这个位置决定了两个特性。第一,RelayRouter 必须对请求内容有感知,但不应深度介入业务逻辑,它只根据路由表做转发和策略控制。第二,RelayRouter 是天然的集成点:模型的限流、密钥轮换、版本切换,都在路由层完成,业务方不需要改代码。比如要把主力模型从 A 换成 B,只需要改路由表,不用推翻业务侧代码。

我用一个比喻来解释:如果把文本工作流比作城市交通系统,上下文装配是规划和导航,模型是目的地,RelayRouter 就是十字路口的信号灯和控制中心。它决定哪个方向的车辆先走、哪条路堵了怎么分流。没有信号灯车也能开,但车一多,必堵。

3.3 和 Agent、API Gateway 的区别

聊路由层时,很多人会把它和另外两个概念搞混:API Gateway 和 Agent。这个问题我被问过太多次,必须捋清楚。

API Gateway 偏重传输层治理:鉴权、限流、路由到不同后端服务、日志。它通常不理解"模型"这个概念,不知道你路由的两个目标里哪个是旗舰、哪个是便宜模型。RelayRouter 在模型语义层工作:它知道模型的上下文窗口、定价、擅长领域、当前健康状态,能做出"这任务该给哪个模型"的判断。Agent 则更高一层:它负责拆解目标、制定计划、循环调用工具和模型,一个 Agent 内部往往会多次访问模型,而这多次访问都可以经过 RelayRouter。反过来,RelayRouter 不决策"要做什么",只决策"这一步调用该去哪"。

它们不是替代关系,而是分层协作。一个小型多 AI 协作系统里,外层是 Agent 编排,内层是 RelayRouter 管模型路由,最底层还有 API Gateway 管传输。我给团队整理过一张分层表:

层级关注点典型问题常见实现
Agent 编排层做什么任务怎么拆、工具怎么选、循环怎么收敛LangChain 类框架、自研编排器
RelayRouter 路由层调用谁这个请求给哪个模型、要不要切换自研路由、LiteLLM 等开源网关
API Gateway 传输层谁在调鉴权、限流、TLS、后端转发Kong、Nginx、云厂商网关

这套分层最大的价值,是让每一层只干自己擅长的活。业务团队不用天天追着模型供应商的版本跑,基建团队也不用被各种业务逻辑拖住。

4. 路由策略与实时性博弈

4.1 四种常见路由策略

路由器的核心是策略,也就是拿什么标准决定请求去哪。我按工程上的常见程度排一下。

一是规则路由:按请求携带的标签、来源、用户等级、业务类型做硬性匹配。比如"日志分析类请求走模型 A,客服闲聊走模型 B"。好处是简单可控、延迟极低;坏处是规则需要人维护,复杂场景覆盖不全。

二是语义路由:把请求文本向量化,和各个模型的"能力画像"做相似度匹配,选最合适的。比如用 embedding 判断这是代码问题、数学题、还是情感对话,再决定用哪个模型。好处是灵活,坏处是要维护向量和模型画像,路由本身也有延迟消耗。

三是成本与延迟感知路由:给每个模型配置打分函数,综合价格、预期延迟、上下文长度、当前负载,实时选出最优解。适合成本敏感的规模化场景,但需要埋点数据支撑打分。

四是故障切换链路:给主模型配一个或多个备用模型,主模型失败或超时后自动降级。这种策略在流式场景下切换成本很高——前面已经吐了 100 个 token,切到新模型从头生成,语义可能完全不同,所以在实时系统里,降级通常在首字迟迟不出时才执行。

我建议团队从规则路由入手,跑通后再引入语义和延迟感知。一上来就玩复杂策略,排查问题时头会很大。

4.2 实时场景下路由的特殊性

路由这件事,做异步任务和做实时对话,复杂度完全不是一个量级。异步场景可以慢慢打分、慢慢选,用户反正看不到过程;实时场景下,路由决策本身就是延迟的一部分,你必须让它在几十毫秒内完成。

由此引出几个工程原则。第一个:路由决策要轻。不要在路由器里做重计算,需要调用外部服务判断意图时,要么异步预判,要么用极快的缓存结果,绝不能让一次路由变成"二次模型调用"。第二个:要为上游不稳定留退路。实时对话里模型可能流式输出到一半断掉,或者响应质量明显异常,路由器在接入时就要做好健康状态监控,比如统计连续错误率、平均首字延迟,达到阈值就动态摘除某个模型。第三个:流式转发要做背压控制。上游速度快、客户端速度慢时,路由器能限速缓冲,而不是无脑往内存里堆;反过来,上游慢、客户端等急了,也要有超时和提示机制。这些不处理好,用户看到的就是"转圈半天没反应"。

4.3 状态感知:实时会话必须做亲和路由

最后聊一个最容易忽略的点:会话亲和性。文本工作流不是一次性的 HTTP 请求,而是一个个有状态的会话。同一个用户的连续对话,必须路由到能访问同一份上下文的执行单元上。否则用户上一句说"我住在杭州",下一句问"那边明天天气怎么样",第二次路由到没有上下文的新会话,模型就会反问"你说哪里"。

但也不建议矫枉过正,认为整个会话必须永远绑定同一个模型。更合理的做法是"上下文亲和 + 模型动态选择":会话 ID 决定用哪份记忆和上下文,请求特征决定这一轮用哪个模型。例如简单问候走轻量模型,同一会话里出现复杂推理请求时,再临时升级到旗舰模型。这样既保住了多轮连贯性,又没浪费成本。

这个需求对路由器的设计要求是:路由时同时携带会话上下文句柄和请求特征,决策必须同时考虑"上下文在哪"和"模型适合谁"。很多自研路由没注意到这点,结果就是多轮对话质量时好时坏。

5. 实操:给文本工作流加一层 RelayRouter

5.1 最小可运行的版本

理论讲再多,不如动手跑一个最小版本。下面这个例子基于 FastAPI + SSE,实现一个极简 RelayRouter:接收聊天请求,从模型注册表里按规则选模型,把流式响应转发给前端。代码删掉了鉴权和持久化,只保留下核心骨架。

先把模型注册表放在 config.json 里:

{ "models": { "fast": { "provider": "openai-compatible", "base_url": "http://localhost:8001/v1", "api_key": "local-key", "context_window": 32768, "cost_per_1k": 0.001 }, "pro": { "provider": "openai-compatible", "base_url": "http://localhost:8002/v1", "api_key": "local-key", "context_window": 128000, "cost_per_1k": 0.03 } } }

然后是路由器主体。核心就一个函数和一个端点:根据请求里的 task_type 选模型,拿到流式响应后原样转发。

import json import httpx from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() with open("config.json") as f: registry = json.load(f)["models"] ROUTES = { "chat": "fast", "reasoning": "pro", "coding": "pro", "summary": "fast", } async def pick_model(task_type: str): # 规则路由 + 故障降级:主选模型不可用就退到另一个 primary = ROUTES.get(task_type, "fast") if registry[primary].get("disabled"): return "pro" if primary == "fast" else "fast" return primary @app.post("/chat") async def chat(req: Request): body = await req.json() history = body.get("history", []) task_type = body.get("task_type", "chat") model_name = await pick_model(task_type) model = registry[model_name] payload = { "model": model_name, "messages": history, "stream": True, } async def relay(): async with httpx.AsyncClient(timeout=30) as client: async with client.stream( "POST", model["base_url"] + "/chat/completions", json=payload, headers={"Authorization": f"Bearer {model['api_key']}"} ) as resp: if resp.status_code != 200: yield f"data: {json.dumps({'error': 'upstream_error'})}\n\n" return async for line in resp.aiter_lines(): if line.startswith("data: "): yield line + "\n\n" return StreamingResponse(relay(), media_type="text/event-stream")

这个版本虽然简陋,但已经把 RelayRouter 的骨架搭出来了:接收请求、读路由表、选模型、转发流式响应。你完全可以在基础上加历史缓存、限流、观测。验证一下,用 curl 调一次:

curl -N -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"task_type": "reasoning", "history": [{"role": "user", "content": "解释一下什么是路由亲和性"}]}'

curl -N 参数很重要,它告诉 curl 不要缓冲,立即输出服务器发来的流式内容。不用 -N 的话,你会傻等所有内容收完才看到输出。这个习惯在测流式接口时特别有用。

5.2 流式转发的两个关键参数

第一个是 chunk 转发逻辑。主流模型流式返回格式是 SSE,每一行 data: 里带着一段增量文本或元数据。RelayRouter 不需要解析内容本身,但要负责把每一行原样或封装后透传给前端。注意不要做"攒满一批再发"的缓冲,要收到一行就发一行,否则首字延迟和吐字节奏全被路由器自己搞坏了。

第二个是超时控制。实时场景下,超时默认值要更激进。我一般会设两个参数:首字超时控制在从请求发出到收到第一行响应的等待时间,实时对话设 5~10 秒;整流超时控制整段流式输出的上限,比如 120 秒。超时后路由器要主动断开连接,并按降级策略尝试切换模型。

我自己踩过的坑是:把超时设成 60 秒的整段超时,结果模型偶发卡住时用户端直接卡死一分钟。后来改成"首字超时 + 流内空闲超时"(比如 30 秒没有新 token 就判死),体验立刻稳了很多。

5.3 验证实时效果:别只看测试通过

代码跑通不代表实时性达标,你还要验证两个维度:数字维度和体验维度。

数字维度上,一口气测这四个指标:首字延迟(TTFT)、平均吐字速度(tokens/s)、流中断率、错误率。我习惯写一个简单压测脚本,往路由器并发打 50 个请求,记录每个请求的首字延迟分布。重点看 p95 和 p99——实时对话看平均值没用,最慢的 5% 用户才是决定口碑的。

体验维度上,找真人做盲测:连续问 5 个有上下文关联的问题,看模型是否每一轮都记得前文;中途打断重说,看响应是否恢复正常;特意问一个超复杂问题,看会不会卡在路由决策上。这些在自动化测试里很难覆盖,但恰恰是用户真实感知"实时感"的来源。

6. 常见问题与排查实录

6.1 我踩过的三个高频坑

第一个坑:路由到错误模型导致输出格式突变。症状是同一个会话里,用户感觉 AI"变笨了"或者"性格变了"。排查后发现是规则路由把某一类请求打到小模型,而小模型不擅长保持格式。解决方法是把"格式要求"作为路由规则的一个因子,一旦业务要求 JSON 输出或特定语气,强制走能力更强的模型。

第二个坑:上游模型流式输出中途重置。症状是流输出到一半突然从头开始,或者直接断流。常见原因是上游服务超时或触发限流。这个问题我在自研路由里修过多次:给每个上游配独立的健康计数,连续 3 次流中断就自动摘除,等 60 秒冷却后再放回来试。

第三个坑:会话上下文重复叠加,prompt 越来越长。每次路由都把完整历史带上,不做截断和压缩,结果到第 20 轮对话时首字延迟翻了一倍。解决方法是加一个历史管理组件:超过窗口上限时,用摘要模型把早期对话压缩成一段摘要,再和近期完整对话拼接。

6.2 排查思路与速查表

文本工作流的排查难点在于跨组件定位。请求要经过应用层、路由器、模型服务、工具调用多个环节,任何一个环节出问题,表现都类似"AI 没反应"或"回答怪"。我的建议是,路由器层必须打三种日志:请求进入日志、路由决策日志(选了什么模型、为什么)、流式结束日志(成功/中断/耗时)。有这三类日志,大部分问题半小时内能定位。

下面是我整理后常贴在团队 wiki 里的速查表:

现象常见原因排查链路解法
首字延迟飙升上下文过长 / 模型负载高 / 路由决策太重看 TTFT 分布、路由决策耗时、上游监控压缩上下文、做路由缓存、配备用模型
流输出到一半断掉上游超时 / 触发限流 / 客户端断连看流式结束状态码和原因码增加重试、健康摘除、前端自动续接
多轮对话答非所问会话上下文没被正确路由到检查会话 ID 是否贯穿所有请求会话亲和路由 + 上下文句柄
回答内容格式混乱路由到不擅长该任务的模型检查路由决策日志规则增加任务类型/格式因子
成本突增大量请求走了旗舰模型看按模型分组的 token 用量报表收紧路由规则、设置预算告警

排查时还有一个很土但很有效的技巧:在路由器里加一个 debug 开关,请求带上?debug=1,响应里附带这次调用的模型名、耗时和 token 数。生产环境关掉,测试环境必开。这能把黑盒变成白盒,比什么都好用。

6.3 关于"中间层是否值得做"的一点经验

最后替纠结的团队说句实话:路由层该不该自研,取决于系统复杂度。如果只有一个大模型、日请求几百次、不需要切换和降级,那直接用官方 SDK 直连,别自找麻烦。但一旦开始接第二个模型、开始处理多 Agent 协作、开始被老板问"哪些请求可以用便宜模型",路由层就不是选择而是必须了。它能帮你把"换模型"从一次伤筋动骨的改造,变成改一行配置的事。

我自己经历的项目里,最典型的一个转变发生在接第三个模型时。当时团队花了大量精力在改业务代码去适配不同模型的响应格式,改完又出各种兼容问题。后来把路由层抽出来,统一做格式转换、错误处理和模型选择,业务代码从此再没因为换模型动过。所以我最后想说的是:实时 AI 产品想跑得好,别老盯着视频界面去调光线、调口型,把精力放在那条文本管道上。Avatar 是面子,工作流是里子。里子不够好,面子再光鲜,用户聊两轮就跑了。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询