ainvoke与并发

FastAPI 一条请求一个 invoke 会阻塞 event loop——LLM/HTTP 工具应走 async 图 API。ainvoke / astream 与 async 节点函数配合，在单进程内并发多会话。

1. 定位

维度	内容
角色	非阻塞图执行
输入 → 输出	`await graph.ainvoke(...)` → 同 sync 语义
核心 API	`ainvoke`、`astream`、async 节点
依赖 LangChain	async ChatModel `ainvoke`

2. 图拓扑

节点表

节点名	职责	类型	写
`fetch`	模拟 IO	async	`data`
`parse`	解析	sync	`result`

边表

源	目标	类型
START	fetch	固定
fetch	parse	固定
parse	END	固定

3. invoke 生命周期（async）

await ainvoke:
  1. 调度 fetch（async 节点 → await asyncio.sleep / httpx）
  2. superstep 合并 data
  3. 调度 parse（sync 节点在线程池或直跑，依实现）
  4. END 返回

4. 原理

4.1 async 节点签名

1	async def fetch(state): ...

LangGraph 识别 coroutine 函数。

4.2 混用 sync/async

全 async 图最佳；sync 节点可能阻塞 loop，IO 节点应 async。

4.3 并发多会话

不同 thread_id 的多个 ainvoke 可 asyncio.gather。

5. 最小可运行示例

import asyncio
from typing import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    data: str
    result: str


async def fetch(_: State) -> dict:
    await asyncio.sleep(0.01)
    return {"data": "payload"}


def parse(state: State) -> dict:
    return {"result": state["data"].upper()}


builder = StateGraph(State)
builder.add_node("fetch", fetch)
builder.add_node("parse", parse)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "parse")
builder.add_edge("parse", END)

graph = builder.compile()


async def main():
    out = await graph.ainvoke({"data": "", "result": ""})
    print(out["result"])  # PAYLOAD
    events = []
    async for ev in graph.astream({"data": "", "result": ""}, stream_mode="updates"):
        events.append(ev)
    print(events)


asyncio.run(main())

6. 执行追踪

步骤	`data`	`result`
fetch 后	payload	“”
parse 后	payload	PAYLOAD

astream：依次收到 {"fetch":...}、{"parse":...}。

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
`ainvoke`	coroutine	异步跑完	FastAPI route	勿混 sync invoke
`astream`	async iter	流式	SSE 13 篇	await for
async 节点	async def	非阻塞 IO	httpx	IO 必 async
`asyncio.gather`	—	多 thread	批处理	每 call 独立 config
checkpointer	同 sync	并发写 DB	Postgres	注意连接池
线程池 sync 节点	内部	CPU 密集	少量	监控阻塞

7. 易踩坑

在 async route 里调 sync invoke：阻塞 event loop。
async 节点内调 sync HTTP 客户端：仍阻塞，换 httpx.AsyncClient。
gather 共用可变 config：thread_id 必须区分。

小结

IO 密集服务用 ainvoke/astream + async 节点。
语义与 sync 一致；checkpointer、stream_mode 参数相同。
多会话并发靠 asyncio.gather + 不同 thread_id。