如何在 FastAPI 中设计可扩展的 LangChain agent 架构？

建议每个 worker 复用 agent 定义，将会话状态放在外部存储中，同一会话内的请求串行执行。请求只携带本次输入和经过授权校验的会话标识，用户历史不要保存在全局 agent 的可变属性中。

异步调用优先使用 await agent.ainvoke(...)，同时检查模型、工具和存储的完整调用链。线程池用于执行同步阻塞操作，不能直接包装 arun() 这样的协程函数。

1. agent 可以复用，但要检查它持有哪些状态

通常不必为每个请求重新构建完整的 agent。各类对象可以按以下生命周期管理：

即使没有全局 memory，自定义工具把当前用户写入 self.user_id，或多个请求共享不能并发使用的数据库 session，也可能导致数据混用。

不要用下面的方式切换用户：

agent.memory = memory_for_current_user
result = await agent.ainvoke(...)

两个请求可能交错执行，后一次赋值会影响前一个请求。

复用 agent 时，会话信息应从本次调用中传入，依赖组件也必须满足并发要求。LangChain 的版本和集成组件各不相同，不能认定所有 agent 实例都天然支持并发调用。

2. 会话标识必须同时满足隔离和授权要求

建议服务端根据认证结果和会话记录生成内部标识，例如：

tenant_id + user_id + session_id

实际编码时，应使用无歧义的序列化方式，或由服务端生成唯一标识，避免简单拼接造成键冲突。

这三个标识的用途不同：

• user_id 标识用户，一个用户可以拥有多个会话。
• session_id 标识会话，由服务端校验其归属。
• request_id 标识一次请求，可用于追踪和幂等处理。

客户端传入 session_id 后，后端必须检查当前用户是否有权访问该会话。即使使用难以猜测的 UUID，也不能省略授权校验。

ConversationBufferMemory 本身不能让多个进程共享状态。每个 worker 即使都维护了一个 dict[session_id, memory]，这些字典仍然相互独立。跨 worker 的会话状态应放在 Redis、PostgreSQL 等各 worker 都能访问的后端中。

3. 新架构可以使用 create_agent 和持久化 checkpointer

initialize_agent 是旧接口，官方参考文档已将其标记为弃用，并指向 create_agent。迁移时还需要调整输入输出格式，不能只替换函数名。参考：initialize_agent

使用 create_agent 时，可以通过 checkpointer 保存会话状态，用 configurable.thread_id 选择会话。每轮调用只传入新增消息，由 checkpointer 恢复此前的状态，通常不需要再次提交全部历史。参考：短期记忆

下面的 FastAPI 示例适用于 Python 3.11+，以及提供这些接口的 LangChain/LangGraph 版本。项目中应锁定依赖版本。

app_services 是项目自己的模块，其中：

• authorize_session(request, session_id) 校验会话归属，返回内部会话键。
• session_guard(session_key) 提供跨 worker 的会话互斥，冲突时可以返回 HTTP 409，具体要求见下一节。
• tools 是已检查并发行为的工具列表。

import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Request
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from pydantic import BaseModel, Field

from app_services import authorize_session, session_guard, tools


class ChatRequest(BaseModel):
    session_id: str = Field(min_length=1)
    query: str = Field(min_length=1)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncConnectionPool(
        conninfo=os.environ["DATABASE_URL"],
        min_size=1,
        max_size=10,
        open=False,
        kwargs={
            "autocommit": True,
            "prepare_threshold": 0,
            "row_factory": dict_row,
        },
    ) as pool:
        await pool.wait()

        checkpointer = AsyncPostgresSaver(pool)
        # 首次使用前，在独立的部署初始化步骤执行：
        # await checkpointer.setup()

        app.state.agent = create_agent(
            model=ChatOpenAI(
                model=os.environ["OPENAI_MODEL"],
                temperature=0,
            ),
            tools=tools,
            checkpointer=checkpointer,
        )
        app.state.agent_slots = asyncio.Semaphore(16)
        yield


app = FastAPI(lifespan=lifespan)


@app.post("/chat")
async def chat(body: ChatRequest, request: Request):
    session_key = await authorize_session(
        request, body.session_id
    )

    try:
        async with asyncio.timeout(90):
            async with request.app.state.agent_slots:
                async with session_guard(session_key):
                    result = await request.app.state.agent.ainvoke(
                        {
                            "messages": [
                                {
                                    "role": "user",
                                    "content": body.query,
                                }
                            ]
                        },
                        config={
                            "configurable": {
                                "thread_id": session_key,
                            }
                        },
                    )
    except TimeoutError as exc:
        raise HTTPException(
            status_code=504,
            detail="Agent execution timed out",
        ) from exc

    return {"response": result["messages"][-1].content}

FastAPI 的 lifespan 适合管理跨请求复用、在应用关闭时释放的资源。每个 worker 都会创建自己的资源实例。参考：FastAPI 生命周期

AsyncPostgresSaver 支持异步 PostgreSQL 持久化，首次使用前需要执行 setup()。连接池应按所安装版本的要求配置。参考：AsyncPostgresSaver 实现

示例中的连接数、并发数和超时时间仅用于演示。生产环境还需要统一管理模型客户端的连接、超时和重试策略，并在应用退出时释放相关资源。

4. 共享存储还需要配合同一会话的并发控制

Redis 或 PostgreSQL 可以让请求读取共享状态，但不会自动安排两条消息的执行顺序。

例如，同一会话的请求 A、B 同时读取历史 H，随后各自生成回答。即使底层追加操作是原子的，B 生成回答时也可能完全没有看到 A。checkpointer 也不能直接当作应用层的会话执行锁。

