这周我在把 Dwarkesh Patel 与 Beren Millidge、John Schulman、Charlie O'Neill 那场关于递归自我改进（RSI）的对谈，做成一个能跑起来的研究情报库。Key 和 Base URL 直接走 TaoToken，官网入口先放这里：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_intro 。选它的原因很实际——这个项目真正烧 Token 的只有两段脚本，批量向量化和检索后的问答生成，按量计费比包月划算，而且不用改 OpenAI 兼容 SDK 的调用习惯。
第一版脚本我偷懒，把整篇逐字稿拼成一个长字符串直接丢给问答接口，立刻撞上上下文长度报错；改成按发言人和轮次切块、批量算向量之后，又因为并发开太高吃到 429。这两个坑修完，从原始逐字稿到可检索结果，整条链路稳定在几分钟内跑完，检索命中率也明显好于整篇切碎的做法。下面把可复制的分块策略、向量化配置和检索脚本完整写一遍。
1. 为什么 RSI 这种长对谈一定要切「发言人 + 轮次」
RSI 这个话题的对谈和普通技术分享不一样。三个人对同一件事的立场是有差异的：一位更关注时间线和瓶颈在哪，一位更关注验证手段和安全边界，还有一位从训练基础设施和算力角度切入。如果按固定字符数硬切，一个 600 字的回答会被拦腰截断，检索时你只能拿到半句话，根本判断不出是谁在什么语境下说的。
所以我的分块规则是三层：
第一层按发言人切。逐字稿里每个人都有稳定的标签格式，用它做一级边界，保证任何一块都不会横跨两个人。
第二层按轮次切。一轮回答如果特别长，就在段落边界处再切一次，目标长度 350 到 500 字，相邻块保留 80 字左右的重叠，避免观点被切断。
第三层给每块打元数据。至少要留下speaker、turn_id、字符起止位置和一个粗粒度的topic标签。元数据不是为了好看，是为了后面做过滤——比如你只想看训练基础设施相关的观点时，可以直接按 speaker 过滤，而不是让向量检索去碰运气。
# chunk_transcript.py import re import json from pathlib import Path SPEAKERS = ["Dwarkesh Patel", "Beren Millidge", "John Schulman", "Charlie O'Neill"] SPEAKER_RE = re.compile(r"^\s*(" + "|".join(re.escape(s) for s in SPEAKERS) + r")\s*[:：]") TARGET_LEN = 450 OVERLAP = 80 def split_by_speaker(text: str): """把逐字稿切成 [(speaker, body), ...]，保留轮次顺序。""" turns, cur_speaker, buf = [], None, [] for line in text.splitlines(): m = SPEAKER_RE.match(line) if m: if cur_speaker and buf: turns.append((cur_speaker, "\n".join(buf).strip())) cur_speaker, buf = m.group(1), [SPEAKER_RE.sub("", line).strip()] else: buf.append(line) if cur_speaker and buf: turns.append((cur_speaker, "\n".join(buf).strip())) return turns def window_split(body: str, target=TARGET_LEN, overlap=OVERLAP): """按段落边界滑窗，避免在句子中间断开。""" paras = [p for p in body.split("\n") if p.strip()] chunks, buf, size = [], [], 0 for p in paras: if size + len(p) > target and buf: chunks.append("\n".join(buf)) tail, tail_size = [], 0 for prev in reversed(buf): if tail_size + len(prev) > overlap: break tail.insert(0, prev) tail_size += len(prev) buf, size = tail, tail_size buf.append(p) size += len(p) if buf: chunks.append("\n".join(buf)) return chunks def build_chunks(raw_path: str, out_path: str): text = Path(raw_path).read_text(encoding="utf-8") records, cid = [], 0 for turn_id, (speaker, body) in enumerate(split_by_speaker(text)): for piece in window_split(body): if len(piece) < 60: # 过滤掉「谢谢」「对」这种无信息块 continue records.append({ "chunk_id": f"c{cid:04d}", "speaker": speaker, "turn_id": turn_id, "text": f"{speaker}：{piece}", # 发言人标签必须带进向量 "char_len": len(piece), }) cid += 1 Path(out_path).write_text( "\n".join(json.dumps(r, ensure_ascii=False) for r in records), encoding="utf-8", ) print(f"chunks={len(records)}") if __name__ == "__main__": build_chunks("transcript.txt", "chunks.jsonl")这里有个容易被忽略的点：拼进向量里的文本一定要带上发言人标签。否则检索出「瓶颈在验证而不是生成」这句话时，你不知道是三位里的哪一位说的，情报价值直接打对折。
2. 向量化脚本接入 TaoToken：Base URL 与批处理重试
这一节的配置是整篇最核心的部分，因为 Token 主要消耗在这里。先去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_key 拿到 Key，控制台里给出两个关键值：Base URL 是https://taotoken.net/api，Key 用来做鉴权。
如果你用的是 OpenAI 兼容的 SDK，需要一个v1版本段，也就是把 Base URL 拼成https://taotoken.net/api/v1；如果是走 Anthropic 协议的工具，直接填https://taotoken.net/api就行，参见第 4 节的 Claude Code 配置。这一点经常有人搞混，配错了表现是 404 或者路径不存在，不是 401。
依赖只有两个：
pip install openai numpy向量化脚本：
# embed_chunks.py import os import json import time from pathlib import Path from openai import OpenAI, RateLimitError, APIStatusError BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") API_KEY = os.environ.get("TAOTOKEN_API_KEY", "YOUR_API_KEY") EMBED_MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small") # 以控制台模型列表为准 client = OpenAI(api_key=API_KEY, base_url=f"{BASE_URL}/v1") BATCH_SIZE = 32 # 一批 32 条，先跑小批量确认稳定再往上调 MAX_RETRY = 5 def embed_batch(texts): delay = 1.0 for attempt in range(MAX_RETRY): try: resp = client.embeddings.create(model=EMBED_MODEL, input=texts) return [d.embedding for d in resp.data] except RateLimitError: # 429：指数退避，别硬刚 time.sleep(delay) delay *= 2 except APIStatusError as e: if e.status_code >= 500: time.sleep(delay) delay *= 2 continue raise raise RuntimeError("批次重试超限，建议调小 BATCH_SIZE") def main(): records = [json.loads(l) for l in Path("chunks.jsonl").read_text(encoding="utf-8").splitlines()] out, total_tokens = [], 0 for i in range(0, len(records), BATCH_SIZE): batch = records[i:i + BATCH_SIZE] vecs = embed_batch([r["text"] for r in batch]) for r, v in zip(batch, vecs): out.append({**r, "vector": v}) total_tokens += sum(len(r["text"]) for r in batch) print(f"progress {i + len(batch)}/{len(records)}") time.sleep(0.2) # 主动让出节奏，减少 429 with open("chunks_embedded.jsonl", "w", encoding="utf-8") as f: for r in out: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"done, chunks={len(out)}, approx_chars={total_tokens}") if __name__ == "__main__": main()运行方式：
export TAOTOKEN_API_KEY=YOUR_API_KEY export TAOTOKEN_BASE_URL=https://taotoken.net/api export EMBED_MODEL=text-embedding-3-small python embed_chunks.py三个实操经验：
第一，BATCH_SIZE不要一上来就设 128。对谈逐字稿的单块长度差异很大，有的块 60 字，有的块 500 字，批量请求的总 Token 波动也大，先 32 跑通再逐步上调。
第二，把chunk_id和向量存进同一行 JSONL。别用两个文件靠顺序对齐，一旦有一次重跑顺序变了，向量和文本就错位了，而且这种错位不会报错，只会让你检索出莫名其妙的结果。
第三，向量只算一次，之后就缓存住。改问答 prompt、改 top-k、改过滤条件都不需要重新向量化。真正需要重算的只有两种情况：切块策略变了，或者换了 embedding 模型。
3. 检索问答层：让每条答案都能追回原话
向量化完成后，问答链路其实很短：把问题向量化，算余弦相似度，取 top-k，拼进上下文，让模型基于命中的片段作答并要求标注出处。
# ask.py import json import sys import numpy as np from openai import OpenAI import os BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") client = OpenAI( api_key=os.environ.get("TAOTOKEN_API_KEY", "YOUR_API_KEY"), base_url=f"{BASE_URL}/v1", ) EMBED_MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small") CHAT_MODEL = os.environ.get("CHAT_MODEL", "gpt-4o-mini") # 以控制台模型列表为准 def load(): rows = [json.loads(l) for l in open("chunks_embedded.jsonl", encoding="utf-8")] mat = np.array([r["vector"] for r in rows], dtype=np.float32) mat /= np.linalg.norm(mat, axis=1, keepdims=True) + 1e-9 return rows, mat def retrieve(question, rows, mat, top_k=6, speaker=None): qv = client.embeddings.create(model=EMBED_MODEL, input=[question]).data[0].embedding qv = np.array(qv, dtype=np.float32) qv /= np.linalg.norm(qv) + 1e-9 sims = mat @ qv idx = np.argsort(-sims) hits = [] for i in idx: if speaker and rows[i]["speaker"] != speaker: continue hits.append((rows[i], float(sims[i]))) if len(hits) >= top_k: break return hits SYSTEM = """你是研究情报库的检索助手。 只能依据下面给出的片段回答，不要补充片段之外的事实。 每条结论后面必须标注来源，格式为 [发言人#chunk_id]。 如果片段不足以回答，直接说明「现有片段不足以回答」。""" def answer(question, speaker=None): rows, mat = load() hits = retrieve(question, rows, mat, speaker=speaker) ctx = "\n\n".join(f"[{r['speaker']}#{r['chunk_id']}] {r['text']}" for r, _ in hits) resp = client.chat.completions.create( model=CHAT_MODEL, temperature=0.2, messages=[ {"role": "system", "content": SYSTEM}, {"role": "user", "content": f"片段：\n{ctx}\n\n问题：{question}"}, ], ) return resp.choices[0].message.content, hits if __name__ == "__main__": q = sys.argv[1] if len(sys.argv) > 1 else "递归自我改进的主要瓶颈是什么？" text, hits = answer(q) print(text) print("\n--- 命中片段 ---") for r, s in hits: print(f"{s:.3f} [{r['speaker']}#{r['chunk_id']}] {r['text'][:60]}...")跑起来大致是这样：
export CHAT_MODEL=<控制台里选定的对话模型> python ask.py "验证手段跟不上自我改进速度会带来什么问题？"输出会是一段带[John Schulman#c0031]这类标注的答案，后面再附上命中片段和相似度分数。分数这一步别省，它决定了你敢不敢信任这条答案：0.55 以上通常是直接相关，0.4 以下基本就是蹭词，宁可让它回「现有片段不足以回答」。
过滤也很有用。问「训练和算力相关的说法」时加speaker="Charlie O'Neill"，能一次性把另外两位的干扰排掉，实测比只调 top-k 有效得多。
4. 用 Claude Code 维护这个库：settings.json 怎么填
脚本能跑之后，剩下的活是持续维护——改切块正则、调 prompt、加新来源。这类改动我用 Claude Code 做，它可以直接读仓库里的文件、改完给你 diff，比在聊天窗口里贴代码块效率高。关键是把它的请求也指向 TaoToken。
Claude Code 走的是 Anthropic 协议，所以ANTHROPIC_BASE_URL直接填https://taotoken.net/api，不要加/v1。配置文件放在~/.claude/settings.json：
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "<在控制台模型列表中选择>", "ANTHROPIC_SMALL_FAST_MODEL": "<轻量任务用的小模型>" } }几个要点：
ANTHROPIC_AUTH_TOKEN放的是你的 Key，不要把真实 Key 提交进 Git 仓库，本地用环境变量注入或者放在.gitignore覆盖的本地文件里更稳。
ANTHROPIC_MODEL和ANTHROPIC_SMALL_FAST_MODEL都要填。前者负责改代码这类重活，后者负责文件摘要、路径补全这类轻活，两个分开配能明显压低日常消耗。
改完配置重启一次终端会话，然后直接问它「读一下 chunk_transcript.py，把发言人正则改成配置文件驱动」。它会给出具体改动，你 review 完再合。
如果你更习惯 Codex 风格的命令行工具，配置是另一套文件，不要把ANTHROPIC_*那组变量塞进 Codex，它读不懂，表现出来就是一直用默认端点。Codex 用~/.codex/config.toml：
model = "<在控制台模型列表中选择>" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api/v1" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"对应在 shell 里导出 Key：
export TAOTOKEN_API_KEY=YOUR_API_KEY注意两边的路径差异：Claude Code 是https://taotoken.net/api，Codex 的 OpenAI 兼容通道是https://taotoken.net/api/v1。这不是哪个写错了，是两套协议对版本段的处理方式不同，抄配置的时候一定要看清楚是哪一套。
5. CC Switch 三件套：多套 Key 和多供应商之间不打架
当你在同一个项目里同时用 Claude Code、Codex，还可能在不同机器上跑同一套脚本时，配置很快就会乱。我自己的做法是用 CC Switch 管理三样东西，我习惯叫它三件套：
第一件，配置档案（Profile）。每个档案存一份完整的端点配置：Claude Code 一份、Codex 一份、跑向量化脚本的 shell 环境一份。档案之间互不影响，切换就是换一套。
第二件，密钥绑定。Key 不写死在档案里，而是用引用方式关联。这样换 Key 只改一处，所有引用了它的档案同步生效，不用逐个文件去翻。
第三件，一键切换与回滚。切换前自动备份当前配置，切错了能回到上一个可用状态。这在你调试模型选择、需要来回对比的时候非常省事，也好排查问题。
三件套配好之后，日常操作就变成：改脚本用 Claude Code 的档案，做批量任务用命令行档案，两边共用同一个 Key 来源。参数不同、Key 同源，改一次全生效。
6. 排障清单：这几类报错基本能覆盖九成问题
401 / invalid api key。先确认 Key 有没有多余空格，尤其是从网页复制、粘贴到.env或 shell 的时候常带换行。再确认 Key 是不是已经生效，刚创建的 Key 建议先去控制台的模型对话页发一条最小请求验证，能返回再说别的。
404 / path not found。九成是 Base URL 拼错了。OpenAI 兼容 SDK 需要/v1，Anthropic 协议不要/v1。如果你把https://taotoken.net/api/v1填进了ANTHROPIC_BASE_URL，就会重复拼接路径，报错形态就是 404。
429 / rate limit。降BATCH_SIZE，加指数退避，批与批之间加sleep。上面embed_batch里的退避逻辑可以直接用，注意别把MAX_RETRY设成 1，那等于没重试。
上下文长度超限。这是导语里那个坑。它通常不是模型的问题，而是你的切块太大了。回头检查TARGET_LEN，把它降到 450 左右，同时确认拼接上下文时只拼 top-k 命中片段，不要把整个库塞进去。
检索结果莫名其妙。先看chunk_id和向量有没有错位，再看向量里有没有带发言人标签，最后才怀疑模型。实践里前两个原因占绝大多数。
回答里编造了片段外的事实。这是 prompt 的锅，不是检索的锅。把 system prompt 里「只能依据片段回答」和「标注来源」这两条加粗强调，同时把temperature压到 0.2 以下。
配置排查到这里如果还是卡住，可以去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_troubleshoot 对照控制台里给出的端点信息和模型列表，确认你填的路径和模型名是一致的，这一步能省掉大量来回试错。
7. 复现清单：从逐字稿到可检索结果
跑通一遍的完整顺序，按这个顺序走基本不会绕路：
- 准备
transcript.txt，确保每位发言人的标签格式统一，这是切块正确的前提。 - 配置环境变量：
TAOTOKEN_API_KEY、TAOTOKEN_BASE_URL=https://taotoken.net/api、EMBED_MODEL、CHAT_MODEL。 - 跑
chunk_transcript.py，检查输出的chunks=...数量是否合理。三位研究者一场长对谈，块数通常在几百量级，块数过少说明切块逻辑没生效。 - 抽查
chunks.jsonl里前几行，确认每块都带了speaker和turn_id，且没有在句子中间被截断。 - 跑
embed_chunks.py，观察进度输出。中途出现 429 是正常的，退避会自动处理。 - 跑
ask.py提三到五个你已知答案的问题，验证命中的片段确实包含答案，并且相似度分数落在合理区间。 - 把这三个脚本和配置文件一起纳入版本管理，Key 用环境变量隔离。
这套流程跑完之后，你得到的不是一份聊天记录，而是一个可以反复追问、每条结论都能追回原话的研究情报库。RSI 这类话题的价值恰恰在这里——三人立场不同，观点的出处和语境比结论本身更重要，而向量检索恰好能把「谁在什么语境下说了什么」保留下来。
8. 下一步：把这套链路接到你自己的项目上
如果你也想把自己手上的长对谈、会议记录、技术文档做成同款情报库，按下面的顺序接入最省时间：
先从模型对话开始验证连通性，发一条最小请求确认 Key 和端点都对：https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_chat
确认没问题后，如果你要长期跑编码和维护脚本，看下 Coding Plan 是否比按量更划算：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_plan
然后把正式要用的 Key 创建出来，把YOUR_API_KEY换掉：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_keys
如果你打算用 Claude Code 来维护这套脚本，配置细节和变量说明在文档里写得很清楚：https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=rsi_rag_ccdoc
Base URL 记住两个形态：OpenAI 兼容通道https://taotoken.net/api/v1，Anthropic 协议通道https://taotoken.net/api。填错了不会报鉴权错，只会报路径错——这也是我在这个项目里踩过的第一个坑。
