| 问题 | 改中转源码 / 放弃接入 | 代码渠道 |
| 上游接口不标准 | 等平台发版支持 | 贴一个 spec 类当场接入,保存即热更新 |
| 登录方式特殊 | 自己维护一套登录脚本 | 声明字段 + 授权钩子,轮询/刷新由框架调度 |
| SSE 格式非标 | 整段重写聊天逻辑 | 只写 parse_chunk 一帧解析 |
| 只想加一个头/一个字段 | 大动干戈 | headers / payload 单钩子搞定 |
| usage 统计缺失 | 自己算 token | 框架按实际内容自动估算,不用管 |
先记住 7 条规则
- 写
class MyChannel:,不要继承 BaseProvider / CustomProvider。
- 方法推荐写
@staticmethod,第一个参数 p 是渠道实例。
- 渠道地址在管理端「渠道配置」里填(必填);spec 用
p.base_url 读,不要写死域名。不打外网的 spec 写 REQUIRES_BASE_URL = False 可豁免。
- usage 不用你算:上游没回 token 统计时框架自动估算;只有拿到真实 usage 才需要覆盖。
- 不要自己
import aiohttp:用 p.send_sse_request(...) / p._make_session(),出站代理、连接复用、请求留痕才生效。
- 不要写
PROVIDER_NAME,加载器会用渠道 id 覆盖。贴入的代码以服务进程权限执行,等同部署一个 Python 文件,只允许可信管理员编辑。
- OAuth / 多字段账号在
account_schema()["fields"] 声明表单字段;运行时字段写 ACCOUNT_FIELDS。两者自动挂成 p.access_token 这类实例属性。
场景索引:你属于哪种情况?
| 你的情况 | 你写什么 |
| 上游是 OpenAI / Anthropic / Responses / Gemini 兼容接口 | 零钩子(渠道配置即可);源码至少留一个最小 init_auth |
| 只多一个签名头 / 租户头 | headers |
| 请求体只多一个字段 | payload |
| 聊天 URL 非标准路径 | build_url |
| 接入地址登录后动态下发 | base_url(覆盖全部出站路径) |
SSE 字段不是 choices[].delta | parse_chunk |
| OAuth 设备码登录 | account_schema + begin_device_flow + poll_device_flow |
| OAuth 回调登录 | account_schema + build_auth_start + handle_auth_callback |
| token 需要定时刷新 | refresh_auth + SCHEDULED_REFRESH = True |
| 运行时刷新 token 要落库 | await p.persist_account_fields({...}) |
| 账密登录换 ticket / Cookie | init_auth + 需要的聊天/模型钩子 |
| 只想拒绝某些模型或消息 | check_message |
| 上游不认 function calling | p.build_tools_prompt + p.parse_tool_calls(解析框架已内置) |
钩子速查
认证、账号和授权
| 钩子 | 签名 | 用途 |
init_auth | (p, is_check=False) -> bool | 首次使用、失效自愈、体检 |
check_auth / is_init / health_check | (p) | 检查登录态 / 凭据就绪 / 健康检查 |
account_schema | () -> dict | 声明前端账号字段和授权入口 |
begin_device_flow / poll_device_flow | (p[, poll_params]) -> dict | 设备码授权启动与单步轮询 |
build_auth_start / handle_auth_callback | (name, …, cfg) | OAuth 跳转与回调 |
refresh_auth | (p, account, cfg) -> dict | 刷新 token,返回窄写字段 |
聊天、模型和请求
| 钩子 | 签名 | 用途 |
fetch_models | (p) -> list[dict] | 自定义模型列表 |
stream_chat | (p, model_id, messages, **kwargs) | 自定义流式聊天,yield 统一帧 |
non_stream_chat | (…) -> dict | 自定义非流式响应 |
headers / payload | (p, …) -> dict | 请求头 / 请求体后处理 |
build_url / base_url | (p[, kwargs]) -> str | 覆盖聊天 URL / 全部出站地址 |
parse_chunk | (p, event_str) -> dict | 覆盖一帧 SSE 解析 |
回调、配额和生命周期
| 钩子 | 用途 |
on_response_headers / update_quota | 响应头回调 / 从响应头更新配额 |
check_message / record_message | 请求前放行判定 / 自定义记账 |
on_channel_attached / close | 配置注入后初始化 / 释放资源 |
clear_conversations | 清理上游会话 |
类级开关
| 开关 | 默认 | 作用 |
SUPPORTS_TOKEN_AUTO_REFRESH | False | 凭据过期后能否自愈;True 时前端标「已过期(等待自动刷新)」 |
SCHEDULED_REFRESH | False | 纳入每日 10:00 主动刷新(需同时写 refresh_auth) |
SUPPORTS_MULTI_MESSAGES | False | 是否把多轮 messages 整批发给上游 |
TOOLS_AS_PROMPT / TOOLS_PROMPT_FORMAT | False / "xml" | 上游不认 function calling 时工具说明进 system;形态 xml/json/hermes |
REQUIRES_BASE_URL | True | 渠道地址是否必填;不打外网的 spec 写 False 豁免 |
ACCOUNT_FIELDS | () | 账号字段名清单,挂实例属性 |
APPLY_CLIENT_PRESET | True | 是否套客户端伪装头;上游校验自家 CLI 头时写 False 硬关 |
p 上最常用的东西
| 属性 / 方法 | 含义 |
p.base_url | 管理端「渠道配置」里填写的渠道地址;不要在 spec 里写死 |
p.api_key / p.username / p.password | 渠道 / 账号配置派生的凭据 |
p.proxy / p.timeout_seconds | 统一出口代理配置 / 请求超时 |
p._channel | 渠道领域对象:base_url、protocol、chat_protocols、billing_mode |
p._make_session() / p.send_sse_request(...) | 遵循渠道代理的 HTTP session / SSE 请求(带请求留痕) |
p.build_openai_response(id, model, content, …) | 构造统一 OpenAI 响应;id 传 generate_completion_id()(已注入) |
p._raise_upstream_error(error) | 复用框架错误归一,不要自己复制一份 |
p.build_tools_prompt / p.prepare_provider_messages / p.parse_tool_calls | 文本工具三件套:说明 prompt / 拼 system / 文本→工具调用解析 |
await p.persist_account_fields({...}) | 运行时刷新出的字段窄写回本账号并同步池内实例 |
流式统一帧
yield {} # 可选但推荐:先通知主链路上游已接通
yield {"content": "正文增量", "thinking": "", "tool_calls": []}
yield {"content": "", "thinking": "思考增量", "tool_calls": []}
yield {"content": "", "thinking": "", "tool_calls": [{
"index": 0, "id": "call_1", "type": "function",
"function": {"name": "search", "arguments": {"q": "天气"}}}]}
# 也可带 usage / finish_reason / done;上游返回真实 usage 时才需要覆盖
样例与演示片段
| 样例 | 演示内容 | 用到的钩子 |
| 最小可跑:EchoChannel | 原样回吐用户最后一条消息,验证「贴代码→出渠道」链路 | init_auth / fetch_models / stream_chat |
| OAuth 设备码授权渠道 | 声明前端字段 + 授权按钮 + 轮询 + 定时刷新令牌;聊天走渠道配置,只用 headers 塞 Authorization | account_schema / begin_device_flow / poll_device_flow / refresh_auth / init_auth / headers |
| OpenAI 兼容反代 | 上游已是标准 /v1/chat/completions;聊天一行不写,只用 init_auth + headers 塞访问密码 | init_auth / headers / payload |
| 账密登录抓取型 | 账号密码换 ticket / Cookie、非标 SSE、登录态维护和失效自愈等组合场景 | init_auth / check_auth / fetch_models / stream_chat / non_stream_chat / clear_conversations |
下面是理解钩子关系的演示片段,不是完整实现。它们故意省略真实认证、签名、持久化、错误处理和协议细节;请按自己的上游改写后再放入「源码」Tab,不要直接用于生产。
兼容接口:只加一个请求头
class TenantChannel:
@staticmethod
async def init_auth(p, is_check=False):
# 标准聊天流程继续由配置驱动实现
return bool(p.api_key and p.base_url)
@staticmethod
def headers(p, base_headers, kwargs):
# 只在框架生成的请求头上补充自己的字段
base_headers["X-Tenant-Id"] = p.username
return base_headers
设备码授权:三个钩子的配合
class DeviceChannel:
@staticmethod
def account_schema():
# 声明字段与授权入口;真实字段按上游协议补充
return {
"fields": [{"key": "access_token", "type": "password", "secret": True}], # 其他字段省略
"auth_start": {"enabled": True, "mode": "device_code"},
}
@staticmethod
async def begin_device_flow(p):
# 请求细节省略,只返回前端展示信息和轮询参数
return {"auth_url": "...", "user_code": "...", "interval": 5,
"poll_params": {"device_code": "..."}}
@staticmethod
async def poll_device_flow(p, poll_params):
# pending 继续轮询;成功字段放进 account_data
return {"status": "authorized", "account_data": {"access_token": "..."}}
非标准 SSE:只覆盖一帧解析
class SseChannel:
@staticmethod
def parse_chunk(p, event_str):
# 上游自定义事件 → 平台统一帧;解析不了可返回 None
data = json.loads(event_str.removeprefix("data:").strip())
return {
"content": data.get("answer", ""),
"thinking": data.get("analysis", ""),
"tool_calls": [],
}
文档页只提供上述说明性片段;完整的认证和协议实现应由你根据上游接口自行设计,并通过框架提供的 p._make_session()、p.send_sse_request() 等能力接入。
让 AI 帮你生成
平台提供一份自包含的需求说明(Markdown):里面是情况判定表、脚本元素到 spec 的映射表、全部钩子的签名与返回契约、注入命名空间、自查清单和最小完整样例。把下面这句话连同你的脚本 / 仓库地址发给任何 AI 编码助手,它读完该文档即可产出符合本页规范的 spec 类:
请根据 https://ai-lubricant.vip100.de5.net/code-channel-ai.md 的要求,写一份自定义代码渠道(spec 类)。
对应的脚本 / 参考代码在:<仓库地址,或直接把代码粘贴在后面>
用法要点:
- 仓库地址可以是 AI 能访问的任何 URL(GitHub 仓库、单文件脚本、API 文档页都可以),也可以直接把代码粘贴在句子后面。
- AI 交付的是一段 Python 源码 + 渠道配置建议(协议、路径、账号字段)。先人工审阅,再贴入「源码」Tab——贴入的代码以服务进程权限执行。
- 同一个句子人类也能用:拿着文档第 10、11 节的交付契约与自查清单,手动核对 AI 的产出即可。
- 直接查看:code-channel-ai.md(同域原文,AI 可直接抓取)。
常见错误
| 错误 | 处理 |
| 渠道地址留空就保存 | 保存被拒绝;填地址,或写 REQUIRES_BASE_URL = False 声明不打外网 |
| 继承 BaseProvider 报错 | 删掉继承,写普通类;框架能力通过 p.xxx 使用 |
| 写了 stream_chat 没有 yield | 流式无内容;至少 yield 统一帧 |
| account_schema 字段名和 account_data 对不上 | 两边 key 必须一致,否则授权后不回填 |
| base_url 写在代码里 | 管理端填的地址不生效;用 p.base_url |
| 自己 import aiohttp 发请求 | 绕开出站代理、连接复用和请求留痕;用 p._make_session() |
| 每个钩子都重写 | 越容易漏协议、工具调用、日志和重试;优先只写一个最小钩子 |
保存即热更新保存 code 后,后端校验语法、扫描唯一 spec 类并热更新适配器,进程内立即生效无需重启。语法错、无 spec、钩子拼错、误继承基类、渠道地址缺失都会返回 HTTP 400,不落库。代码在服务进程权限下执行,只应由可信管理员编辑;它不是沙箱。
← 大模型网关能力详解 · 渠道配置操作 →