因豆包2.0虽兼容OpenAI协议，但需X-Signature签名、X-Timestamp时间戳等自定义请求头，且认证机制与ChatOpenAI默认逻辑不兼容，直接传base_url和api_key会返回401或400错误。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
为什么不能直接用
接
豆包
？
因为豆包
大模型
（尤其是 2.0 版本）虽然兼容 OpenAI 协议，但实际行为存在关键差异：
默认走
路径、依赖
字段认证、且不处理豆包必需的
签名头。直接传
和
会返回
或
，错误信息里常带
。
必须绕过
的默认签名逻辑，自己构造带时间戳和 HMAC-SHA256 签名的请求头。官方 SDK（
）虽内置该逻辑，但它与 LangChain 的
接口不兼容——没法直接塞进
或
流程里。
怎么写一个真正能用的
？
核心是继承
，重写
方法，手动发 HTTP 请求并解析流式响应。别碰
的同步封装，豆包生产环境必须用异步 + 流式，否则超时或丢 chunk。
必须用
或
，同步 client 在高并发下会卡死连接池
必须是秒级整数，不是毫秒；
要对
拼接后做 HMAC-SHA256（注意不是只对 payload）
流式响应是
，每行以
开头，需逐行解析 JSON，跳过空行和
行
别信文档里说的 “自动重试”，豆包 v2 签名含时间戳，重试必须重新生成 header，否则 401
示例关键片段：
立即进入
“
豆包AI人工智官网入口
”；
立即学习
“
豆包AI人工智能在线问答入口
”；
token 计算不准会导致什么？
豆包按实际消耗 token 计费，但它的 tokenizer 和 OpenAI 不同：中文字符平均占 1.8–2.2 token（OpenAI 是 1.3–1.5），且系统消息、函数调用描述也会被计费。LangChain 默认用
算
，结果比实际少 25%–35%，轻则预算超支，重则触发 QPS 限流（豆包按 token/秒 限流，不是请求数）。
必须替换为豆包官方 tokenizer 或近似实现：
优先用豆包提供的
Python 包（
），它能精确匹配服务端逻辑
若不可用，用
+ 字符长度加权估算：中文字符 × 2 + 英文单词 × 1.2 + 符号 × 1，再加 10% buffer
所有 prompt 构造前先调
，超阈值（如 8k）就截断或触发 RAG 分块，别等 API 返回
多轮对话状态怎么不串？
豆包本身不维护会话 state，
列表全靠你传。LangChain 的
在多线程/异步环境下共享 memory 实例，会导致 A 用户的 history 被 B 用户读到。这不是 LangChain bug，是误用。
正确做法只有两个：
每次请求都新建
实例（轻量，无连接池开销）
把 conversation history 存在外部（Redis / 数据库），用
隔离，
参数只传当前轮 + 最近 3 轮历史（避免 token 溢出）
别试图用
做内存级会话管理——它底层还是共享对象，压测时错误率飙升到 12% 以上。
ChatOpenAIChatOpenAI/v1/chat/completionsapi_keyX-Signaturebase_urlapi_key401 Unauthorized400 Bad Request"missing required header: X-Signature"ChatOpenAIdoubao-sdkBaseChatModelLLMChainRunnableDoubaoChatModelBaseChatModel_generateinvokeaiohttphttpx.AsyncClientX-TimestampX-Signaturetimestamp + api_key + secrettext/event-streamdata:event:class DoubaoChatModel(BaseChatModel):
api_key: str
secret: str
base_url: str = "https://open.bigmodel.cn/api/llm/v2"
async def _generate(
self, messages: List[BaseMessage], **kwargs: Any
) -> ChatResult:
payload = {"messages": [m.dict() for m in messages], "stream": True}
timestamp = str(int(time.time()))
signature = hmac.new(
self.secret.encode(),
f"{timestamp}{self.api_key}".encode(),
hashlib.sha256
).hexdigest()
headers = {
"X-API-Key": self.api_key,
"X-Timestamp": timestamp,
"X-Signature": signature,
"Content-Type": "application/json",
}
async with httpx.AsyncClient() as client:
async with client.stream("POST", self.base_url, json=payload, headers=headers) as resp:
content = ""
async for line in resp.aiter_lines():
if line.startswith("data:") and line.strip() != "data:":
try:
chunk = json.loads(line[5:])
if "content" in chunk.get("choices", [{}])[0].get("delta", {}):
content += chunk["choices"][0]["delta"]["content"]
except json.JSONDecodeError:
continue
return ChatResult(generations=[ChatGeneration(message=AIMessage(content=content))])
tiktokengpt-3.5-turbotokenizerpip install doubao-tokenizerjiebacount_tokens()413 Payload Too LargemessagesConversationBufferMemoryDoubaoChatModelsession_idmessagesRunnableWithMessageHistory