LangChain 接小鱼API 只需要在 ChatOpenAI 里补一个 base_url，其余写法与上游文档完全一致。下面这段是能直接跑起来的最小版本：
# pip install -U langchain langchain-openai
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="deepseek-v3.2-thinking",
api_key="YOUR_KEY",
base_url="https://xyuapi.top/v1",
temperature=0.7,
timeout=120,
max_retries=2,
)
msg = llm.invoke("用三句话说明什么是向量数据库")
print(msg.content)api_key 换成 xyuai.cc 后台生成的令牌，base_url 填 https://xyuapi.top/v1。注意 LangChain 里 base_url 是顶层参数，不是塞进 model_kwargs 里；塞错位置不会被报错，只会静默走默认地址，然后返回 401。
先用三条真实业务提示词各跑一次，确认输出格式和原来一致。很多人迁移完只跑了一句「你好」就宣布成功，结果上线后才发现长输出的模型名不一致、流式解析出错。三条真实数据的验证成本很低，能挡掉后面大部分返工。
LangChain 的包结构在这两年做过一次大拆分：核心逻辑进了 langchain-core，各家模型集成各自独立发包。现在装东西要分清三层，langchain、langchain-core、langchain-openai。只装 langchain 而不装 langchain-openai，导入 ChatOpenAI 时会直接报模块不存在。
pip install -U langchain langchain-openai langchain-community
pip install -U langchain-text-splitters # 文档切分单独一个包
pip show langchain langchain-openai | findstr Version版本上有个硬约束：langchain-openai 的主版本号要和 langchain-core 匹配，混装新旧版本会出现 ImportError: cannot import name 'BaseChatModel' 这类看起来莫名其妙的错误。升级时三个包一起升。
$env:OPENAI_API_KEY="YOUR_KEY"
$env:OPENAI_BASE_URL="https://xyuapi.top/v1"ChatOpenAI 会自动读取这两个环境变量，代码里就只用写 model 一个参数。放进环境变量而不是写死在源码里，能避免误提交到代码仓库后被爬虫扫走。
| 参数 | 该怎么填 | 常见错误 |
|---|---|---|
base_url | https://xyuapi.top/v1 | 漏写 /v1，或写成 /v1/ 结尾 |
api_key | 后台生成的令牌 | 复制时带换行或空格 |
model | 模型列表里的完整名字 | 用别家平台的简称 |
timeout | 120 秒起步 | 沿用默认值导致长回答超时 |
max_retries | 2 到 3 | 设成 0，遇到 429 直接失败 |
model 必须是字符串，写错名字会返回 404，LangChain 包装后会抛 NotFoundError。temperature 的可接受范围随模型而变，思考型模型多数只接受默认值，传了 temperature=1.5 之类的越界值会被上游拒绝。
思考型模型首字延迟高，三千字的长回答可能要等四十秒。timeout 设 30 秒的表现是偶发成功、偶发超时，很难定位。建议对话类 120 秒，长文类 300 秒。max_retries 只对 429 和 5xx 生效，超时不重试，所以业务层还要再包一层。
max_tokens 直接决定单次回复的上限。设得太小，长回答会在中途被切断，finish_reason 返回的是长度上限而不是正常结束；设得偏大没有坏处，模型不会因为上限高就多写。按模型实际支持的范围给一个宽松值即可，比如 8192。
from pydantic import BaseModel, Field
class Product(BaseModel):
name: str = Field(description="商品名称")
price: float = Field(description="价格，单位元")
tags: list[str] = Field(description="不超过 5 个标签")
structured = llm.with_structured_output(Product)
item = structured.invoke("从这段描述里抽取商品信息：小熊保温杯，售价 59.9 元，304 不锈钢，送礼首选")
print(item.name, item.price)with_structured_output 走的是函数调用通道，比让模型自己输出 JSON 再解析可靠得多。字段描述要写清楚，模型是按描述来填的。
不同模型对参数的宽容度不一样，报 400 时先做一件事：把模型换成 gemini-2.5-pro 再试一次，能通就说明参数本身没错，是模型侧的限制。这样可以把「代码问题」和「模型差异」快速分开。
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是电商客服助手，回答不超过 100 字，不承诺具体到货时间。"),
("human", "用户问题：{question}\n订单状态：{status}"),
])变量名用花括号占位，invoke 时传字典。变量名拼错会在调用时报 KeyError，比模型返回错误更容易排查。
from langchain_core.output_parsers import StrOutputParser
chain = prompt | llm | StrOutputParser()
answer = chain.invoke({"question": "什么时候发货", "status": "已付款待发货"})
print(answer)管道符本质是 Runnable 的 __or__ 重载，左边的输出喂给右边当输入。任何实现了 Runnable 接口的对象都能串进去，包括检索器、解析器、自定义函数。
在链路中间插一个 RunnableLambda 把中间值打印出来即可，或者调用 chain.invoke(..., config={"callbacks": [handler]}) 传一个回调处理器。常见做法是先单独跑 prompt.invoke({...}) 看渲染出来的消息，确认占位符替换没问题，再把整条链接上。
for chunk in llm.stream("写一段 300 字的商品卖点"):
print(chunk.content, end="", flush=True)stream 返回的是迭代器，直接进 for 就能逐字打印。链式结构上也可以 .stream()，LangChain 会自动沿管道向后传递流式片段，但要求链路里每一环都支持流式，中间插了不支持流式的处理器就会退化成一整块输出。
async for event in chain.astream_events({"question": "介绍一下会员权益"}, version="v2"):
if event["event"] == "on_chat_model_stream":
piece = event["data"]["chunk"].content
if piece:
print(piece, end="", flush=True)version="v2" 不能省，不写会直接抛异常。事件名里 on_chat_model_stream 是模型吐字的信号，想抓工具调用就用 on_tool_start 和 on_tool_end。
流式请求不会自动重连，连接中途断掉时迭代器直接结束，不会抛异常。稳妥做法是记录已收到的文本长度，一旦在正常结束标记出现之前就退出循环，就带着已生成内容重新请求一次，让模型续写，重试上限设两次。
可以，但要求链路里每一环都支持流式。中间插一个必须拿到完整输入才能工作的解析器，整条链就会退化成一整块输出，前端看到的效果是文字从头到尾一次性出现。真遇到这种情况，把那个环节挪到链路末端就行。
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=120,
separators=["\n\n", "\n", "。", "！", "？", " ", ""],
)
chunks = splitter.split_documents(docs)chunk_overlap 不要设成 0，段落被硬切时上下文会丢。中文场景记得把中文标点加进分隔符列表，否则会按空格乱切。
向量化有两条路：本地模型和接口。本地用 BAAI/bge-small-zh-v1.5 这类中文小模型，机器上就能跑，数据不出本机；接口方式则用平台的向量化型号，具体名字以后台模型列表为准，写错同样返回模型不存在。文档量大时本地跑更省，检索延迟也更可控。
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
store = FAISS.from_documents(chunks, embeddings) # embeddings 用上面任一方案
retriever = store.as_retriever(search_kwargs={"k": 4})
rag = (
{"context": retriever | (lambda ds: "\n\n".join(d.page_content for d in ds)),
"question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)提示词里要明确写一句「只根据提供的资料回答，资料里没有的信息直接说不知道」，否则模型会拿自己的记忆补全，看着像答对了，其实是编的。
from langchain_core.tools import tool
@tool
def query_price(model_name: str) -> str:
"""查询指定模型名的一次调用单价，单位元。"""
table = {"gemini-2.5-pro": "0.031", "deepseek-v3.2-thinking": "0.049"}
return table.get(model_name, "未收录该型号")函数的 docstring 就是给模型看的工具说明，写得含糊模型就不会正确调用。参数类型标注也是必需的，模型靠它生成参数。
llm_with_tools = llm.bind_tools([query_price])
msg = llm_with_tools.invoke("gemini-2.5-pro 一次多少钱？")
if msg.tool_calls:
for call in msg.tool_calls:
print("模型想调用：", call["name"], call["args"])bind_tools 只负责把工具定义发给模型，真正的执行和回填要自己写循环：把工具的返回结果包成 ToolMessage 追加进消息列表，再请求一次模型。LangChain 也提供 create_agent 这类封装好的执行器来自动完成这个循环，不同大版本的工厂函数名不一样，升级时留意一下导入路径。
一是循环没有出口，模型反复调同一个工具，必须设最大步数上限。二是工具抛异常直接打断整个流程，工具内部要把异常兜住并返回一句可读的失败说明，让模型有机会换路径。三是每轮循环都会产生一次计费调用，按次计费下这个成本增长是线性的，跑之前先估算步数。
| 现象 | 原因 | 处理 |
|---|---|---|
| 401 认证失败 | base_url 没生效，或令牌带空格 | 打印实际地址，strip() 令牌 |
| 404 模型不存在 | 模型名写了别家的简称 | 对照后台模型列表逐字核对 |
| 400 参数不支持 | 传了该模型不接受的参数 | 去掉温度、top_p 等可选参数 |
| 首字很慢后超时 | timeout 过短 | 提到 120 秒，长任务 300 秒 |
| 429 频率限制 | 并发过高 | 降到 3 到 5 并发，加指数退避 |
| 导入报错 | 三个包版本不匹配 | 三个包一起升到同代版本 |
把 base_url 塞进 model_kwargs={"base_url": ...} 是最隐蔽的一类错误，代码不报错，请求照样发出去，只是发到了别的地方。判断方法是打印 llm.openai_api_base 看实际值。
同一个提示词在 A 模型上正常、在 B 模型上报 400，原因往往在参数上而不是提示词。思考型模型常拒绝 temperature 和 top_p，把这两个参数去掉再试。
timeout 在 LangChain 里是整次请求的时限，包含排队、推理和输出三个阶段。长输出的任务要按「最慢情况」设，而不是按平均值设。超时后 max_retries 不会重试，需要业务层自己处理。
langchain 生态包多，最容易出现的是装了两个不同主版本的 langchain-core。pip check 能查出这类冲突，pip list | findstr langchain 能看清实际装了哪些包、什么版本。
| 模型名 | 单价 | 适合的环节 |
|---|---|---|
gemini-2.5-pro | 0.031 元/次 | 批量分类、初筛、简单问答 |
deepseek-v3.2-thinking | 0.049 元/次 | 中文生成、结构化抽取 |
claude-sonnet-4-5-thinking | 0.09 元/次 | 长文写作、多步链式任务 |
gpt-5.5 | 0.2 元/次 | 难题推理、Agent 主控 |
全部按调用次数计费，与输入输出长度无关。这一点在 RAG 场景里特别重要：把整篇长文档塞进上下文，和只塞四段检索结果，价格是一样的。
一条链里不必从头到尾用同一个模型。检索改写、意图分类这类环节用 gemini-2.5-pro 就够，生成环节用 deepseek-v3.2-thinking 或 claude-sonnet-4-5-thinking，只有真正需要多步推理的 Agent 主控才上 gpt-5.5。这样拆完之后，同样的链成本能降下来一截，质量反而更稳，因为每个环节用的都是它擅长的那一档。
把 LangChain 接到小鱼API，实质上就是在 ChatOpenAI 里多填一个 base_url，剩下的提示词模板、链、检索器、工具调用全都按原有方式写。真正需要留神的是三类不报错的错误：参数放错位置、超时设得太短、依赖包版本不匹配。这三类各自都有一个几分钟就能做完的确认动作，提前做掉比线上排查省事得多。