聊天接口通常允许不同会话并发，同一会话则需要明确选择一种处理方式：

• 已有请求执行时，返回 HTTP 409，由客户端稍后重试。
• 按会话进入有界队列，依次执行。
• 产品确实需要并行分支时，为不同分支使用独立的会话或线程标识。

互斥范围应覆盖一次完整的 agent 调用，包括历史读取、模型和工具执行，以及状态保存。只锁最后的写入操作，仍然可能基于旧历史生成回答。

asyncio.Lock 只能协调当前进程内的协程，多个 worker 之间需要共享的协调机制。使用 Redis 租约锁时，至少要处理持有者标识、原子释放、续租、进程崩溃，以及租约丢失后停止执行的问题。仅设置固定的过期时间，无法保证运行较久的 agent 始终独占会话。

如果要求严格的消息顺序，通常更容易通过按会话调度的队列实现。队列也必须保证同一会话不会被两个执行者同时消费。

5. 保留旧 executor 时，如何接入 Redis 历史

隔离 memory 不需要一次性迁移整个应用。支持 Runnable 接口的旧 executor 可以使用 RunnableWithMessageHistory，在调用时根据 session_id 获取历史对象。

使用下面的片段有几个前提：executor 不再自行管理 memory，prompt 中包含 MessagesPlaceholder("chat_history")，输入键为 input，最终回答键为 output。

import os

from langchain_community.chat_message_histories import (
    RedisChatMessageHistory,
)
from langchain_core.runnables.history import (
    RunnableWithMessageHistory,
)


def get_session_history(session_id: str):
    return RedisChatMessageHistory(
        session_id=session_id,
        url=os.environ["REDIS_URL"],
        key_prefix="chat:",
    )


agent_with_history = RunnableWithMessageHistory(
    executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="output",
)

完成授权校验并取得会话互斥后调用：

result = await agent_with_history.ainvoke(
    {"input": query},
    config={
        "configurable": {
            "session_id": session_key,
        }
    },
)

response = result["output"]

history 对象负责访问对应会话的数据，无须把每个会话的 Python 对象永久缓存到 worker 内存中。RunnableWithMessageHistory 的键名必须与实际 executor 的输入输出一致。参考：RunnableWithMessageHistory

示例使用的是 langchain_community 中的 RedisChatMessageHistory。不同 Redis 集成包里的同名类可能使用不同的构造参数，不能混用。参考：RedisChatMessageHistory

高流量下，还要检查该实现是否反复创建客户端或连接池，以及异步方法的实际执行方式。存在 aget_messages()、aadd_messages() 方法，并不代表底层一定使用原生异步 Redis I/O。参考：BaseChatMessageHistory

同一条执行路径应只保留一个历史管理入口，避免旧 memory、历史包装器和 checkpointer 重复保存同一轮消息。

6. 异步阻塞要定位到具体组件

await agent.arun(...) 或 await agent.ainvoke(...) 的调用链中仍可能包含同步代码。排查延迟时，需要区分以下情况：

• 事件循环被同步 I/O 或计算阻塞。
• 请求正在等待连接池、线程池或并发名额。
• 模型服务响应慢或正在重试。

后两种情况会增加延迟，但不一定阻塞事件循环。

优先检查自定义工具、同步 HTTP 请求、数据库驱动、历史读写和回调。能改用原生异步客户端的组件，应直接异步调用。

暂时无法替换的同步 I/O，可以放在线程中执行：

async def lookup_tool(query: str) -> str:
    return await asyncio.to_thread(
        blocking_lookup,
        query,
    )

线程池应接收同步函数，不能这样包装协程：

# 错误：线程通常只会得到一个尚未执行的协程对象
await loop.run_in_executor(None, agent.arun, query)

如果需要把整个同步 executor 放到线程中，可以调用它的同步入口：

result = await asyncio.to_thread(
    sync_executor.invoke,
    {"input": query},
)

这要求 executor 及其依赖适合在线程中使用，不能随意移入绑定主事件循环的异步客户端。通常先隔离具体的阻塞工具，更容易控制资源消耗。参考：to_thread、run_in_executor

纯 Python 的 CPU 密集任务通常应使用进程池或独立任务服务。请求超时不会强制终止已经运行的线程，外部工具操作也可能继续执行。

7. 多 worker 下需要统一哪些配置

四个 worker 各自拥有一份 agent、连接池和进程内变量。普通 Python 内存不会在这些进程之间自动同步。参考：FastAPI 部署概念

上线前需要落实以下事项：

• 所有 worker 使用相同的状态存储、会话键规则和命名空间。
• 统一 agent、prompt、工具和状态 schema 的发布版本，处理滚动升级期间的兼容性。
• 按全部 worker 计算连接数和并发上限。例如，每个 worker 的 Semaphore(16) 在四个 worker 下允许最多 64 次调用进入执行区，无法起到全局限流的作用。
• 为模型调用、工具调用和排队分别设置预算。流量超过容量时，使用有界队列或明确拒绝请求，避免无限等待。
• 为历史设置保留和裁剪策略。即使存储了完整历史，也不必每次都全部送给模型；裁剪时要保持工具调用与工具结果的消息关系有效。
• 对有副作用的工具实现幂等处理，避免请求重试、agent 重复运行时，再次执行同一项业务操作。

验收至少应覆盖不同用户之间的数据隔离、同一会话的并发请求，以及请求切换 worker 后的会话恢复。记录模型耗时、工具耗时、状态读写耗时、排队时间和事件循环延迟，用这些数据判断下一步该增加执行容量，还是修复具体的阻塞组件。

备注：内容仅供参考。