很多做 AI 应用的团队都会遇到同一个组合需求让用户“带上自己的 API Key”来调用大模型搜索能力同时平台方又必须把每一次调用的账记清楚——谁调的、用的哪个模型、花了多少 token、响应延迟多高、有没有报错。前者是 BYOKBring Your Own Key自带密钥模式后者是 AI 搜索追踪。这两件事看上去不相关落地时却必须放在一起设计。本文将围绕一个可运行的最小实现完整拆解“BYOK AI 搜索追踪”库的设计与编码过程内容包括密钥管理器、SQLite 存储层、OpenAI 兼容的搜索客户端、装饰器追踪工具以及 MIT License 相关的合规注意点。无论你是想给现有项目接入 AI 搜索功能还是想把 LLM 调用做成可审计、可计费的后端能力这篇文章都能提供一套可以直接复用的思路。1. 背景与核心概念1.1 什么是 BYOKBYOK 全称是 Bring Your Own Key直译就是“带上你自己的密钥”。这个概念最早在云服务和 SaaS 领域比较常见比如企业把自有的加密密钥托管到云厂商的 KMSKey Management Service里密钥的生成、轮换和管理权仍然掌握在自己手中。到了大模型和 AI 搜索场景BYOK 的含义就变得更贴近日常开发库或平台不内置任何供应商的 API Key由调用方在运行时把 Key 注入进来。你可以用自己的 OpenAI Key、DeepSeek Key、通义千问 Key也可以是公司内部统一申请的企业 Key。BYOK 模式之所以受欢迎核心原因有几个成本归属清晰谁调用谁付费。用平台统一转发的 Key费用全部记在平台头上月底对账容易扯皮。数据边界可控Key 不经过第三方中转服务请求直接从你的服务发往模型供应商敏感字段的暴露面更小。切换灵活今天用 A 厂商的模型明天换 B 厂商只需要换 Key 和接口地址不需要改业务代码。合规友好很多企业客户不允许自己公司的密钥出现在另一个服务商的数据库里BYOK 天然规避了这个问题。不过 BYOK 也带来一个副作用调用方变成“多租户”的每个租户消耗的 token 不同、报错不同、用量不同。如果不去追踪出问题的时候连“这波开销是哪个客户产生的”都查不到。所以 BYOK 与追踪通常是一起出现的。1.2 为什么要做 AI 搜索追踪AI 搜索不是简单的关键词匹配它背后往往是一次完整的 LLM 调用模型接收用户的搜索意图结合上下文生成回答和引用。这类调用的成本、延迟和成功率高度依赖模型、提示词长度和供应商状态。追踪层需要记录的关键维度至少包括维度说明request_id每次调用的唯一标识用于问题定位和对账provider使用的是哪家供应商 / 哪个租户的 Keymodel实际使用的模型名例如 gpt-4o-miniquery用户搜索的原始输入tokensprompt_tokens、completion_tokens、total_tokenslatency_ms从发起到拿到响应的总耗时statussuccess / error以及错误码created_at调用发生时间有了这些数据你能做三件事成本归因按 provider 汇总 token 消耗精确算出每个业务方、每个客户的使用量。质量排查用户投诉“搜索变慢了”先看延迟曲线和错误率而不是盲目调参数。限流与预算预警当某个 Key 的日消耗接近阈值时自动熔断或告警。需要强调的是追踪不等于把用户输入的明文原样存起来。查询内容可能包含业务敏感信息设计存储时要考虑脱敏、自动清理和访问权限。1.3 MIT License 能做什么、不能做什么项目采用 MIT License意味着源码允许任何人免费使用、修改、分发甚至闭源商用。MIT 是目前最宽松的开源协议之一很多知名开源项目都使用它。MIT License 的核心条款可以概括为允许商用、修改、分发、私人使用。义务在源码或文档中保留原始的版权声明和许可声明。免责软件按“原样”提供作者不对使用后果承担任何担保责任。所以如果你基于这个库做二次开发可以直接把它集成到自己的商业产品里不需要开源自己的代码。唯一要注意的是不要删除 LICENSE 文件不要声称这个库是你原创的。另外提醒一句MIT 只是这个库本身的许可证。如果你的项目依赖了其他协议的第三方组件比如某个 SDK 是 Apache 2.0 或 GPL那整个项目的许可证兼容性需要单独评估。2. 环境准备与项目结构2.1 运行环境本文示例代码用 Python 编写依赖非常少几乎可以在任何主流操作系统上运行。版本方面没有强绑定尽量以你自己环境中的稳定版本为准。我演示时使用的环境大致如下Python 3.10 及以上pip 和虚拟环境工具 venv系统自带的 SQLite 3唯一的第三方依赖是requests用于发起 HTTP 请求。如果你不想装任何第三方库把requests换成urllib.request也可以但代码会稍显啰嗦。2.2 项目目录规划代码结构如下byok-ai-search-tracker/ ├── byok_tracker/ │ ├── __init__.py │ ├── key_manager.py # BYOK 密钥管理器 │ ├── models.py # 数据模型 │ ├── storage.py # 追踪记录存储 │ ├── ai_search.py # AI 搜索客户端 │ └── tracker.py # 装饰器追踪工具 ├── examples/ │ └── demo.py # 示例入口 ├── requirements.txt ├── LICENSE # MIT License 文本 └── README.md这个结构把“密钥管理”“搜索调用”“记录存储”拆成了独立模块。后续接入真实业务时每个模块都可以单独替换。比如把 SQLite 存储换成 ClickHouse或者把搜索客户端换成公司内部的 RPC 接口都不影响其他模块。2.3 模块职责划分模块职责依赖key_manager.py注册、获取、移除 API Key提供脱敏方法无models.py定义 SearchRecord 数据结构无storage.py负责建表、写入、查询和统计sqlite3ai_search.py发起 OpenAI 兼容的模型调用记录结果key_manager、storage、modelstracker.py用装饰器包裹任意函数自动记录调用情况storage、models这样分层之后每个文件的逻辑都很短适合新手读懂也适合团队里多人并行维护。3. 核心设计思路3.1 调用链路一次带追踪的 AI 搜索请求调用流程可以用下面的 ASCII 图表示业务调用方 │ query ▼ AISearchClient │ ① 向 KeyManager 取当前 provider 的 Key │ ② 携带 Key 调用 OpenAI 兼容接口 ▼ 模型供应商 / 搜索服务 │ ① 返回响应内容 usage 用量信息 ▼ AISearchClient 清洗结果 │ 组装 SearchRecord写入 Storage ▼ SQLite / 其他存储 │ 查询、统计、导出 ▼ 运营看板 / 对账系统这里有一个容易被忽略的设计点密钥只存在于“请求发出前”到“请求完成后”这个极短的生命周期内。Request 结束之后密钥对象并不跟随记录一起写入存储。存储层永远只保存 provider 名称不保存 Key 明文避免数据库泄漏导致密钥批量泄露。3.2 数据模型设计数据模型是整个追踪系统的地基。字段太少后面统计分析会捉襟见肘字段太多写入路径变重维护成本上升。核心模型SearchRecord使用 Python 的 dataclass 定义from dataclasses import dataclass, field from datetime import datetime dataclass class SearchRecord: request_id: str provider: str model: str query: str response_summary: str prompt_tokens: int completion_tokens: int total_tokens: int latency_ms: int status: str created_at: str field( default_factorylambda: datetime.now().isoformat() )response_summary刻意只存回答的前 200 个字符而不是全文。原因是追踪记录的核心用途是“定位问题”和“统计成本”不是备份搜索结果。存全文会让数据库膨胀很快也无助于解决大多数问题。3.3 密钥安全边界在设计密钥管理器时我给自己定了三条纪律密钥不进日志。打印 Key 时必须通过mask()脱敏。密钥不进追踪库。记录表里只存 provider不存 Key 明文。支持运行时注入。Key 从环境变量或配置中心读取不写死在代码里。这三条纪律看起来简单实际项目中踩坑的往往就是这些细节。比如某次异常堆栈里不小心把请求头打出来了Key 就跟着泄漏了。所以密钥管理器里一定要提供脱敏函数并且养成“任何输出都先过一遍脱敏”的习惯。4. 完整实战实现 byok_tracker下面进入正题把上文的设计变成可运行的代码。请按顺序创建文件。4.1 准备依赖与项目骨架先在项目根目录创建requirements.txtrequests2.31.0创建虚拟环境并安装依赖python3 -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install -r requirements.txt然后把包初始化文件byok_tracker/__init__.py写好from .ai_search import AISearchClient from .key_manager import KeyManager from .storage import SQLiteStorage from .tracker import track_search __version__ 0.1.0 __all__ [AISearchClient, KeyManager, SQLiteStorage, track_search]4.2 实现密钥管理器 key_manager.py密钥管理器的核心功能有三个注册 Key、按 provider 获取 Key、脱敏显示。这里用字典加线程锁实现简单直接。# byok_tracker/key_manager.py import threading from typing import Dict, Optional class KeyManager: BYOK 密钥管理器。 负责保存、校验和分发用户自带的 API Key 同时避免密钥出现在日志和追踪记录中。 def __init__(self) - None: self._keys: Dict[str, str] {} self._lock threading.Lock() def register(self, provider: str, api_key: str) - None: if not api_key or not api_key.strip(): raise ValueError(api_key 不能为空) with self._lock: self._keys[provider] api_key.strip() def get(self, provider: str) - Optional[str]: with self._lock: return self._keys.get(provider) def remove(self, provider: str) - None: with self._lock: self._keys.pop(provider, None) def mask(self, api_key: str) - str: 脱敏显示只保留前 4 位和后 4 位。 if len(api_key) 8: return **** return f{api_key[:4]}****{api_key[-4:]}这里最值得注意的方法是mask()。它不参与业务逻辑却能在排查问题时保护密钥。你可以在日志里放心地输出key_manager.mask(api_key)而不会暴露完整 Key。4.3 实现数据模型 models.py模型文件我们已经在 3.2 节展示过了完整代码如下# byok_tracker/models.py from dataclasses import dataclass, field from datetime import datetime dataclass class SearchRecord: request_id: str provider: str model: str query: str response_summary: str prompt_tokens: int completion_tokens: int total_tokens: int latency_ms: int status: str created_at: str field( default_factorylambda: datetime.now().isoformat() )4.4 实现存储层 storage.py存储层使用 SQLite因为它是 Python 标准库的一部分不需要额外部署服务非常适合本地开发和中小流量场景。需要注意两点所有写操作都加线程锁避免并发写入时出现database is locked。表结构在初始化时通过CREATE TABLE IF NOT EXISTS自动创建。# byok_tracker/storage.py import sqlite3 import threading from typing import Dict, List, Optional from .models import SearchRecord class SQLiteStorage: 基于 SQLite 的本地追踪存储。 def __init__(self, db_path: str ai_search_tracker.db) - None: self.db_path db_path self._lock threading.Lock() self._init_db() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self) - None: with self._lock: conn self._connect() try: conn.execute( CREATE TABLE IF NOT EXISTS search_records ( request_id TEXT PRIMARY KEY, provider TEXT NOT NULL, model TEXT NOT NULL, query TEXT, response_summary TEXT, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, latency_ms INTEGER DEFAULT 0, status TEXT, created_at TEXT ) ) conn.commit() finally: conn.close() def insert(self, record: SearchRecord) - None: with self._lock: conn self._connect() try: conn.execute( INSERT INTO search_records ( request_id, provider, model, query, response_summary, prompt_tokens, completion_tokens, total_tokens, latency_ms, status, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( record.request_id, record.provider, record.model, record.query, record.response_summary, record.prompt_tokens, record.completion_tokens, record.total_tokens, record.latency_ms, record.status, record.created_at, ), ) conn.commit() finally: conn.close() def query( self, start: Optional[str] None, end: Optional[str] None, provider: Optional[str] None, limit: int 100, ) - List[Dict]: sql SELECT * FROM search_records WHERE 11 params: list [] if start: sql AND created_at ? params.append(start) if end: sql AND created_at ? params.append(end) if provider: sql AND provider ? params.append(provider) sql ORDER BY created_at DESC LIMIT ? params.append(limit) with self._lock: conn self._connect() try: rows conn.execute(sql, params).fetchall() return [dict(row) for row in rows] finally: conn.close() def stats(self) - Dict: with self._lock: conn self._connect() try: total conn.execute( SELECT COUNT(*) AS cnt, SUM(total_tokens) AS tokens, SUM(latency_ms) AS latency FROM search_records ).fetchone() by_provider conn.execute( SELECT provider, COUNT(*) AS cnt FROM search_records GROUP BY provider ).fetchall() return { total_requests: total[cnt] or 0, total_tokens: total[tokens] or 0, total_latency_ms: total[latency] or 0, by_provider: { row[provider]: row[cnt] for row in by_provider }, } finally: conn.close()query()方法支持按时间范围、provider 和数量限制过滤正好对应对账和排查两个核心场景。stats()方法则直接返回汇总指标方便接到监控面板上。测试时可以把db_path设为:memory:SQLite 会在内存中创建临时库适合跑单元测试。4.5 实现 AI 搜索客户端 ai_search.py搜索客户端是整个库的“门面”。它负责取出 Key、发起请求、解析响应、组装记录、写入存储。为了让示例具备通用性这里采用 OpenAI 兼容的chat/completions接口这套接口目前已经被大量模型供应商兼容。# byok_tracker/ai_search.py import time import uuid from typing import Dict, Optional import requests from .key_manager import KeyManager from .models import SearchRecord from .storage import SQLiteStorage class AISearchClient: AI 搜索客户端。 使用 OpenAI 兼容的聊天补全接口调用方可以替换为 DeepSeek、Qwen、vLLM 等任意兼容服务。 def __init__( self, keys: KeyManager, storage: SQLiteStorage, endpoint: str https://api.openai.com/v1/chat/completions, model: str gpt-4o-mini, timeout: int 60, ) - None: self.keys keys self.storage storage self.endpoint endpoint self.model model self.timeout timeout def search( self, query: str, provider: str default, system_prompt: str 你是一个搜索助手请结合已有知识回答用户问题。, ) - Dict: api_key self.keys.get(provider) if not api_key: raise RuntimeError( fprovider [{provider}] 未注册 API Key ) headers { Authorization: fBearer {api_key}, Content-Type: application/json, } payload { model: self.model, messages: [ {role: system, content: system_prompt}, {role: user, content: query}, ], temperature: 0.2, } start time.monotonic() try: resp requests.post( self.endpoint, headersheaders, jsonpayload, timeoutself.timeout, ) latency_ms int((time.monotonic() - start) * 1000) resp.raise_for_status() data resp.json() except requests.RequestException as exc: latency_ms int((time.monotonic() - start) * 1000) status_code ( exc.response.status_code if exc.response is not None else network_error ) self._save_error_record( queryquery, providerprovider, latency_mslatency_ms, errorfhttp_{status_code}, ) raise # 解析用量信息OpenAI 兼容接口结构基本一致 usage data.get(usage, {}) prompt_tokens usage.get(prompt_tokens, 0) completion_tokens usage.get(completion_tokens, 0) total_tokens usage.get(total_tokens, 0) content data[choices][0][message][content] record SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelself.model, queryquery, response_summarycontent[:200], prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, total_tokenstotal_tokens, latency_mslatency_ms, statussuccess, ) self.storage.insert(record) return { request_id: record.request_id, answer: content, usage: { prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, total_tokens: total_tokens, }, latency_ms: latency_ms, } def _save_error_record( self, query: str, provider: str, latency_ms: int, error: str, ) - None: record SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelself.model, queryquery, response_summaryerror, prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statuserror, ) self.storage.insert(record)这里有两个容易踩坑的地方提前说明超时时间不能太短。大模型接口的响应经常要十几秒甚至更久timeout60是一个相对保守的起点。如果设成 5 秒高峰期会误报大量超时。错误也必须落库。很多团队只记录成功调用导致故障复盘时没有数据支撑。上面的_save_error_record()保证失败请求同样有记录只是statuserrortoken 相关字段置 0。4.6 实现装饰器追踪 tracker.py有些场景下你并不想直接使用AISearchClient而是想追踪自己封装的搜索函数。这时可以用装饰器track_search把任意函数的调用情况自动记录下来。# byok_tracker/tracker.py import functools import time import uuid from typing import Callable from .models import SearchRecord from .storage import SQLiteStorage def track_search( storage: SQLiteStorage, provider: str default, model: str unknown, ): 装饰器记录任意搜索/LLM 调用函数的执行情况。 def decorator(func: Callable) - Callable: functools.wraps(func) def wrapper(*args, **kwargs): query kwargs.get(query) or (args[0] if args else ) start time.monotonic() try: result func(*args, **kwargs) latency_ms int((time.monotonic() - start) * 1000) text result if isinstance(result, str) else str(result) storage.insert(SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelmodel, querystr(query), response_summarytext[:200], prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statussuccess, )) return result except Exception as exc: latency_ms int((time.monotonic() - start) * 1000) storage.insert(SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelmodel, querystr(query), response_summaryferror: {exc}, prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statuserror, )) raise return wrapper return decorator使用方式很简单track_search(storageSQLiteStorage(search.db), providerfinance, modelmy_search_v1) def my_search(query: str) - str: # 这里可以是 RAG、向量检索、或者调用你们内部搜索服务 return 搜索结果摘要装饰器方案的优势是侵入性小适合在一个函数上快速开启追踪不用改函数内部逻辑。4.7 示例入口与运行验证最后写一个示例脚本把整个链路串起来。# examples/demo.py import os from byok_tracker import AISearchClient, KeyManager, SQLiteStorage def main(): # 1. 初始化组件 keys KeyManager() storage SQLiteStorage(demo_tracker.db) # 2. 注册用户自带的 API KeyBYOK api_key os.getenv(AI_API_KEY, ) if not api_key: raise SystemExit( 请先设置环境变量 AI_API_KEY例如 export AI_API_KEYsk-xxx ) keys.register(default, api_key) # 3. 创建搜索客户端 client AISearchClient( keyskeys, storagestorage, endpointos.getenv( AI_ENDPOINT, https://api.openai.com/v1/chat/completions, ), modelos.getenv(AI_MODEL, gpt-4o-mini), ) # 4. 执行一次搜索 result client.search(用一句话解释什么是 BYOK) print(回答:, result[answer]) print(用量:, result[usage]) print(耗时:, result[latency_ms], ms) print(请求ID:, result[request_id]) # 5. 查看追踪结果 print(\n最近 10 条追踪记录) for row in storage.query(limit10): print(row) print(\n汇总统计) print(storage.stats()) if __name__ __main__: main()运行前先注册密钥export AI_API_KEYsk-你的密钥 python examples/demo.py如果你的服务商接口地址和模型名不同通过环境变量覆盖即可export AI_ENDPOINThttps://your-provider.example.com/v1/chat/completions export AI_MODELyour-model-name python examples/demo.py输出类似下面这样实际 token 数量和耗时取决于模型与网络回答: BYOK 是 Bring Your Own Key 的缩写意思是用户自己提供密钥... 用量: {prompt_tokens: 35, completion_tokens: 120, total_tokens: 155} 耗时: 820 ms 请求ID: 3f2b8c9e-... 最近 10 条追踪记录 [{request_id: 3f2b8c9e-..., provider: default, ...}] 汇总统计 {total_requests: 1, total_tokens: 155, total_latency_ms: 820, by_provider: {default: 1}}到这里一个带 BYOK 密钥管理和调用追踪的 AI 搜索库就已经完整跑通了。5. 常见问题与排查思路下面整理的是这类库在实际接入时最常遇到的问题每条都给出排查方向。问题现象常见原因解决思路接口返回 401/403API Key 错误、已过期或权限不足检查 Key 是否有效查看供应商控制台的权限配置报错 “provider 未注册 API Key”KeyManager.register()没有被调用确认初始化流程中注册了对应 provider追踪表为空写入前抛异常异常没有被记录检查search()是否在异常分支调用了_save_error_record()SQLite 报 database is locked多个进程同时写同一个库文件使用 WAL 模式或改用 PostgreSQL/ClickHouse密钥出现在日志中打印请求头或异常对象全局搜索Authorization相关打印强制使用mask()pip 安装依赖失败网络源不可达切换可用的 pip 镜像源后重试Docker 拉取镜像报failed to resolve reference docker.io/library/python:3.10镜像名拼写错误或网络无法访问 Docker Hub检查镜像名或配置可用的镜像加速源5.1 API Key 未生效返回 401/403这类问题最常见。排查顺序建议是确认环境变量是否真的传进来了可以在脚本里临时打印keys.mask(api_key)。确认 Key 没有多余空格register()里已经做了strip()但如果从文件读取 Key 可能带换行符。确认该 Key 在供应商后台有对应的模型访问权限。很多平台默认只给部分模型开放权限。确认接口地址与 Key 是否匹配不同平台的 endpoint 差异很大。5.2 追踪记录没有写入数据库如果接口调用成功但数据库里查不到记录优先检查是不是异常分支没有落库。我们设计search()时成功和失败都会写入但如果你的业务代码里在client.search()外层又包了一层 try-except并且异常被吞掉就会出现“看起来没记录”的情况。5.3 SQLite 并发写入报 database is locked我们的示例代码在单进程内用线程锁规避了这个问题。但如果是多个进程同时写入同一个 SQLite 文件线程锁就不够用了。解决办法有三种开启 SQLite WAL 模式可以显著降低读写锁冲突。把所有写入操作收敛到同一个进程通过消息队列异步落库。数据量上来后把存储层替换为 PostgreSQL、MySQL 或 ClickHouse。5.4 密钥泄漏到日志或追踪记录中这是最需要警惕的问题。排查清单一共有三步全局搜索代码里的logger.info、print确认没有直接打印 Key 或请求头。检查日志采集系统里是否保留了请求体。访问追踪库的权限必须收敛只有对账、审计相关人员能查search_records表。6. 最佳实践与工程建议6.1 密钥管理环境变量、KMS 与最小权限示例代码从环境变量读取 Key这只是方便演示。生产环境更推荐的做法是Key 存放在公司内部的密钥管理系统如 Vault、云厂商 KMS中服务启动时通过 SDK 拉取。每个业务方使用独立的 Key 或独立的 provider 标识便于成本归因和故障隔离。遵循最小权限原则接口只传需要的模型权限不要给一个“全模型可用”的超管 Key。定期轮换 Key轮换时先注册新 Key再移除旧 Key避免服务中断。6.2 成本控制与配额告警接入 BYOK 模式后成本责任分散到各个调用方但作为平台方仍然要做总量管控。建议在存储层之上增加两个能力配额检查在search()入口处检查 provider 当日 token 消耗超过阈值直接拒绝。异步对账任务每小时跑一次stats()输出各 provider 的 token 消耗报表超过预算自动告警。6.3 存储扩展与可观测性SQLite 用于开发和小流量没问题但生产环境建议替换为独立数据库。替换时只需要重新实现SQLiteStorage中insert、query、stats三个方法上层代码可以完全不动。此外建议给每次调用增加一个结构化日志输出格式类似request_idxxx providerfinance modelgpt-4o-mini tokens155 latency_ms820 statussuccess这样即使数据库出问题你仍然能从日志链路还原调用
BYOK与AI搜索追踪：Python库设计与实现
很多做 AI 应用的团队都会遇到同一个组合需求让用户“带上自己的 API Key”来调用大模型搜索能力同时平台方又必须把每一次调用的账记清楚——谁调的、用的哪个模型、花了多少 token、响应延迟多高、有没有报错。前者是 BYOKBring Your Own Key自带密钥模式后者是 AI 搜索追踪。这两件事看上去不相关落地时却必须放在一起设计。本文将围绕一个可运行的最小实现完整拆解“BYOK AI 搜索追踪”库的设计与编码过程内容包括密钥管理器、SQLite 存储层、OpenAI 兼容的搜索客户端、装饰器追踪工具以及 MIT License 相关的合规注意点。无论你是想给现有项目接入 AI 搜索功能还是想把 LLM 调用做成可审计、可计费的后端能力这篇文章都能提供一套可以直接复用的思路。1. 背景与核心概念1.1 什么是 BYOKBYOK 全称是 Bring Your Own Key直译就是“带上你自己的密钥”。这个概念最早在云服务和 SaaS 领域比较常见比如企业把自有的加密密钥托管到云厂商的 KMSKey Management Service里密钥的生成、轮换和管理权仍然掌握在自己手中。到了大模型和 AI 搜索场景BYOK 的含义就变得更贴近日常开发库或平台不内置任何供应商的 API Key由调用方在运行时把 Key 注入进来。你可以用自己的 OpenAI Key、DeepSeek Key、通义千问 Key也可以是公司内部统一申请的企业 Key。BYOK 模式之所以受欢迎核心原因有几个成本归属清晰谁调用谁付费。用平台统一转发的 Key费用全部记在平台头上月底对账容易扯皮。数据边界可控Key 不经过第三方中转服务请求直接从你的服务发往模型供应商敏感字段的暴露面更小。切换灵活今天用 A 厂商的模型明天换 B 厂商只需要换 Key 和接口地址不需要改业务代码。合规友好很多企业客户不允许自己公司的密钥出现在另一个服务商的数据库里BYOK 天然规避了这个问题。不过 BYOK 也带来一个副作用调用方变成“多租户”的每个租户消耗的 token 不同、报错不同、用量不同。如果不去追踪出问题的时候连“这波开销是哪个客户产生的”都查不到。所以 BYOK 与追踪通常是一起出现的。1.2 为什么要做 AI 搜索追踪AI 搜索不是简单的关键词匹配它背后往往是一次完整的 LLM 调用模型接收用户的搜索意图结合上下文生成回答和引用。这类调用的成本、延迟和成功率高度依赖模型、提示词长度和供应商状态。追踪层需要记录的关键维度至少包括维度说明request_id每次调用的唯一标识用于问题定位和对账provider使用的是哪家供应商 / 哪个租户的 Keymodel实际使用的模型名例如 gpt-4o-miniquery用户搜索的原始输入tokensprompt_tokens、completion_tokens、total_tokenslatency_ms从发起到拿到响应的总耗时statussuccess / error以及错误码created_at调用发生时间有了这些数据你能做三件事成本归因按 provider 汇总 token 消耗精确算出每个业务方、每个客户的使用量。质量排查用户投诉“搜索变慢了”先看延迟曲线和错误率而不是盲目调参数。限流与预算预警当某个 Key 的日消耗接近阈值时自动熔断或告警。需要强调的是追踪不等于把用户输入的明文原样存起来。查询内容可能包含业务敏感信息设计存储时要考虑脱敏、自动清理和访问权限。1.3 MIT License 能做什么、不能做什么项目采用 MIT License意味着源码允许任何人免费使用、修改、分发甚至闭源商用。MIT 是目前最宽松的开源协议之一很多知名开源项目都使用它。MIT License 的核心条款可以概括为允许商用、修改、分发、私人使用。义务在源码或文档中保留原始的版权声明和许可声明。免责软件按“原样”提供作者不对使用后果承担任何担保责任。所以如果你基于这个库做二次开发可以直接把它集成到自己的商业产品里不需要开源自己的代码。唯一要注意的是不要删除 LICENSE 文件不要声称这个库是你原创的。另外提醒一句MIT 只是这个库本身的许可证。如果你的项目依赖了其他协议的第三方组件比如某个 SDK 是 Apache 2.0 或 GPL那整个项目的许可证兼容性需要单独评估。2. 环境准备与项目结构2.1 运行环境本文示例代码用 Python 编写依赖非常少几乎可以在任何主流操作系统上运行。版本方面没有强绑定尽量以你自己环境中的稳定版本为准。我演示时使用的环境大致如下Python 3.10 及以上pip 和虚拟环境工具 venv系统自带的 SQLite 3唯一的第三方依赖是requests用于发起 HTTP 请求。如果你不想装任何第三方库把requests换成urllib.request也可以但代码会稍显啰嗦。2.2 项目目录规划代码结构如下byok-ai-search-tracker/ ├── byok_tracker/ │ ├── __init__.py │ ├── key_manager.py # BYOK 密钥管理器 │ ├── models.py # 数据模型 │ ├── storage.py # 追踪记录存储 │ ├── ai_search.py # AI 搜索客户端 │ └── tracker.py # 装饰器追踪工具 ├── examples/ │ └── demo.py # 示例入口 ├── requirements.txt ├── LICENSE # MIT License 文本 └── README.md这个结构把“密钥管理”“搜索调用”“记录存储”拆成了独立模块。后续接入真实业务时每个模块都可以单独替换。比如把 SQLite 存储换成 ClickHouse或者把搜索客户端换成公司内部的 RPC 接口都不影响其他模块。2.3 模块职责划分模块职责依赖key_manager.py注册、获取、移除 API Key提供脱敏方法无models.py定义 SearchRecord 数据结构无storage.py负责建表、写入、查询和统计sqlite3ai_search.py发起 OpenAI 兼容的模型调用记录结果key_manager、storage、modelstracker.py用装饰器包裹任意函数自动记录调用情况storage、models这样分层之后每个文件的逻辑都很短适合新手读懂也适合团队里多人并行维护。3. 核心设计思路3.1 调用链路一次带追踪的 AI 搜索请求调用流程可以用下面的 ASCII 图表示业务调用方 │ query ▼ AISearchClient │ ① 向 KeyManager 取当前 provider 的 Key │ ② 携带 Key 调用 OpenAI 兼容接口 ▼ 模型供应商 / 搜索服务 │ ① 返回响应内容 usage 用量信息 ▼ AISearchClient 清洗结果 │ 组装 SearchRecord写入 Storage ▼ SQLite / 其他存储 │ 查询、统计、导出 ▼ 运营看板 / 对账系统这里有一个容易被忽略的设计点密钥只存在于“请求发出前”到“请求完成后”这个极短的生命周期内。Request 结束之后密钥对象并不跟随记录一起写入存储。存储层永远只保存 provider 名称不保存 Key 明文避免数据库泄漏导致密钥批量泄露。3.2 数据模型设计数据模型是整个追踪系统的地基。字段太少后面统计分析会捉襟见肘字段太多写入路径变重维护成本上升。核心模型SearchRecord使用 Python 的 dataclass 定义from dataclasses import dataclass, field from datetime import datetime dataclass class SearchRecord: request_id: str provider: str model: str query: str response_summary: str prompt_tokens: int completion_tokens: int total_tokens: int latency_ms: int status: str created_at: str field( default_factorylambda: datetime.now().isoformat() )response_summary刻意只存回答的前 200 个字符而不是全文。原因是追踪记录的核心用途是“定位问题”和“统计成本”不是备份搜索结果。存全文会让数据库膨胀很快也无助于解决大多数问题。3.3 密钥安全边界在设计密钥管理器时我给自己定了三条纪律密钥不进日志。打印 Key 时必须通过mask()脱敏。密钥不进追踪库。记录表里只存 provider不存 Key 明文。支持运行时注入。Key 从环境变量或配置中心读取不写死在代码里。这三条纪律看起来简单实际项目中踩坑的往往就是这些细节。比如某次异常堆栈里不小心把请求头打出来了Key 就跟着泄漏了。所以密钥管理器里一定要提供脱敏函数并且养成“任何输出都先过一遍脱敏”的习惯。4. 完整实战实现 byok_tracker下面进入正题把上文的设计变成可运行的代码。请按顺序创建文件。4.1 准备依赖与项目骨架先在项目根目录创建requirements.txtrequests2.31.0创建虚拟环境并安装依赖python3 -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install -r requirements.txt然后把包初始化文件byok_tracker/__init__.py写好from .ai_search import AISearchClient from .key_manager import KeyManager from .storage import SQLiteStorage from .tracker import track_search __version__ 0.1.0 __all__ [AISearchClient, KeyManager, SQLiteStorage, track_search]4.2 实现密钥管理器 key_manager.py密钥管理器的核心功能有三个注册 Key、按 provider 获取 Key、脱敏显示。这里用字典加线程锁实现简单直接。# byok_tracker/key_manager.py import threading from typing import Dict, Optional class KeyManager: BYOK 密钥管理器。 负责保存、校验和分发用户自带的 API Key 同时避免密钥出现在日志和追踪记录中。 def __init__(self) - None: self._keys: Dict[str, str] {} self._lock threading.Lock() def register(self, provider: str, api_key: str) - None: if not api_key or not api_key.strip(): raise ValueError(api_key 不能为空) with self._lock: self._keys[provider] api_key.strip() def get(self, provider: str) - Optional[str]: with self._lock: return self._keys.get(provider) def remove(self, provider: str) - None: with self._lock: self._keys.pop(provider, None) def mask(self, api_key: str) - str: 脱敏显示只保留前 4 位和后 4 位。 if len(api_key) 8: return **** return f{api_key[:4]}****{api_key[-4:]}这里最值得注意的方法是mask()。它不参与业务逻辑却能在排查问题时保护密钥。你可以在日志里放心地输出key_manager.mask(api_key)而不会暴露完整 Key。4.3 实现数据模型 models.py模型文件我们已经在 3.2 节展示过了完整代码如下# byok_tracker/models.py from dataclasses import dataclass, field from datetime import datetime dataclass class SearchRecord: request_id: str provider: str model: str query: str response_summary: str prompt_tokens: int completion_tokens: int total_tokens: int latency_ms: int status: str created_at: str field( default_factorylambda: datetime.now().isoformat() )4.4 实现存储层 storage.py存储层使用 SQLite因为它是 Python 标准库的一部分不需要额外部署服务非常适合本地开发和中小流量场景。需要注意两点所有写操作都加线程锁避免并发写入时出现database is locked。表结构在初始化时通过CREATE TABLE IF NOT EXISTS自动创建。# byok_tracker/storage.py import sqlite3 import threading from typing import Dict, List, Optional from .models import SearchRecord class SQLiteStorage: 基于 SQLite 的本地追踪存储。 def __init__(self, db_path: str ai_search_tracker.db) - None: self.db_path db_path self._lock threading.Lock() self._init_db() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self) - None: with self._lock: conn self._connect() try: conn.execute( CREATE TABLE IF NOT EXISTS search_records ( request_id TEXT PRIMARY KEY, provider TEXT NOT NULL, model TEXT NOT NULL, query TEXT, response_summary TEXT, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, latency_ms INTEGER DEFAULT 0, status TEXT, created_at TEXT ) ) conn.commit() finally: conn.close() def insert(self, record: SearchRecord) - None: with self._lock: conn self._connect() try: conn.execute( INSERT INTO search_records ( request_id, provider, model, query, response_summary, prompt_tokens, completion_tokens, total_tokens, latency_ms, status, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( record.request_id, record.provider, record.model, record.query, record.response_summary, record.prompt_tokens, record.completion_tokens, record.total_tokens, record.latency_ms, record.status, record.created_at, ), ) conn.commit() finally: conn.close() def query( self, start: Optional[str] None, end: Optional[str] None, provider: Optional[str] None, limit: int 100, ) - List[Dict]: sql SELECT * FROM search_records WHERE 11 params: list [] if start: sql AND created_at ? params.append(start) if end: sql AND created_at ? params.append(end) if provider: sql AND provider ? params.append(provider) sql ORDER BY created_at DESC LIMIT ? params.append(limit) with self._lock: conn self._connect() try: rows conn.execute(sql, params).fetchall() return [dict(row) for row in rows] finally: conn.close() def stats(self) - Dict: with self._lock: conn self._connect() try: total conn.execute( SELECT COUNT(*) AS cnt, SUM(total_tokens) AS tokens, SUM(latency_ms) AS latency FROM search_records ).fetchone() by_provider conn.execute( SELECT provider, COUNT(*) AS cnt FROM search_records GROUP BY provider ).fetchall() return { total_requests: total[cnt] or 0, total_tokens: total[tokens] or 0, total_latency_ms: total[latency] or 0, by_provider: { row[provider]: row[cnt] for row in by_provider }, } finally: conn.close()query()方法支持按时间范围、provider 和数量限制过滤正好对应对账和排查两个核心场景。stats()方法则直接返回汇总指标方便接到监控面板上。测试时可以把db_path设为:memory:SQLite 会在内存中创建临时库适合跑单元测试。4.5 实现 AI 搜索客户端 ai_search.py搜索客户端是整个库的“门面”。它负责取出 Key、发起请求、解析响应、组装记录、写入存储。为了让示例具备通用性这里采用 OpenAI 兼容的chat/completions接口这套接口目前已经被大量模型供应商兼容。# byok_tracker/ai_search.py import time import uuid from typing import Dict, Optional import requests from .key_manager import KeyManager from .models import SearchRecord from .storage import SQLiteStorage class AISearchClient: AI 搜索客户端。 使用 OpenAI 兼容的聊天补全接口调用方可以替换为 DeepSeek、Qwen、vLLM 等任意兼容服务。 def __init__( self, keys: KeyManager, storage: SQLiteStorage, endpoint: str https://api.openai.com/v1/chat/completions, model: str gpt-4o-mini, timeout: int 60, ) - None: self.keys keys self.storage storage self.endpoint endpoint self.model model self.timeout timeout def search( self, query: str, provider: str default, system_prompt: str 你是一个搜索助手请结合已有知识回答用户问题。, ) - Dict: api_key self.keys.get(provider) if not api_key: raise RuntimeError( fprovider [{provider}] 未注册 API Key ) headers { Authorization: fBearer {api_key}, Content-Type: application/json, } payload { model: self.model, messages: [ {role: system, content: system_prompt}, {role: user, content: query}, ], temperature: 0.2, } start time.monotonic() try: resp requests.post( self.endpoint, headersheaders, jsonpayload, timeoutself.timeout, ) latency_ms int((time.monotonic() - start) * 1000) resp.raise_for_status() data resp.json() except requests.RequestException as exc: latency_ms int((time.monotonic() - start) * 1000) status_code ( exc.response.status_code if exc.response is not None else network_error ) self._save_error_record( queryquery, providerprovider, latency_mslatency_ms, errorfhttp_{status_code}, ) raise # 解析用量信息OpenAI 兼容接口结构基本一致 usage data.get(usage, {}) prompt_tokens usage.get(prompt_tokens, 0) completion_tokens usage.get(completion_tokens, 0) total_tokens usage.get(total_tokens, 0) content data[choices][0][message][content] record SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelself.model, queryquery, response_summarycontent[:200], prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, total_tokenstotal_tokens, latency_mslatency_ms, statussuccess, ) self.storage.insert(record) return { request_id: record.request_id, answer: content, usage: { prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, total_tokens: total_tokens, }, latency_ms: latency_ms, } def _save_error_record( self, query: str, provider: str, latency_ms: int, error: str, ) - None: record SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelself.model, queryquery, response_summaryerror, prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statuserror, ) self.storage.insert(record)这里有两个容易踩坑的地方提前说明超时时间不能太短。大模型接口的响应经常要十几秒甚至更久timeout60是一个相对保守的起点。如果设成 5 秒高峰期会误报大量超时。错误也必须落库。很多团队只记录成功调用导致故障复盘时没有数据支撑。上面的_save_error_record()保证失败请求同样有记录只是statuserrortoken 相关字段置 0。4.6 实现装饰器追踪 tracker.py有些场景下你并不想直接使用AISearchClient而是想追踪自己封装的搜索函数。这时可以用装饰器track_search把任意函数的调用情况自动记录下来。# byok_tracker/tracker.py import functools import time import uuid from typing import Callable from .models import SearchRecord from .storage import SQLiteStorage def track_search( storage: SQLiteStorage, provider: str default, model: str unknown, ): 装饰器记录任意搜索/LLM 调用函数的执行情况。 def decorator(func: Callable) - Callable: functools.wraps(func) def wrapper(*args, **kwargs): query kwargs.get(query) or (args[0] if args else ) start time.monotonic() try: result func(*args, **kwargs) latency_ms int((time.monotonic() - start) * 1000) text result if isinstance(result, str) else str(result) storage.insert(SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelmodel, querystr(query), response_summarytext[:200], prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statussuccess, )) return result except Exception as exc: latency_ms int((time.monotonic() - start) * 1000) storage.insert(SearchRecord( request_idstr(uuid.uuid4()), providerprovider, modelmodel, querystr(query), response_summaryferror: {exc}, prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency_ms, statuserror, )) raise return wrapper return decorator使用方式很简单track_search(storageSQLiteStorage(search.db), providerfinance, modelmy_search_v1) def my_search(query: str) - str: # 这里可以是 RAG、向量检索、或者调用你们内部搜索服务 return 搜索结果摘要装饰器方案的优势是侵入性小适合在一个函数上快速开启追踪不用改函数内部逻辑。4.7 示例入口与运行验证最后写一个示例脚本把整个链路串起来。# examples/demo.py import os from byok_tracker import AISearchClient, KeyManager, SQLiteStorage def main(): # 1. 初始化组件 keys KeyManager() storage SQLiteStorage(demo_tracker.db) # 2. 注册用户自带的 API KeyBYOK api_key os.getenv(AI_API_KEY, ) if not api_key: raise SystemExit( 请先设置环境变量 AI_API_KEY例如 export AI_API_KEYsk-xxx ) keys.register(default, api_key) # 3. 创建搜索客户端 client AISearchClient( keyskeys, storagestorage, endpointos.getenv( AI_ENDPOINT, https://api.openai.com/v1/chat/completions, ), modelos.getenv(AI_MODEL, gpt-4o-mini), ) # 4. 执行一次搜索 result client.search(用一句话解释什么是 BYOK) print(回答:, result[answer]) print(用量:, result[usage]) print(耗时:, result[latency_ms], ms) print(请求ID:, result[request_id]) # 5. 查看追踪结果 print(\n最近 10 条追踪记录) for row in storage.query(limit10): print(row) print(\n汇总统计) print(storage.stats()) if __name__ __main__: main()运行前先注册密钥export AI_API_KEYsk-你的密钥 python examples/demo.py如果你的服务商接口地址和模型名不同通过环境变量覆盖即可export AI_ENDPOINThttps://your-provider.example.com/v1/chat/completions export AI_MODELyour-model-name python examples/demo.py输出类似下面这样实际 token 数量和耗时取决于模型与网络回答: BYOK 是 Bring Your Own Key 的缩写意思是用户自己提供密钥... 用量: {prompt_tokens: 35, completion_tokens: 120, total_tokens: 155} 耗时: 820 ms 请求ID: 3f2b8c9e-... 最近 10 条追踪记录 [{request_id: 3f2b8c9e-..., provider: default, ...}] 汇总统计 {total_requests: 1, total_tokens: 155, total_latency_ms: 820, by_provider: {default: 1}}到这里一个带 BYOK 密钥管理和调用追踪的 AI 搜索库就已经完整跑通了。5. 常见问题与排查思路下面整理的是这类库在实际接入时最常遇到的问题每条都给出排查方向。问题现象常见原因解决思路接口返回 401/403API Key 错误、已过期或权限不足检查 Key 是否有效查看供应商控制台的权限配置报错 “provider 未注册 API Key”KeyManager.register()没有被调用确认初始化流程中注册了对应 provider追踪表为空写入前抛异常异常没有被记录检查search()是否在异常分支调用了_save_error_record()SQLite 报 database is locked多个进程同时写同一个库文件使用 WAL 模式或改用 PostgreSQL/ClickHouse密钥出现在日志中打印请求头或异常对象全局搜索Authorization相关打印强制使用mask()pip 安装依赖失败网络源不可达切换可用的 pip 镜像源后重试Docker 拉取镜像报failed to resolve reference docker.io/library/python:3.10镜像名拼写错误或网络无法访问 Docker Hub检查镜像名或配置可用的镜像加速源5.1 API Key 未生效返回 401/403这类问题最常见。排查顺序建议是确认环境变量是否真的传进来了可以在脚本里临时打印keys.mask(api_key)。确认 Key 没有多余空格register()里已经做了strip()但如果从文件读取 Key 可能带换行符。确认该 Key 在供应商后台有对应的模型访问权限。很多平台默认只给部分模型开放权限。确认接口地址与 Key 是否匹配不同平台的 endpoint 差异很大。5.2 追踪记录没有写入数据库如果接口调用成功但数据库里查不到记录优先检查是不是异常分支没有落库。我们设计search()时成功和失败都会写入但如果你的业务代码里在client.search()外层又包了一层 try-except并且异常被吞掉就会出现“看起来没记录”的情况。5.3 SQLite 并发写入报 database is locked我们的示例代码在单进程内用线程锁规避了这个问题。但如果是多个进程同时写入同一个 SQLite 文件线程锁就不够用了。解决办法有三种开启 SQLite WAL 模式可以显著降低读写锁冲突。把所有写入操作收敛到同一个进程通过消息队列异步落库。数据量上来后把存储层替换为 PostgreSQL、MySQL 或 ClickHouse。5.4 密钥泄漏到日志或追踪记录中这是最需要警惕的问题。排查清单一共有三步全局搜索代码里的logger.info、print确认没有直接打印 Key 或请求头。检查日志采集系统里是否保留了请求体。访问追踪库的权限必须收敛只有对账、审计相关人员能查search_records表。6. 最佳实践与工程建议6.1 密钥管理环境变量、KMS 与最小权限示例代码从环境变量读取 Key这只是方便演示。生产环境更推荐的做法是Key 存放在公司内部的密钥管理系统如 Vault、云厂商 KMS中服务启动时通过 SDK 拉取。每个业务方使用独立的 Key 或独立的 provider 标识便于成本归因和故障隔离。遵循最小权限原则接口只传需要的模型权限不要给一个“全模型可用”的超管 Key。定期轮换 Key轮换时先注册新 Key再移除旧 Key避免服务中断。6.2 成本控制与配额告警接入 BYOK 模式后成本责任分散到各个调用方但作为平台方仍然要做总量管控。建议在存储层之上增加两个能力配额检查在search()入口处检查 provider 当日 token 消耗超过阈值直接拒绝。异步对账任务每小时跑一次stats()输出各 provider 的 token 消耗报表超过预算自动告警。6.3 存储扩展与可观测性SQLite 用于开发和小流量没问题但生产环境建议替换为独立数据库。替换时只需要重新实现SQLiteStorage中insert、query、stats三个方法上层代码可以完全不动。此外建议给每次调用增加一个结构化日志输出格式类似request_idxxx providerfinance modelgpt-4o-mini tokens155 latency_ms820 statussuccess这样即使数据库出问题你仍然能从日志链路还原调用
