管理:從無狀態(tài) API 到有狀態(tài)會話的 TaoToken 實(shí)踐)
1. 無狀態(tài) API 為什么總在第三輪對話開始“失憶”如果你用最樸素的方式調(diào)過大模型 Chat API大概率經(jīng)歷過這個場景第一輪問“幫我寫個 Python 讀取 CSV 的函數(shù)”模型答得挺好第二輪說“改成支持分塊讀取”它卻反問你“什么 CSV要讀什么”——不是模型笨而是你根本沒把上一輪的消息帶過去。大模型的 Chat Completions 接口本質(zhì)上是無狀態(tài)的。服務(wù)端不記得你上一秒說過什么每次 HTTP 請求都是一個獨(dú)立的宇宙。你想讓它“記得”唯一的辦法就是在這一次請求的messages數(shù)組里把之前所有的 user / assistant 消息按順序重新塞一遍。這就是 LLM 多輪對話狀態(tài)管理最底層的約束狀態(tài)不在服務(wù)端而在你的客戶端代碼里。這件事聽起來簡單真做起來會撞上三堵墻。第一堵墻是上下文窗口有限。主流模型的上下文從 8K 到 128K token 不等看著挺大但一輪對話動輒幾百上千 token二十輪下來很容易頂?shù)教旎ò?。一旦超出要么請求直接?bào)錯要么模型開始“遺忘”最早的內(nèi)容。第二堵墻是Token 成本隨輪次線性增長。假設(shè)每輪問答平均 500 token第 1 輪你發(fā) 500第 10 輪你要發(fā) 5000第 20 輪你要發(fā) 10000。用戶聊得越久你越燒錢而且燒的錢大部分花在重復(fù)發(fā)送歷史消息上。第三堵墻是狀態(tài)漂移。就算你老老實(shí)實(shí)把歷史全帶上模型也可能在長上下文里抓錯重點(diǎn)。用戶第三輪說的“就按剛才那個格式”到第十輪模型已經(jīng)分不清“剛才”指的是哪一版格式了。信息沒丟但語義焦點(diǎn)丟了。所以 LLM 多輪對話狀態(tài)管理要解決的核心問題不是“怎么存歷史”而是在有限的上下文窗口里保留對當(dāng)前這輪對話最有價值的信息同時把 Token 消耗壓住。圍繞這個目標(biāo)業(yè)界通常拆成四個機(jī)制來做消息壓縮、摘要替換、關(guān)鍵信息提取、會話持久化。下面我會用 TaoToken 作為統(tǒng)一的 API 通道把這套東西從零搭一遍你能直接復(fù)制去跑。先說清楚 TaoToken 在這里扮演什么角色。它是一個統(tǒng)一的大模型 API 網(wǎng)關(guān)你用一個 Key 就能訪問多家模型Base URL 固定省得每換一個模型就改一遍 SDK 配置。對多輪對話場景來說這點(diǎn)很關(guān)鍵——因?yàn)闋顟B(tài)管理邏輯里經(jīng)常要調(diào)用“摘要模型”和“主對話模型”如果兩個模型來自不同廠商、走不同通道你的代碼里會塞滿各種 base_url 判斷。統(tǒng)一通道之后切換模型只是改一個 model 字符串的事。適合讀這篇的人正在做客服機(jī)器人、銷售助手、代碼助手這類需要連續(xù)多輪交互的后端同學(xué)或者你已經(jīng)用單輪 API 跑通了 demo但一上多輪就發(fā)現(xiàn)上下文亂套、成本失控。接下來從環(huán)境準(zhǔn)備到可復(fù)制的會話管理器再到真實(shí)報(bào)錯排查一步步來。2. TaoToken 前置準(zhǔn)備統(tǒng)一 Key 與 API 通道配置在寫會話管理器之前得先把調(diào)用通道打通。這一步不做后面所有代碼都跑不起來。我用 TaoToken 的原因是它把多模型的接入收斂成一個 Base URL 加一個 Key配置一次到處能用特別適合我們這種要在摘要模型和主模型之間來回切的場景。2.1 獲取 API Key 與確認(rèn) Base URL先到控制臺創(chuàng)建一個 API Key。地址是https://taotoken.net/console登錄后在 API Keys 頁面點(diǎn)創(chuàng)建復(fù)制出來的那串sk-開頭的字符串就是你的憑證。注意它只完整顯示一次先存到環(huán)境變量里別硬編碼進(jìn)代碼。Base URL 統(tǒng)一用https://taotoken.net/api注意這里不帶任何查詢參數(shù)。很多人第一次配會把官網(wǎng)地址和 API 地址搞混官網(wǎng)是https://taotoken.net/但 SDK 里填的 base_url 必須是帶/api的那個否則請求會打到網(wǎng)頁服務(wù)器上返回 HTML你的 JSON 解析直接崩。我習(xí)慣用環(huán)境變量管理Linux / macOS 下這樣寫export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api2.2 用 OpenAI SDK 驗(yàn)證通道TaoToken 的接口兼容 OpenAI 的 Chat Completions 協(xié)議所以直接用 openai 官方 SDK 就行不用裝額外的包。先裝依賴pip install openai然后寫一個最小驗(yàn)證腳本確認(rèn) Key 和 Base URL 都對import os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 只回復(fù)兩個字通了}], ) print(resp.choices[0].message.content)跑出來打印“通了”說明通道沒問題。如果報(bào) 401八成是 Key 復(fù)制時帶了空格或者引號如果報(bào)連接錯誤檢查 base_url 是不是漏了/api。2.3 為什么多輪對話特別需要統(tǒng)一通道這里多說一句設(shè)計(jì)上的考慮。多輪對話狀態(tài)管理里我們至少要用到兩類模型調(diào)用一類是主對話模型負(fù)責(zé)和用戶交互另一類是摘要模型負(fù)責(zé)把早期歷史壓縮成短文本。這兩類調(diào)用對模型的要求不一樣——主模型要強(qiáng)摘要模型要便宜快。如果它們走不同的廠商通道你的代碼里就得維護(hù)兩套 client、兩套鑒權(quán)、兩套錯誤處理。用 TaoToken 之后兩個調(diào)用共用一個 client只是model參數(shù)不同# 主對話 main_resp client.chat.completions.create(modelgpt-4o, messagescontext) # 摘要壓縮 sum_resp client.chat.completions.create(modelgpt-4o-mini, messagessummary_prompt)切換模型、對比效果、做 A/B 測試都只是改一個字符串。對狀態(tài)管理這種需要反復(fù)調(diào)優(yōu)壓縮策略的場景省下來的配置時間相當(dāng)可觀。通道打通后下面進(jìn)入正題會話管理器怎么寫。3. 可復(fù)制的會話狀態(tài)管理器配置與實(shí)現(xiàn)這一節(jié)是全文的技術(shù)核心。我會給出一個能直接跑的 Python 版本會話管理器包含會話存儲、上下文窗口管理、摘要壓縮、實(shí)體提取四個部分。代碼結(jié)構(gòu)參考了生產(chǎn)環(huán)境的做法但做了簡化你可以按需替換存儲后端。3.1 會話狀態(tài)的數(shù)據(jù)結(jié)構(gòu)先定義兩個基礎(chǔ)結(jié)構(gòu)單條消息和整個會話狀態(tài)。消息除了 role 和 content我還帶了 token 估算值和時間戳方便后面做窗口裁剪。from dataclasses import dataclass, field from typing import List, Dict import time dataclass class ChatMessage: role: str # system / user / assistant content: str timestamp: float field(default_factorytime.time) token_count: int 0 dataclass class ConversationState: session_id: str history: List[ChatMessage] field(default_factorylist) entities: Dict[str, str] field(default_factorydict) total_tokens: int 0 def add_message(self, msg: ChatMessage): self.history.append(msg) self.total_tokens msg.token_count def history_tokens(self) - int: return sum(m.token_count for m in self.history)token 估算這里先用一個粗糙的公式中文字符約 1 字 1 token英文約 4 字符 1 token。生產(chǎn)環(huán)境建議換成 tiktoken 之類的精確分詞器但做邏輯驗(yàn)證夠用了。def estimate_tokens(text: str) - int: # 粗略估算中文按字符數(shù)英文按 4 字符 1 token chinese sum(1 for c in text if \u4e00 c \u9fff) other len(text) - chinese return chinese max(1, other // 4)3.2 上下文窗口管理的三段式策略這是整個管理器最關(guān)鍵的部分。當(dāng)歷史 token 超出預(yù)算時不能簡單粗暴地砍掉最早的幾條那樣會丟失關(guān)鍵約束。我采用的是“系統(tǒng)消息 摘要 最近消息 實(shí)體信息”的四段式組裝其中摘要和實(shí)體信息負(fù)責(zé)保留遠(yuǎn)期記憶最近消息負(fù)責(zé)保留即時上下文。class ContextWindowManager: def __init__(self, max_tokens: int 8000): self.max_tokens max_tokens def build_context(self, state: ConversationState, client) - List[dict]: history state.history if state.history_tokens() self.max_tokens: return [{role: m.role, content: m.content} for m in history] # 預(yù)算分配最近消息 60%摘要 30%實(shí)體 10% budget self.max_tokens system_msgs [m for m in history if m.role system] for m in system_msgs: budget - m.token_count recent_budget int(budget * 0.6) summary_budget int(budget * 0.3) recent self._take_recent(history, recent_budget) early history[: len(history) - len(recent)] context [] context.extend({role: m.role, content: m.content} for m in system_msgs) if early: summary self._summarize(early, summary_budget, client) context.append({role: system, content: f[對話摘要] {summary}}) if state.entities: ent_text 已知信息: ; .join(f{k}{v} for k, v in state.entities.items()) context.append({role: system, content: ent_text}) context.extend({role: m.role, content: m.content} for m in recent) return context def _take_recent(self, history, budget): recent, used [], 0 for m in reversed(history): if used m.token_count budget: break recent.insert(0, m) used m.token_count return recent def _summarize(self, messages, max_tokens, client) - str: text \n.join(f{m.role}: {m.content} for m in messages) prompt ( f把下面的對話壓縮成不超過 {max_tokens} token 的摘要 f保留關(guān)鍵決策、約束條件和用戶偏好去掉寒暄\n{text} ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], ) return resp.choices[0].message.content注意_summarize里用的是便宜的小模型因?yàn)檎蝿?wù)對模型能力要求不高用大模型純屬浪費(fèi)。這也是統(tǒng)一通道的好處——同一個 client 換個 model 名就行。3.3 實(shí)體提取讓關(guān)鍵信息不隨歷史被壓縮摘要再聰明也會有信息損失所以對結(jié)構(gòu)化的關(guān)鍵信息訂單號、日期、用戶偏好我單獨(dú)抽出來存進(jìn)entities每輪都注入上下文。這樣即使原始消息被摘要替換這些信息依然在。import re def extract_entities(state: ConversationState, message: str): patterns { order_id: r訂單號[:]\s*(\w), date: r(\d{4}[-/]\d{2}[-/]\d{2}), email: r([\w.-][\w-]\.[\w.]), } for key, pat in patterns.items(): m re.search(pat, message) if m: state.entities[key] m.group(1)規(guī)則提取覆蓋高頻模式夠用且零延遲。等你有精力了再上 NER 模型處理“上周三那個單子”這種口語表達(dá)。3.4 會話持久化配置會話狀態(tài)不能只放內(nèi)存進(jìn)程一重啟就全沒了。我用 Redis 存熱會話設(shè)置 1 小時 TTL冷數(shù)據(jù)異步落庫。配置片段如下import json import redis class SessionStore: def __init__(self, hostlocalhost, port6379, ttl3600): self.r redis.Redis(hosthost, portport, decode_responsesTrue) self.ttl ttl def save(self, state: ConversationState): key fconv:{state.session_id} data { history: [m.__dict__ for m in state.history], entities: state.entities, total_tokens: state.total_tokens, } self.r.setex(key, self.ttl, json.dumps(data, ensure_asciiFalse)) def load(self, session_id: str): raw self.r.get(fconv:{session_id}) if not raw: return None data json.loads(raw) state ConversationState(session_idsession_id) state.entities data[entities] state.total_tokens data[total_tokens] for m in data[history]: state.history.append(ChatMessage(**m)) return state如果你本地沒裝 Redis可以先用一個內(nèi)存字典頂上接口保持一致后面換真存儲不用改調(diào)用方代碼。3.5 組裝完整管理器把上面幾塊拼起來就是對外暴露的process_messageclass ConversationManager: def __init__(self, client, store, max_tokens8000): self.client client self.store store self.window ContextWindowManager(max_tokens) def process_message(self, session_id: str, user_input: str) - str: state self.store.load(session_id) or ConversationState(session_idsession_id) user_msg ChatMessage(user, user_input, token_countestimate_tokens(user_input)) state.add_message(user_msg) extract_entities(state, user_input) context self.window.build_context(state, self.client) resp self.client.chat.completions.create( modelgpt-4o, messagescontext, ) reply resp.choices[0].message.content reply_msg ChatMessage(assistant, reply, token_countestimate_tokens(reply)) state.add_message(reply_msg) self.store.save(state) return reply到這里一個具備上下文壓縮、實(shí)體保留、持久化能力的會話管理器就成型了。下一節(jié)驗(yàn)證它到底記不記得住。4. 多輪對話驗(yàn)證從請求到上下文一致性檢查代碼寫完不驗(yàn)證等于沒寫。這一節(jié)我用一個跨多輪的測試腳本檢查三件事短期記憶是否保持、遠(yuǎn)期信息是否在壓縮后仍可召回、Token 消耗是否被壓住。4.1 構(gòu)造一個會“埋信息”的測試對話測試思路是第一輪埋一個關(guān)鍵信息比如訂單號中間插入若干輪無關(guān)閑聊把歷史撐長最后再問那個訂單號。如果管理器工作正常模型應(yīng)該能答出來。manager ConversationManager(client, SessionStore()) sid test-session-001 # 第 1 輪埋訂單號 print(manager.process_message(sid, 我的訂單號是 A12345幫我查下狀態(tài))) # 第 2-8 輪無關(guān)閑聊撐大歷史 for i in range(2, 9): manager.process_message(sid, f隨便聊點(diǎn)別的第 {i} 個話題今天天氣不錯) # 第 9 輪召回測試 print(manager.process_message(sid, 我剛才說的訂單號是多少))4.2 觀察成功結(jié)果正常輸出應(yīng)該類似第 1 輪回復(fù)好的訂單 A12345 的狀態(tài)是... ... 第 9 輪回復(fù)您剛才提到的訂單號是 A12345。關(guān)鍵看第 9 輪。如果模型答出 A12345說明實(shí)體提取生效了——即使中間 7 輪閑聊把原始消息擠進(jìn)了摘要區(qū)entities里的訂單號依然被注入到了上下文。如果答不出來說明實(shí)體提取的正則沒匹配上或者注入邏輯沒生效。4.3 檢查 Token 消耗曲線再驗(yàn)證一下壓縮是否真的省了錢。在process_message里加一行日志打印每輪實(shí)際發(fā)送的 token 數(shù)context self.window.build_context(state, self.client) sent_tokens sum(estimate_tokens(m[content]) for m in context) print(f[session{session_id}] 本輪發(fā)送 {sent_tokens} tokens, 歷史共 {state.history_tokens()} tokens)跑完 9 輪你會看到前幾輪發(fā)送量隨歷史增長但一旦歷史超過max_tokens發(fā)送量就被壓在一個穩(wěn)定值附近不再上漲。這就是三段式窗口管理的效果——?dú)v史繼續(xù)變長但發(fā)給模型的上下文被摘要和裁剪控制住了。4.4 用統(tǒng)一通道做模型對比因?yàn)樽叩氖?TaoToken 統(tǒng)一通道你可以把主模型從gpt-4o換成別的跑同一套測試腳本對比不同模型在長上下文里的召回能力。改一行就行resp self.client.chat.completions.create(modelclaude-3-5-sonnet, messagescontext)這種對比在排查“到底是狀態(tài)管理的問題還是模型能力的問題”時特別有用。如果換個模型召回就正常了那說明你的壓縮策略沒問題是原模型長上下文能力弱。5. 常見報(bào)錯排查401、local proxy failed 與 choices 解析失敗多輪對話場景的報(bào)錯比單輪更隱蔽因?yàn)殄e誤可能出在摘要調(diào)用、主調(diào)用、存儲三個環(huán)節(jié)中的任意一個。下面是我實(shí)際踩過的幾類按報(bào)錯原文對照排查。5.1 401 Unauthorized報(bào)錯原文通常是openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}排查順序第一確認(rèn)環(huán)境變量TAOTOKEN_API_KEY真的被讀到了在腳本里print(os.environ.get(TAOTOKEN_API_KEY)[:8])看前幾位第二檢查 Key 有沒有多余空格或換行從控制臺復(fù)制時容易帶上第三確認(rèn) Key 沒有過期或被刪除。多輪場景里如果摘要調(diào)用和主調(diào)用用了不同的 client 實(shí)例可能出現(xiàn)一個配對了、一個沒配對的情況統(tǒng)一用一個 client 能避免這類問題。5.2 local proxy failed / connection error報(bào)錯原文類似openai.APIConnectionError: Connection error.或者帶local proxy failed字樣。這類基本是網(wǎng)絡(luò)層問題不是 Key 的問題。排查確認(rèn)base_url是https://taotoken.net/api別寫成官網(wǎng)地址確認(rèn)本機(jī)沒有殘留的代理環(huán)境變量干擾echo $HTTP_PROXY看一下如果有就臨時 unset 掉再試。多輪場景里如果摘要調(diào)用超時整個process_message會拋異常建議給摘要調(diào)用單獨(dú)包一層 try失敗時降級為直接截?cái)喽皇亲屨唽υ挶赖簟?.3 reading choices 解析失敗報(bào)錯原文TypeError: NoneType object is not subscriptable或者KeyError: choices。這通常發(fā)生在你直接對響應(yīng)做resp[choices]而不是resp.choices或者響應(yīng)體根本不是預(yù)期的 JSON。最常見的原因是 base_url 配錯請求打到了網(wǎng)頁服務(wù)器返回的是 HTMLSDK 解析失敗。另一個原因是模型名寫錯了某些通道對未知模型會返回非標(biāo)準(zhǔn)錯誤體。檢查model字符串拼寫以及 base_url 是否帶/api。5.4 上下文超限報(bào)錯報(bào)錯原文This models maximum context length is 8192 tokens, however you requested 9500 tokens這說明你的窗口管理沒生效或者max_tokens設(shè)得比模型實(shí)際窗口還大。檢查兩點(diǎn)ContextWindowManager的max_tokens要留出回復(fù)的余量比如模型窗口 8192你設(shè) 6000 比較穩(wěn)確認(rèn)build_context真的被調(diào)用了而不是某條分支直接返回了完整歷史。5.5 會話狀態(tài)丟失現(xiàn)象是重啟服務(wù)后之前聊的內(nèi)容全沒了。這基本是存儲層的問題確認(rèn) Redis 真的在跑redis-cli ping返回 PONG確認(rèn)save在每輪結(jié)束后被調(diào)用確認(rèn) TTL 沒設(shè)得太短。如果用的是內(nèi)存字典版本那重啟丟失是預(yù)期行為換 Redis 即可。5.6 三件套配置速查如果你用的是 Cline、CC Switch 這類工具接入配置項(xiàng)永遠(yuǎn)是三件套缺一不可配置項(xiàng)值Base URLhttps://taotoken.net/apiAPI Key控制臺創(chuàng)建的sk-開頭字符串Model ID如gpt-4o、claude-3-5-sonnet等任何“連不上”的問題先回頭核對這三項(xiàng)八成能定位。6. 把狀態(tài)管理接進(jìn)你的真實(shí)項(xiàng)目走到這里你已經(jīng)有了一個能跑的多輪對話狀態(tài)管理器。最后說幾個把它接進(jìn)真實(shí)項(xiàng)目時的實(shí)用建議都是我在實(shí)際項(xiàng)目里踩出來的。第一摘要策略要按業(yè)務(wù)調(diào)。客服場景里“約束條件”最重要摘要 prompt 里要強(qiáng)調(diào)保留閑聊場景里“最近幾輪”最重要可以把最近消息的預(yù)算從 60% 提到 80%。別一套參數(shù)用到底。第二實(shí)體提取先規(guī)則后模型。正則能覆蓋 80% 的高頻模式剩下 20% 的口語表達(dá)再考慮上 NER。一上來就上模型延遲和成本都不劃算。第三給摘要調(diào)用加降級。摘要模型偶爾會超時或返回空這時候不要讓整輪對話失敗直接退化成“截?cái)嘧钤绲南ⅰ币材苡?。健壯性比完美壓縮重要。第四監(jiān)控發(fā)送 token 數(shù)。在process_message里埋點(diǎn)把每輪實(shí)際發(fā)送的 token 數(shù)上報(bào)。一旦發(fā)現(xiàn)某類會話的發(fā)送量持續(xù)上漲說明壓縮策略對這類對話失效了需要針對性調(diào)整。如果你還沒配好通道先去https://taotoken.net/api-keys拿 Key接入細(xì)節(jié)看https://taotoken.net/doc。想先直觀感受一下多輪對話的效果可以直接在https://taotoken.net/models里對話測試把同一段對話復(fù)制進(jìn)去觀察模型在不同上下文長度下的表現(xiàn)差異。長期要做編碼助手或 Agent 這類高頻多輪場景的https://taotoken.net/coding-plan里的方案對控制成本更有幫助。狀態(tài)管理這件事代碼寫對只是第一步參數(shù)調(diào)對才是長期功夫。