给inbound销售智能体做90%自动化，TaoToken 的 Fallback

1. 从 429 和 504 开始排障inbound 销售智能体为什么必须做失败降级给 inbound 销售智能体做 90% 自动化时TaoToken 的 Fallback 链要先解决429、503、504这类失败降级。在配置任何重试逻辑前先到 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_intro 拿 Key并把 Base URL 设为https://taotoken.net/api。Vercel 把 inbound 销售开发做成高度自动化的公开案例真正可复用的不是团队编制变化而是把模型调用当成有失败率的依赖一次限流不能让线索丢在队列里。下面不讨论销售团队规模只做可跟做的接入与排障错误码如何归一化主备模型如何切换规则兜底和人工队列如何接上以及怎样用测试用例证明降级后线索不丢。一个典型的 inbound 处理流程通常是收到表单、邮件、聊天窗口或落地页事件清洗并抽取字段邮箱、公司、职位、需求、预算、时间、来源渠道意图分类预约演示、采购咨询、技术支持、合作、招聘、垃圾线索线索打分高意向、中意向、低意向、需人工判断路由直接进销售日历、进入 SDR 队列、自动回复、转人工写回 CRM 或表格并给线索发确认回执。其中第 2 到第 5 步都可能调用模型。如果没有 fallback第 2 步超时会导致第 6 步永远不发回执第 3 步限流会让线索进入死信第 4 步503会让高价值线索被当成普通线索处理。失败降级的目标不是“永不失败”而是定义清楚哪些错误重试、哪些错误换模型、哪些错误走规则、哪些错误直接转人工并且每一种都有日志和测试。这里最容易犯的错误是把 fallback 写成try/except后无限重试。销售线索有实时性用户提交表单后 5 分钟没收到回复意向就会下降。所以更合理的策略是“短预算重试 快速切换 兜底回执”。主模型最多尝试 2 次备用模型最多尝试 1 次整个请求预算控制在 20 秒以内。超过预算就进入规则模板和队列不再阻塞用户线程。2. 先拿 Key 再配链TaoToken 官网、Base URL 与密钥隔离在写任何 fallback 配置前先把凭证和入口统一。去 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_key_setup 完成账号与 Key 管理在控制台的 API Keys 页面创建 Key后面文末 CTA 也给出了直达入口。所有代码、CLI、CC Switch 统一使用 Base URLhttps://taotoken.net/api。注意Base URL 不要带 UTMUTM 只用于官网和 deep link 的跳转追踪。Key 用占位符YOUR_API_KEY真实 Key 放环境变量或密钥管理不进 Git。建议先建立本地环境变量export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEYYOUR_API_KEY export PRIMARY_MODELyour-primary-model-id export SECONDARY_MODELyour-secondary-model-id export SMALL_FAST_MODELyour-small-fast-model-id如果工具支持读取.env可以把这些变量放到项目根目录.env但.env必须加入.gitignore。生产环境建议使用部署平台的 Secret、KMS 或密钥管理服务。不要在代码里写死 Key也不要把 Key 写进前端表单、日志、异常堆栈、浏览器本地存储。可以用一条本地命令检查 Base URL 和 Key 是否可用curl -sS https://taotoken.net/api/models \ -H Authorization: Bearer ${TAOTOKEN_API_KEY} \ -H Content-Type: application/json | head -c 800如果模型列表接口路径与预期不同以 TaoToken 模型对话页和 Claude Code 文档为准。但对于 inbound 智能体调用层只需要先固定两件事Base URLhttps://taotoken.net/api鉴权Authorization: Bearer YOUR_API_KEY为什么失败降级要先统一 Base URL因为 fallback 链里主模型、备用模型、小模型、分类模型最好走同一个入口错误码语义才一致。否则不同上游有的返回429有的返回503有的用业务码包一层映射表会被拆成多套。统一到 TaoToken 的 Base URL 后调用层可以先做一层错误归一化再决定重试还是降级。统一 Key 之后还要统一日志字段。建议每条线索至少记录trace_id一次 inbound 事件的全局追踪 IDlead_id线索唯一 IDprovider当前调用的供应商标识例如taotokenmodel当前模型status_code归一化后的 HTTP 状态码或异常类型latency_ms耗时fallback_level当前降级层级0 表示主模型1 表示备用模型2 表示规则模板3 表示队列4 表示人工attempt当前模型第几次尝试。没有这些字段fallback 上线后你无法回答“到底有多少线索走了降级”“备用模型是否稳定”“规则模板命中率多少”。销售自动化不是只让模型说话而是让失败路径可观测。3. 错误码映射表哪些错误该重试哪些该换模型哪些必须转人工错误码映射是 fallback 的核心。不要把所有异常都 catch 后无脑重试。下面这张表可以直接作为第一版实现依据后续根据 TaoToken 实际返回和业务日志调整。HTTP 状态 / 异常含义是否重试Fallback 动作备注400请求格式错误否修复 prompt 或 payload重试会重复失败401鉴权失败否快速失败并告警检查YOUR_API_KEY403权限不足否快速失败并告警检查 Key 权限404模型不存在否切换备用模型模型名配置错误408请求超时是同模型重试 1 次再切备用短退避409冲突是指数退避后重试可能并发写冲突413载荷过大否截断上下文后重试邮件线程可能过长422语义/参数错误否规则模板兜底不应反复调用模型429限流是退避后重试再切备用模型高峰最常见500服务端错误是切备用模型不建议长时间重试主模型502网关错误是切备用模型上游抖动503服务不可用是切备用模型或入队立即降级504网关超时是切备用模型控制总超时timeout客户端超时是切备用模型主 12s备 8scontent_filter内容安全拒绝否转人工审核不要自动重写绕过unknown未知异常是尝试备用再入队记录原始异常类型这张表的关键原则401、403、400这类配置错误不要降级到备用模型因为它们不是供应商抖动而是你的请求或 Key 有问题。备用模型只会掩盖问题。404是模型配置错误应该立即切备用模型同时告警人工修配置。429、500、502、503、504、timeout是典型的可降级错误。先短退避再换模型。422、content_filter不要反复重试。前者说明模型无法按预期解析后者说明内容需要人工确认。两者都适合进入规则模板或人工队列。所有降级都要有最终兜底给用户回执把线索写入队列而不是静默失败。如果你只记一句话鉴权错误快速失败限流超时立即降级语义安全错误转人工最终一定要入队并回执。4. 五段式 Fallback 配置主模型、备用模型、规则模板、队列、人工为了让 inbound 销售智能体在失败时仍然完成“抽取、打分、路由”的关键动作可以把 fallback 链设计成五段L0TaoToken 主模型负责完整抽取、意图分类和打分L1TaoToken 备用模型负责在主模型限流/超时时快速接管L2规则模板负责用正则和关键词做最小可用抽取L3异步队列负责先回执、后重试避免用户等待L4人工审核只处理高价值、高风险或规则无法判断的线索。下面是一份可复制的 YAML 配置示例放在你的 inbound agent 项目里即可作为配置骨架version: 1 inbound_agent: base_url: https://taotoken.net/api api_key_env: TAOTOKEN_API_KEY timeout_ms: 12000 total_budget_ms: 20000 fallback_chain: - name: taotoken-primary model_env: PRIMARY_MODEL max_attempts: 2 backoff_ms: [500, 1500] retry_on: [408, 409, 429, 500, 502, 503, 504, timeout] - name: taotoken-secondary model_env: SECONDARY_MODEL max_attempts: 1 backoff_ms: [800] retry_on: [429, 500, 502, 503, 504, timeout] degrade: on_401: fail_fast on_403: fail_fast on_404: next_model on_422: rule_template on_content_filter: human_review final: enqueue_and_ack observability: log_fields: - trace_id - lead_id - provider - model - status_code - latency_ms - fallback_level - attempt这里有几个细节值得注意。第一base_url固定为https://taotoken.net/api不要在各个模型配置里重复写不同入口。主模型和备用模型都走同一个 Base URL只通过模型名区分。第二timeout_ms和total_budget_ms分开。单个请求超时 12 秒不代表整个 inbound 处理可以等 12 秒又 12 秒。总预算 20 秒超过就进入 L2/L3。第三retry_on只放可恢复错误。不要把400、401、403放进去。否则销售线索处理会被配置错误拖死。第四final必须是enqueue_and_ack。也就是最终降级动作是“入队并回执”不是“抛异常”。对 inbound 销售来说用户提交表单后没有回执比模型答得不完美更严重。第五规则模板不是摆设。它至少应该能识别邮箱正则提取电话允许带国家码和分隔符公司名从邮件签名、表单字段或域名推断需求关键词demo、报价、采购、集成、API、合作紧急程度ASAP、本周、本月、下季度路由规则包含 demo 且带企业邮箱进销售队列只有技术支持关键词进支持队列无法识别进人工。规则模板结果可以不完全准确但它能让线索不丢并给人工一个可修改的草稿。5. 可复制的 Python 调用层统一 Base URL 与模型链下面给出一个最小可用的 Python 调用层。它使用 OpenAI 兼容 SDK 风格Base URL 指向https://taotoken.net/apiKey 从环境变量读取。你可以把模型名替换成在 TaoToken 模型对话页看到的实际模型 ID。import os import time import random import logging from typing import Any from openai import OpenAI from openai import APITimeoutError, APIConnectionError, APIStatusError logging.basicConfig(levellogging.INFO) logger logging.getLogger(inbound-fallback) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) API_KEY os.getenv(TAOTOKEN_API_KEY, YOUR_API_KEY) PRIMARY_MODEL os.getenv(PRIMARY_MODEL, your-primary-model-id) SECONDARY_MODEL os.getenv(SECONDARY_MODEL, your-secondary-model-id) SMALL_FAST_MODEL os.getenv(SMALL_FAST_MODEL, your-small-fast-model-id) client OpenAI( base_urlBASE_URL, api_keyAPI_KEY, timeout12.0, max_retries0, ) RETRY_CODES {408, 409, 429, 500, 502, 503, 504} FAIL_FAST_CODES {400, 401, 403} def classify_error(exc: Exception) - str: if isinstance(exc, APITimeoutError): return timeout if isinstance(exc, APIConnectionError): return connection if isinstance(exc, APIStatusError): code exc.status_code if code in FAIL_FAST_CODES: return fail_fast if code in RETRY_CODES: return retry_or_fallback if code 404: return next_model if code in (413, 422): return rule_template return unknown return unknown def rule_template_extract(text: str) - dict[str, Any]: import re email_match re.search(r[\w.\-][\w.\-]\.\w, text or ) phone_match re.search(r(\?\d[\d\s\-]{7,}\d), text or ) lowered (text or ).lower() intent unknown if any(word in lowered for word in [demo, 演示, 试用]): intent demo elif any(word in lowered for word in [报价, 价格, 采购]): intent pricing elif any(word in lowered for word in [api, 集成, integration]): intent integration elif any(word in lowered for word in [支持, support, 故障]): intent support score 0 if email_match: score 20 if phone_match: score 10 if intent in {demo, pricing, integration}: score 40 if any(word in lowered for word in [企业, 公司, 团队, 预算]): score 10 return { email: email_match.group(0) if email_match else None, phone: phone_match.group(0) if phone_match else None, intent: intent, score: min(score, 100), route: sales_queue if score 50 else human_queue, source: rule_template, } def enqueue_and_ack(lead: dict[str, Any], reason: str) - dict[str, Any]: logger.warning(enqueue_and_ack lead_id%s reason%s, lead.get(lead_id), reason) # 这里替换为你的队列写入逻辑例如数据库、Redis、SQS、RabbitMQ。 # 注意不要在本示例中直连生产库队列写入应由你的应用服务完成。 return { ok: True, fallback_level: 3, action: enqueue_and_ack, reason: reason, lead: lead, } def call_with_fallback(messages: list[dict[str, str]], lead_id: str) - dict[str, Any]: trace_id ftrace-{lead_id}-{int(time.time() * 1000)} models [ (taotoken-primary, PRIMARY_MODEL, 2), (taotoken-secondary, SECONDARY_MODEL, 1), ] last_error None for level, (provider_name, model, max_attempts) in enumerate(models): for attempt in range(1, max_attempts 1): start time.time() try: resp client.chat.completions.create( modelmodel, messagesmessages, temperature0.2, timeout12.0 if level 0 else 8.0, ) latency_ms int((time.time() - start) * 1000) logger.info( success trace_id%s lead_id%s provider%s model%s attempt%s fallback_level%s latency_ms%s, trace_id, lead_id, provider_name, model, attempt, level, latency_ms, ) content resp.choices[0].message.content return { ok: True, fallback_level: level, provider: provider_name, model: model, content: content, latency_ms: latency_ms, } except Exception as exc: last_error exc action classify_error(exc) latency_ms int((time.time() - start) * 1000) logger.warning( failure trace_id%s lead_id%s provider%s model%s attempt%s action%s latency_ms%s error%s, trace_id, lead_id, provider_name, model, attempt, action, latency_ms, type(exc).__name__, ) if action fail_fast: raise if action rule_template: fallback rule_template_extract(str(messages)) fallback[lead_id] lead_id return { ok: True, fallback_level: 2, action: rule_template, lead: fallback, } if action in {retry_or_fallback, timeout, connection}: if attempt max_attempts: backoff random.uniform(0.3, 1.2) * attempt time.sleep(backoff) continue break if action next_model: break break text str(messages) fallback rule_template_extract(text) fallback[lead_id] lead_id fallback[last_error] type(last_error).__name__ if last_error else unknown return enqueue_and_ack(fallback, reasonall_models_failed)这段代码的重点不是语法本身而是行为约定主模型最多 2 次备用模型最多 1 次401/403/400直接抛出避免掩盖配置问题422直接走规则模板429/500/502/503/504/timeout先重试再切换最终失败进入enqueue_and_ack线索不会丢。上线前把PRIMARY_MODEL、SECONDARY_MODEL、SMALL_FAST_MODEL换成实际模型 ID。建议主模型选质量更高的备用模型选低延迟、高吞吐的。不同模型返回结构可能不同所以不要在主流程里强依赖某一种输出格式。最好让模型输出 JSON并在调用层做 schema 校验校验失败也进入规则模板。6. Claude Code、Codex、CC Switch 三件套配置如果你用 Claude Code 做本地排障和配置生成需要把 Claude Code 的入口指向 TaoToken。Claude Code 使用settings.json和ANTHROPIC_*环境变量不要把这套变量套到 Codex 上。示例{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: your-claude-model-id, ANTHROPIC_SMALL_FAST_MODEL: your-fast-model-id } }放置位置通常为用户目录下的 Claude Code 配置。实际路径和字段以 TaoToken 的 Claude Code 文档为准。修改后重启 Claude Code并用一个最小请求验证。不要把真实 Key 提交到 Git如果必须写进本地配置也只在本地环境使用YOUR_API_KEY替换。Codex 使用config.toml不要写ANTHROPIC_*。示例model your-codex-model-id model_provider taotoken model_reasoning_effort medium [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat然后在 shell 中设置export TAOTOKEN_API_KEYYOUR_API_KEY如果你的 Codex 版本使用responses协议把wire_api改为对应值但base_url仍是https://taotoken.net/api。模型名以实际可用模型为准。CC Switch 三件套可以理解为三个配置入口项目根目录.env放TAOTOKEN_API_KEY、TAOTOKEN_BASE_URL、PRIMARY_MODEL、SECONDARY_MODELClaude Code 的settings.json放ANTHROPIC_BASE_URL、ANTHROPIC_AUTH_TOKEN、ANTHROPIC_MODELCodex 的config.toml放model_provider、base_url、env_key。在 CC Switch 中新增 provider 时可以按下面信息填写名称TaoToken Base URLhttps://taotoken.net/api API KeyYOUR_API_KEY Key 环境变量TAOTOKEN_API_KEY 用途inbound fallback 主/备模型、Claude Code、Codex这样切换工具时不会出现 Claude Code 能通、Codex 不通或者 Codex 通、项目脚本不通的情况。统一 Base URL 和 Key 环境变量是把失败降级做稳定的前提。7. 测试用例用 429/503/超时验证券商降级不丢线索Fallback 配置写完后必须有测试用例。不要等生产出现429才验证。下面用 pytest 演示如何验证错误码映射和模型链切换。你可以把call_with_fallback替换成上一节的实现。import pytest from unittest.mock import MagicMock from openai import APIStatusError def make_status_error(status_code: int): response MagicMock() response.status_code status_code return APIStatusError( messagefmock status {status_code}, responseresponse, body{error: {code: status_code}}, ) pytest.mark.parametrize( status_code,expected, [ (400, fail_fast), (401, fail_fast), (403, fail_fast), (404, next_model), (408, retry_or_fallback), (413, rule_template), (422, rule_template), (429, retry_or_fallback), (500, retry_or_fallback), (502, retry_or_fallback), (503, retry_or_fallback), (504, retry_or_fallback), ], ) def test_error_mapping(status_code, expected): from your_module import classify_error assert classify_error(make_status_error(status_code)) expected def test_primary_429_then_secondary_success(monkeypatch): from your_module import call_with_fallback calls [] class FakeMessage: content {intent: demo, score: 80} class FakeChoice: message FakeMessage() class FakeCompletion: choices [FakeChoice()] def fake_create(model, messages, temperature, timeout): calls.append(model) if len(calls) 1: raise make_status_error(429) return FakeCompletion() monkeypatch.setattr(your_module.client.chat.completions.create, fake_create) result call_with_fallback( messages[{role: user, content: 我想预约 demo}], lead_idlead-001, ) assert result[ok] is True assert result[fallback_level] 1 assert len(calls) 2 def test_all_failed_enqueues_and_acks(monkeypatch): from your_module import call_with_fallback def fake_create(model, messages, temperature, timeout): raise make_status_error(503) monkeypatch.setattr(your_module.client.chat.completions.create, fake_create) result call_with_fallback( messages[{role: user, content: 需要采购 API 集成}], lead_idlead-002, ) assert result[ok] is True assert result[action] enqueue_and_ack assert result[lead][lead_id] lead-002这些测试覆盖三类关键场景鉴权错误快速失败不进入备用模型主模型429后备用模型成功fallback_level变成 1所有模型失败后进入队列并回执action为enqueue_and_ack。建议再加入超时测试。可以用 monkeypatch 让第一次调用抛出APITimeoutError第二次返回正常结果。也可以加入规则模板测试传入一段包含邮箱和“demo”的文本断言intentdemo、score50、routesales_queue。测试不需要真实 Key也不应该打真实生产库。所有 SQL、队列写入命令都在本地测试环境执行生产数据由应用服务按权限处理。测试通过后再做一个本地演练把PRIMARY_MODEL设置为一个不存在的模型观察是否触发404 - next_model把SECONDARY_MODEL也设置为不存在观察是否进入rule_template和enqueue_and_ack。这样你就能确认 fallback 链不是纸面配置。8. 上线检查清单与 CTA从模型对话到 Claude Code 文档上线前按下面清单逐项确认Base URL 是否统一为https://taotoken.net/apiKey 是否使用YOUR_API_KEY占位并放入环境变量或密钥管理401/403/400是否快速失败不掩盖配置问题404是否切换备用模型429/500/502/503/504/timeout是否短退避后切换422/content_filter是否进入规则模板或人工队列总预算是否控制在 20 秒以内最终失败是否enqueue_and_ack日志是否包含trace_id、lead_id、model、status_code、fallback_level是否有 429、503、超时、全失败的测试用例Claude Code 的settings.json和 Codex 的config.toml是否分开配置CC Switch 三件套是否都指向同一个 Base URL 和同一个 Key 环境变量。如果这些项都满足你的 inbound 销售智能体就有了一条可用的失败降级链主模型正常时追求质量限流时切备用语义异常时走规则全部失败时入队并回执。它不会让一次429变成一条丢失的销售线索。下一步可以按这个顺序验证先打开模型对话验证模型返回和错误码行为https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_chat如果要把 inbound 智能体长期跑在开发环境查看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_coding_plan创建自己的 API Key把YOUR_API_KEY替换为真实密钥https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_api_keys需要配置 Claude Code 时对照 Claude Code 文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentfallback_claude_code先把 Base URL 固定为https://taotoken.net/apiKey 用YOUR_API_KEY占位再按错误码映射跑一遍测试你的 inbound fallback 链就有了可上线的最小闭环。