ainvoke与并发

FastAPI 路由里若用同步 invoke 调 OpenAI，等待 HTTP 期间整个事件循环被占，并发上不去。Runnable 提供 ainvoke / astream / abatch，与 asyncio 集成；参数与同步版一致，仅须 await。

段末注释：asyncio = Python 异步 I/O 框架；ainvoke = Runnable 的异步单次调用入口。

1. 一句话定位

维度	内容
角色	能力层异步 IO：非阻塞调模型/检索
输入 → 输出	与 invoke 相同类型
典型调用入口	`await chain.ainvoke(...)`、`async for chunk in chain.astream(...)`
与 LangGraph	`await graph.ainvoke` 同理

2. 实现逻辑

1. async def handler(): result = await chain.ainvoke(input, config=...)
2. ChatModel._agenerate 走 async HTTP 客户端
3. astream：async for chunk in chain.astream(...) 增量输出
4. abatch：await chain.abatch([in1, in2], config={..., "max_concurrency": 5})
5. 勿在 async 路由里 chain.invoke()（阻塞）
6. 若库仅同步，用 asyncio.to_thread(chain.invoke, ...) 兜底

字段级变形：与 sync 相同；事件循环在 await 期间可处理其他请求。

3. 原理说明

3.1 默认 async 实现

ainvoke(input, config=None)（方法）
功能：异步单次。参数与 invoke 相同。部分 Runnable 默认 run_in_executor 调 sync；ChatModel partner 应提供真 async。
最小维度：同同步 invoke（messages len≥1 或 prompt dict 键齐全）。

1	await chain.ainvoke({"x": "浙江的省会"})

3.2 astream / astream_events

astream(input, config=None)（异步生成器）
功能：流式 chunk。最小：至少 yield 1 次；拼起来与 ainvoke 同义。

1	chunks = [c async for c in chain.astream({"x": "江苏的省会"})]

astream_events(input, version="v2", ...)（异步生成器）
功能：细粒度事件（on_chain_start、on_chat_model_stream 等），供 UI 展示中间步。
默认 version="v2"。最小消费：迭代到 event=="on_chain_end"。

1
2
3

async for ev in chain.astream_events({"x": "hi"}, version="v2"):
    if ev["event"] == "on_chat_model_stream":
        print(ev["data"]["chunk"].content, end="")

3.3 并发与 rate limit

abatch(inputs, config=None, *, return_exceptions=False)（方法）
inputs len≥1；空列表 → []。return_exceptions 默认 False。

max_concurrency（config 键）
类型 int / None，默认 None（实现自行定）。有意义的下限 ≥1；过大易 429。

1	await chain.abatch([{"x": "广东的省会"}, {"x": "四川的省会"}], config={"max_concurrency": 2})

3.4 FastAPI 模式

路由必须 await chain.ainvoke(...)，不要在 async def 里调同步 invoke。

1
2
3

@app.post("/chat")
async def chat(body: Input):
    return await chain.ainvoke(body.dict())

Input 最小：覆盖链的 input_variables。

4. 最小可运行示例

1	pip install -U langchain-openai

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(
    model="qwen3.5:9b",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
    temperature=0,
)
chain = (
    ChatPromptTemplate.from_template("只输出一个城市名。问题：{x}")
    | model
    | StrOutputParser()
)

async def main():
    print(await chain.ainvoke({"x": "浙江的省会"}))
    chunks = [c async for c in chain.astream({"x": "江苏的省会"})]
    print("".join(chunks))
    print(await chain.abatch(
        [{"x": "广东的省会"}, {"x": "四川的省会"}],
        config={"max_concurrency": 2},
    ))

asyncio.run(main())
# 预期形态：杭州；南京；['广州', '成都']（措辞随模型变）

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点	配置指导
`ainvoke`	async 方法	在事件循环里异步跑一次链	FastAPI 里用同步 invoke 会阻塞整个 loop	与 invoke 同参	异步路由必用
`astream`	async 生成器	异步产出 chunk，供 UI 打字机	消费者慢会造成反压	UI 打字机	注意 backpressure
`abatch`	async 方法	一批输入并发（受 max_concurrency）	无上限易瞬间 429	嵌入/批问	必须设并发上限
`max_concurrency`	config int / None	限制 abatch/并行同时飞行数	过大打满 quota；过小吞吐低	5～20	视厂商限额
partner async	HTTP 客户端实现	是否真异步而非线程里调 sync	假异步仍占 worker	aiohttp 等	查 langchain-openai 文档
`asyncio.to_thread`	兜底	把不得不同步的短 CPU 活丢到线程	不能替代 IO 的真 async	CPU 短任务	非 IO 才用