LangServe

🚀 LangServe 是什么？它解决了什么问题？¶

LangServe 是 LangChain 官方推出的部署工具，它能将你的 LangChain 链（LCEL 链、Agent 等）自动转化为生产级的 REST API。它不是一个独立的框架，而是基于 FastAPI 构建的一个高级封装，能够感知 LangChain 的 Runnable 接口，并暴露相应的端点。

在 LangServe 出现之前，开发者要将一条链部署为 API，通常需要手动编写大量的 Web 框架（如 FastAPI）代码：定义输入输出模型、处理请求参数、处理流式响应、编写 Swagger 文档、实现错误处理等。这是一套重复且易出错的“体力活”。LangServe 通过简单的 add_routes 函数，将这些常见的部署需求自动化了。

LangServe 具体解决了以下问题：

零样板代码部署：只需几行代码就能将任何 LCEL 链（Runnable 对象）暴露为 REST API，无需手动编写路由函数。
自动生成 OpenAPI 文档：由于 LangChain 的 Runnable 有明确的输入输出 schema（通过 with_types 或 Pydantic 模型），LangServe 能够自动生成符合 OpenAPI 规范的文档，并附带 Swagger UI。
原生流式支持：自动为链创建 /stream 和 /stream_log 等端点，支持 Server-Sent Events (SSE) 和 WebSocket，让流式输出即开即用。
内置 Playground：提供一个可选的 Web 界面，开发者可以在浏览器中直接测试链，调整参数，查看输出，极大地方便了调试和演示。
客户端自动生成：LangServe 服务端可以提供一个 /openapi.json，LangChain 的 RemoteRunnable 客户端可以直接通过这个 URL 远程调用链，实现了“服务端代码即客户端定义”。
配置与扩展性：支持通过 config 动态传递回调、标签等运行时配置；可以集成 FastAPI 的依赖注入、中间件等，便于添加认证、日志等功能。
生产级特性：基于 FastAPI，天然支持高性能异步、自动验证、错误处理等生产必备能力。

简单来说，LangServe 把 LangChain 开发者的主要精力从“如何部署”转移回了“如何构建更好的链”，它把部署过程标准化、自动化了，是 LangChain 从原型到产品的桥梁。

💻 用 LangServe 将一个 LCEL 链部署为 API，最少需要几行代码？写出示例。¶

最少只需要 6 行有效代码（除去导入和模型初始化）。以下是一个最简示例，将一个简单的问答链部署为 API：

# 1. 导入必要模块
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 2. 创建FastAPI应用
app = FastAPI(title="我的LLM服务")

# 3. 构建LCEL链
prompt = ChatPromptTemplate.from_template("请解释：{topic}")
llm = ChatOpenAI(model="gpt-3.5-turbo")
chain = prompt | llm

# 4. 将链添加为路由（核心！）
add_routes(app, chain, path="/explain")

这段代码做了以下事情：

创建了一个 FastAPI 实例。
定义了一个简单的 LCEL 链：接受 topic，生成解释。
使用 add_routes 将这个链挂载到 /explain 路径上。

启动服务（uvicorn app:app --reload）后，你将自动获得以下端点：

POST /explain/invoke：单次调用，返回完整结果。
POST /explain/stream：流式调用，返回 SSE 流。
POST /explain/batch：批量调用。
GET /explain/playground/：一个在线测试页面（需显式开启 playground=True）。

如果想进一步控制输入输出，可以为链添加类型注解，LangServe 会自动生成对应的 JSON Schema：

from pydantic import BaseModel
class TopicInput(BaseModel):
    topic: str
chain = prompt | llm

# 指定输入类型
add_routes(app, chain.with_types(input_type=TopicInput), path="/explain")

这样 OpenAPI 文档就会清晰地展示 topic 是一个必填的字符串字段。可见 LangServe 将部署复杂度降到了最低，让开发者把精力集中在链的逻辑上。

📖 LangServe 如何自动生成 OpenAPI 文档？客户端如何利用它？¶

自动生成 OpenAPI 文档的原理 LangServe 基于 FastAPI 构建，FastAPI 本身具备从 Python 类型注解（Pydantic 模型）自动生成 OpenAPI 文档的能力。当使用 add_routes(app, chain) 时，LangServe 内部会：

检查链的输入输出类型。如果链通过 with_types(input_type=..., output_type=...) 指定了 Pydantic 模型，LangServe 就直接使用它们；否则，它会使用默认的字典类型，但仍可推断基本结构。
创建 FastAPI 路由，并将这些类型信息传递给路由装饰器（如 @router.post），这样 FastAPI 的 OpenAPI 生成器就能自动构建 API 文档。
在启动应用后，访问 /docs 或 /redoc 即可看到完整的 API 文档，包括请求体示例、响应模型、流式端点说明等。

客户端如何利用 OpenAPI

客户端有两种主要方式利用 LangServe 的 OpenAPI 文档：

方式一：使用 RemoteRunnable（推荐） LangChain 提供了一个 RemoteRunnable 类，它就像一个远程的链代理。你只需提供服务端的 URL（通常是 /invoke 或 /stream 的路径，也可以是根路径），RemoteRunnable 会自动从服务端获取 OpenAPI 规范，解析输入输出 schema，并在本地暴露相同的接口。这使得远程调用就像在本地执行一样。

from langserve import RemoteRunnable

remote_chain = RemoteRunnable("http://localhost:8000/explain/")
# 同步调用
result = remote_chain.invoke({"topic": "黑洞"})
# 流式调用
for chunk in remote_chain.stream({"topic": "黑洞"}):
    print(chunk, end="")

RemoteRunnable 内部使用 httpx 进行 HTTP 请求，并且支持同步/异步、重试、回调等。它极大简化了客户端的开发，不需要手动处理 HTTP 请求和响应解析。

方式二：直接使用 HTTP 客户端对于非 Python 客户端，可以基于 OpenAPI 文档生成任意语言的 SDK（例如通过 OpenAPI Generator），或者直接使用 curl、fetch 等发送 POST 请求到 /invoke，请求体为 JSON，响应体也是 JSON。对于流式，则需要处理 SSE 协议。

额外说明：如果你在链中定义了自定义的 Pydantic 输入输出模型，这些模型会被完整地暴露在 OpenAPI 中，使得客户端能够获得精确的类型检查和自动补全（在 Python 中），这对于大型团队协作非常有价值。

🎮 如何在 LangServe 中启用 playground？它能做什么？¶

启用 playground 非常简单，只需在 add_routes 中设置 playground=True：

add_routes(app, chain, path="/chat", playground=True)

然后访问 http://localhost:8000/chat/playground/ 即可看到一个交互式 Web 界面。Playground 默认提供一个类似聊天窗口的布局，但它的能力远超一个简单的聊天框。

Playground 能做什么？

实时测试链：开发者可以在浏览器中直接输入测试数据，立即查看链的输出，而无需编写任何前端代码。这是调试链逻辑最快捷的方式。
查看流式输出：如果链支持流式，Playground 会自动以打字机效果展示生成的 token，让你直观感受流式响应的效果。
修改运行配置：可以在界面中动态设置 config 参数，例如传入 metadata、tags、或者调整模型参数（如 temperature），并观察不同配置下的输出变化。
查看中间步骤：如果你的链包含多个步骤（例如 Agent 的中间推理），Playground 可以展示 stream_log 的输出，让你清晰地看到每一步的输入输出，这对于调试复杂的多步链非常有帮助。
查看 API 文档：Playground 页面通常还集成了 Swagger UI 的链接，方便直接跳转到 /docs 查看详细的 API 定义。
即时的反馈循环：对于非技术人员（如产品经理），Playground 提供了一个无代码环境来体验 LLM 功能，有助于需求对齐和早期验证。

注意事项：Playground 默认是开发工具，不建议在生产环境暴露（除非有适当的安全措施），因为它可以执行任意链调用，可能消耗大量 API 额度。在安全方面，你可以通过 FastAPI 的依赖注入或中间件为 Playground 添加认证，确保只有授权用户能访问。

🌊 LangServe 支持流式输出吗？客户端应该如何消费流式响应？¶

支持，且是原生支持。 LangServe 会为你的链自动生成 /stream 和 /stream_log 端点，只要你的链本身支持流式（即 LLM 组件开启了 streaming=True）。你不需要编写任何额外的流式处理逻辑。

服务端行为：

POST /stream 端点接收同样的 JSON 输入，但返回一个 text/event-stream 流。每当链的 stream 方法产出新的事件（通常是 token），服务端就将其包装为 SSE 事件发送给客户端。
POST /stream_log 则提供更详细的事件，包括 on_chain_start、on_llm_start、on_llm_new_token 等信息，适用于需要监控中间状态的场景。

客户端消费方式：

使用 RemoteRunnable（Python）

remote_chain = RemoteRunnable("http://localhost:8000/chat/")
for chunk in remote_chain.stream({"topic": "量子计算"}):
    print(chunk, end="", flush=True)

内部，RemoteRunnable.stream 会发送 POST 请求到 /stream，并逐行解析 SSE 响应，将每个 data 字段的内容反序列化为 Python 对象（例如字符串或 AIMessageChunk）并 yield。

使用 httpx 直接处理 SSE（Python）

import httpx, json
with httpx.stream("POST", "http://localhost:8000/chat/stream", json={"topic": "AI"}) as r:
    for line in r.iter_lines():
        if line.startswith("data:"):
            payload = line[5:].strip()
            if payload:  # 忽略空行
                print(json.loads(payload), end="")

前端 JavaScript 消费（浏览器）使用 EventSource API（注意：EventSource 只支持 GET，但 LangServe 的流式端点通常是 POST。你需要使用 fetch API 并手动处理 ReadableStream，或者使用支持 POST 的 SSE 库）。典型的 fetch 流式处理代码如下：

const response = await fetch("/chat/stream", {
    method: "POST",
    headers: {"Content-Type": "application/json"},
    body: JSON.stringify({ topic: "AI" })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value);
    // 解析 SSE 格式并处理
    document.getElementById("output").innerText += chunk;
}

注意：流式端点要求链内部 LLM 启用 streaming=True，否则虽然端点存在，但会退化为一次返回全部内容，失去了流式的意义。另外，如果客户端断开连接，服务端应能感知并停止生成，这依赖于 Web 服务器的异步能力，FastAPI 可以优雅处理。

🔐 在 LangServe 中，如何为 API 添加认证和授权？¶

LangServe 本身不提供内置的认证功能，但由于它是基于 FastAPI 的，你可以利用 FastAPI 强大的依赖注入系统来灵活地实现认证和授权。推荐的做法是通过 per_req_config_modifier 或 FastAPI 的 Depends 来控制访问权限。

方法一：使用 FastAPI 的 Depends 保护整个路由

你可以在添加路由之前，为 FastAPI 应用添加全局依赖，或者使用 APIRouter 并应用依赖。例如使用 Bearer Token 认证：

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from langserve import add_routes

app = FastAPI()
security = HTTPBearer()

# 验证函数
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    if token != "my-secret-token":
        raise HTTPException(status_code=403, detail="Invalid token")
    return token

# 将依赖应用到整个应用或特定路由
app.add_middleware(...)  # 也可以使用中间件
# 或者为 add_routes 创建一个带有依赖的 APIRouter
from fastapi import APIRouter
router = APIRouter(dependencies=[Depends(verify_token)])
add_routes(router, chain, path="/chat")
app.include_router(router)

方法二：使用 per_req_config_modifier 注入认证信息 per_req_config_modifier 是一个函数，它在每个请求处理前被调用，可以修改传递给链的 config。你可以在其中进行认证检查，并将认证后的用户信息放入 config 的 metadata 中，供链内部使用。

from fastapi import Request, HTTPException

def auth_modifier(config, request: Request):
    # 从请求头获取token并验证
    token = request.headers.get("Authorization")
    if not token or token != "Bearer valid-token":
        raise HTTPException(status_code=403)
    # 将用户信息注入config
    config["metadata"]["user_id"] = "user123"
    return config

add_routes(app, chain, path="/chat", per_req_config_modifier=auth_modifier)

这个方法的好处是认证逻辑与业务链解耦，并且可以在不修改链的情况下传递上下文。

方法三：结合 API 网关

在生产环境中，更常见的做法是在 LangServe 服务前面放置一个 API 网关（如 Kong、Nginx、Cloudflare Zero Trust），由网关处理认证、限流、路由等。这样 LangServe 服务本身可以专注于业务逻辑，不需要处理认证细节。这是微服务架构的最佳实践。

权限授权：如果需要更细粒度的权限控制（比如不同用户只能访问特定的工具），可以在链内部或工具函数中检查 config 中传递的 user_id，并根据角色决定是否执行操作。同时配合 per_req_config_modifier 将用户角色注入，实现完整的访问控制。

安全提醒：不要将 LangServe 暴露在没有认证的公网上，尤其是 Playground，它可能被恶意利用消耗大量 API 费用。始终加上认证层，并限制速率。

📂 你如何在 LangServe 中处理文件上传？（例如用户上传 PDF）¶

LangServe 默认的 add_routes 不支持文件上传，因为它期望的输入是 JSON 格式。但你可以通过自定义 FastAPI 端点来扩展 LangServe 应用，处理文件上传，然后将文件内容传递给链。

实现步骤：

创建自定义上传端点：在你的 FastAPI 应用中，使用标准的 UploadFile 处理文件。
读取文件内容：例如对于 PDF，使用 PyPDF2 或 langchain_community.document_loaders.PyPDFLoader 加载文本。
调用你的 LCEL 链：将提取的文本作为输入的一部分，调用链的 ainvoke 或其他方法，返回结果。

示例代码：

from fastapi import FastAPI, File, UploadFile, HTTPException
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import tempfile, os

app = FastAPI()

# 标准LCEL链
prompt = ChatPromptTemplate.from_template("请总结以下文档：{text}")
llm = ChatOpenAI(model="gpt-3.5-turbo")
chain = prompt | llm

# 添加常规链路由
add_routes(app, chain, path="/summarize")

# 自定义文件上传和调用链的端点
@app.post("/upload-summarize")
async def upload_summarize(file: UploadFile = File(...)):
    if not file.filename.endswith(".pdf"):
        raise HTTPException(400, "仅支持PDF文件")

    # 保存临时文件
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        # 使用LangChain的PyPDFLoader加载文本
        from langchain_community.document_loaders import PyPDFLoader
        loader = PyPDFLoader(tmp_path)
        docs = loader.load()
        text = "\n".join([doc.page_content for doc in docs])
    finally:
        os.unlink(tmp_path)

    # 调用链
    result = await chain.ainvoke({"text": text})
    return {"summary": result}

这样，你就可以通过 POST /upload-summarize 上传文件并获取摘要。如果你希望这个自定义端点也能享受 LangServe 的自动文档生成，可以使用 FastAPI 的 response_model 和 Pydantic 模型来清晰定义输出结构。

进阶方案：如果你有一个更复杂的链，它本身期望的输入包含文件路径或文本，你可以将文件上传处理逻辑封装为一个自定义工具或Runnable，然后在 add_routes 时使用 per_req_config_modifier 来预处理请求，但通常直接自定义端点更简单直接。

注意：上传的文件应做大小限制、类型检查，并尽快清理临时文件，避免资源泄漏和安全隐患。在生产环境中，你可以将文件上传至对象存储（如 S3），然后传递 URL 给链，避免直接处理大文件。

🔧 LangServe 如何与 FastAPI 中间件集成？比如 CORS、日志等。¶

LangServe 本质上是一个 FastAPI 应用（或其子应用），因此它可以无缝集成任何 FastAPI 兼容的中间件。添加中间件的方式与普通 FastAPI 应用完全一致：在创建 app 后，通过 app.add_middleware() 注册。

CORS 中间件：如果前端与 LangServe 不同源，必须配置 CORS。使用 fastapi.middleware.cors.CORSMiddleware。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from langserve import add_routes

app = FastAPI()

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 添加LangServe路由
add_routes(app, chain, path="/chat")

日志中间件：可以使用 starlette.middleware.base.BaseHTTPMiddleware 来自定义日志记录，或者集成第三方如 loguru、structlog。但要注意，HTTP 中间件会阻塞请求体读取，可能影响流式端点。更好的做法是使用 FastAPI 的依赖注入（Depends）或自定义路由来处理日志，而不是全局中间件。

认证中间件：前面问题中已有详述，同样通过 FastAPI 的依赖或中间件实现。

关键点：如果你的中间件需要访问请求体，务必注意流式端点（/stream）依赖请求体的流式传输，中间件消费了请求体会导致流式失败。因此，对于流式端点，应避免读取或修改 request.body()。

集成示例：我曾在一个项目中使用 slowapi（基于 limits 库）作为中间件来限制访问频率，将其直接挂载到 app 上，对所有路由（包括 LangServe 的）生效。配置起来与纯 FastAPI 项目没有区别。

⚖️ 你如何在 LangServe 中配置并发限制和请求队列？¶

LangServe 自身不提供内置的并发控制和请求队列，但可以借助 FastAPI 的异步特性和一些外部工具来实现。

方法一：使用 asyncio.Semaphore 在链内部限制并发这是最直接的方式，可以在创建链时传入一个全局信号量，在 ainvoke 或 astream 的路径上包裹。

import asyncio
semaphore = asyncio.Semaphore(10)  # 最多10个并发LLM调用

async def limited_chain(input):
    async with semaphore:
        return await chain.ainvoke(input)

然后将 limited_chain 包装成 RunnableLambda 再通过 add_routes 暴露。这种方式只限制核心 LLM 调用，不会阻塞其他非 LLM 操作。

方法二：利用 FastAPI 的后台任务或中间件实现排队你可以实现一个“令牌桶”或“漏桶”中间件。例如，使用 asyncio.Queue 构建一个任务队列，每个请求先入队，然后由后台 Worker 逐个处理。但这对流式响应不友好，因为流式需要长时间连接。

更实用的做法是用 arq 或 celery 等任务队列，将耗时请求转为后台任务，通过回调或轮询获取结果。但这改变了交互模式，不再是同步的 REST 响应。

方法三：在网关层实现生产环境通常会在 LangServe 前面放置 Nginx 或 Envoy，由它们进行连接数限制和排队。例如，Nginx 的 limit_conn_zone 和 limit_req_zone 可以限制单 IP 的并发连接和请求速率。超出限制的请求会收到 429 Too Many Requests，由客户端决定重试。这是最不侵入业务代码的方式。

方法四：利用 ASGI 服务器的参数例如使用 uvicorn 启动时，可以指定 --limit-concurrency 来限制最大并发连接数。超过的连接会被拒绝，但这是一种粗略的控制，不够精细。

实践经验：我在一次客服机器人的部署中，遇到了突发流量导致 LLM API 配额耗尽的问题。后来我采用了“网关限流 + 应用层信号量”的双层保护。在 Nginx 限制了单 IP 50 QPS，同时在链内部对 LLM 调用加了 Semaphore(20)，保证了系统在高压下依然稳定。

🧪 使用 LangServe 时，如何实现 A/B 测试？即根据路由将请求分流到不同的链。¶

A/B 测试在生产中非常关键，LangServe 本身没有内建分流机制，但可以灵活实现。

方案一：在 LangServe 上层使用 Nginx/Envoy 按权重分流这是最推荐的方式，因为它对应用无侵入。例如，使用 Nginx 的 split_clients 模块，根据用户 IP 或自定义 Header 将流量按比例分发到两个不同的 LangServe 实例（或同一实例的不同路由）。

split_clients "${remote_addr}" $variant {
    50%    variant_a;
    50%    variant_b;
}
location /chat {
    proxy_pass http://backend_$variant;
}

其中 backend_a 和 backend_b 分别指向部署了不同版本链的 LangServe 服务。这种方式对客户端透明，且支持复杂的规则（如按 User-Agent、Cookie 等）。

方案二：在 FastAPI 路由层面动态选择链

你可以编写一个包装端点，内部根据一定逻辑（如用户 ID 哈希、配置中心）决定调用哪条链。

@app.post("/ab_chat")
async def ab_chat(request: ChatRequest):
    # 根据用户ID尾号分流
    if request.user_id[-1] in "01":
        chain = chain_v1
    else:
        chain = chain_v2
    return await chain.ainvoke(request.dict())

然后将此自定义端点与 add_routes 共存。这种方式灵活性高，但需要自行处理输入输出。

方案三：利用 LangServe 的 per_req_config_modifier 动态切换链你可以在配置函数里根据请求参数（如 Header 中的 x-experiment）修改 configurable 字段，从而让同一条 LCEL 链依据配置选择不同的子链（通过 RunnableBranch 实现）。这要求你的链在设计时就支持分支。

chain = RunnableBranch(
    (lambda x: x["config"]["experiment"] == "A", chain_a),
    chain_b  # 默认
).with_config(...)

然后在 per_req_config_modifier 中注入实验标识。

实践建议：在线服务我首选网关分流，因为它不占用应用资源，且重启服务不会影响分流规则。对于内部测试，我会用路由层面的动态选择，方便随时调整。

🚀 LangServe 部署到生产时，你一般使用什么 ASGI 服务器？（uvicorn, gunicorn 等）¶

LangServe 应用本质是 FastAPI，因此所有符合 ASGI 的服务器都能用。我的选择取决于部署环境和并发模型。

单机部署：

Uvicorn：最简单，适合开发和小规模服务。使用 uvicorn.run(app, host="0.0.0.0", port=8000)。但 Uvicorn 默认只有一个 worker，不能充分利用多核 CPU。
Gunicorn + UvicornWorker：生产环境最常用。Gunicorn 作为进程管理器，每个 Worker 内嵌一个 Uvicorn 实例。这样既可以利用多核，又可以保持异步高性能。

gunicorn app:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Hypercorn：另一个支持 HTTP/2 的 ASGI 服务器，但社区较小。

容器化部署（Kubernetes）：

通常直接用 Uvicorn（一个容器一个进程），因为 K8s 本身负责多副本和负载均衡。每个 Pod 运行 uvicorn app:app --host 0.0.0.0 --port 8000。通过 HPA（水平自动伸缩器）根据 CPU/内存或自定义指标扩缩容。

关键配置：

超时设置：流式请求可能持续时间较长（数十秒），必须调整 Gunicorn/Uvicorn 的超时参数（--timeout 120），否则连接会被过早断开。
并发连接数：Uvicorn 默认的 --limit-max-connections 可根据内存大小适当调高，避免拒绝请求。
资源限制：在 Gunicorn 中设置 --max-requests 和 --max-requests-jitter，定期重启 Worker，防止内存泄漏累积。

额外建议：如果是面向 Web 客户端的流式服务，强烈建议在 Uvicorn 前面加一层 Nginx，负责 SSL 终端、静态文件、缓存、限流等，保护应用服务器。

🌐 如果 LangServe 服务需要水平扩展，你会怎么设计？共享什么状态？¶

LangServe 服务是无状态的，这使其水平扩展相对简单。扩展时，需要注意哪些状态需要共享、哪些可以独立。

架构设计：

无状态服务：每个 LangServe 实例是独立的，不存储任何本地状态。用户请求可以由任意实例处理。
负载均衡：在服务前端放置负载均衡器（如 Nginx、HAProxy、云服务商的 ELB），将流量分发到多个实例。对于 WebSocket（如果需要），需配置 sticky session（根据 session_id 路由到同一实例），因为 LangServe 本身不强制。
共享存储：
LLM API Key 和配置：通过环境变量或配置中心（如 Consul、Kubernetes ConfigMap）统一管理，各实例相同。
对话记忆：如果使用了需要持久化的 Memory，必须使用共享存储如 Redis 或 PostgreSQL。每个实例通过相同 session ID 访问共享的 ChatMessageHistory，保证用户对话连贯。
工具/缓存：如 Redis 用作 LLM 响应缓存、速率限制计数器，需要所有实例共享同一个 Redis 集群。
独立存储：
本地日志：每个实例可以输出到 stdout，由日志收集系统（如 Loki、ELK）统一采集。
回调处理器：如果回调中包含本地文件写入等有状态操作，需要改为网络输出或共享存储。

扩展实例：在 Kubernetes 中，通过 Deployment 设置 replicas，配合 HPA 根据 CPU/内存使用率自动扩缩。由于是异步服务，通常 CPU 不会成为瓶颈，更可能的是并发连接数或 LLM API 限制，所以 HPA 可以基于自定义指标（如活跃请求数）触发。

共享什么？

必须共享：LLM 缓存（Redis）、聊天记忆（Redis/DB）、限流计数器（Redis）。
无需共享：本地模型文件（如 LangChain 代码本身）、瞬时请求上下文、临时文件。

实践：我们一个 RAG 服务就是采用这种架构：4 个 Pod 的 LangServe，前面挂载 Nginx Ingress Controller，共享 Redis 用于缓存和记忆，PG 存储审计日志。当流量翻倍时，手动增加副本数，服务零中断。

🤖 LangServe 是否适合部署 Agent？会有什么额外挑战？¶

LangServe 适合部署 Agent，但 Agent 的复杂性和长时间运行特性带来了额外挑战。

适合之处：

LangServe 天然支持 stream 和 stream_log，可以实时推送 Agent 的思考过程和工具调用结果，给用户良好的交互体验。
RemoteRunnable 可以像调用普通链一样调用 Agent，开发客户端非常方便。

额外挑战与应对：

执行时间长，超时风险高：Agent 可能需要数十步推理，持续几分钟。HTTP 请求容易超时。解决方案：
将 Agent 作为后台任务，客户端轮询或通过 WebSocket 获取进度。
使用 stream_log SSE 流式输出，即使执行很久，连接不断开。
调大 Gateway 和 Uvicorn 的超时时间（--timeout 600）。
状态管理：Agent 的 intermediate_steps 如果丢失，重试会重复执行。需要将中间状态存入共享存储（Redis），实现断点续传。
成本和安全：Agent 可能无限循环或调用高危工具。必须：
严格限制 max_iterations。
在工具内部加入权限校验和人工确认（如删除操作）。
通过回调监控异常行为，必要时自动中断。
错误恢复：Agent 某一步失败（如 API 超时）不应导致整个请求失败。应在 AgentExecutor 内捕获异常，返回 AgentAction 重试或返回友好错误。
并发控制：Agent 消耗资源多，需要更严格的并发限制，防止系统被单个用户的任务占满。

我部署过的 Agent：一个自动生成报告的 Agent，用户提交主题，Agent 搜索、分析、撰写。我们用了异步 astream_events 将每一步推送到前端，并在 Nginx 设置了 10 分钟超时。同时，Agent 内部用 Semaphore(5) 控制并发，防止 API 过载。

🩺 监控 LangServe 服务的健康状态，你会暴露哪些端点？¶

健康检查是服务可靠性的基础。除了最基本的“存活”检查，还应暴露“就绪”和“依赖”状态。

暴露的端点：

/healthz（存活检查）：返回 200，表示进程在运行。通常由 Kubernetes livenessProbe 使用。
/readyz（就绪检查）：检查关键依赖是否可用，如 LLM API 连通性、Redis 连接、数据库连接等。如果不可用，返回 503。Kubernetes readinessProbe 使用，防止流量进入不健康的实例。
/metrics（指标）：Prometheus 抓取端点，暴露 LLM 调用次数、延迟、错误率等（通过前面提到的回调实现）。
/debug/pprof 或 py-spy 接口（可选）：性能剖析，仅在内部网络开放。

实现示例：

from fastapi import FastAPI
import redis, httpx

app = FastAPI()

@app.get("/readyz")
async def readyz():
    # 检查Redis
    try:
        r = redis.Redis(...)
        r.ping()
    except:
        return {"status": "unhealthy", "detail": "redis down"}, 503
    # 检查LLM API
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get("https://api.openai.com/v1/models", timeout=5)
            if resp.status_code != 200:
                raise Exception("OpenAI API error")
    except:
        return {"status": "degraded", "detail": "llm api unreachable"}, 200
    return {"status": "ok"}

额外监控：

通过 Prometheus AlertManager 设置告警，例如错误率 >5%、P99 延迟 >10s、实例数异常等。
在 Grafana 仪表盘中集成健康状态指示灯。

实践：我曾因 LLM API 临时故障导致服务雪崩，后来在 /readyz 中加入了对 API 的主动探测，当探测失败时自动将 Pod 标为 Not Ready，Kubernetes 停止向该 Pod 转发流量，避免请求堆积。

💣 你有没有在生产环境中使用 LangServe 的经验？踩过哪些坑？¶

有，而且坑不少。LangServe 让部署变得极快，但生产环境要求高可用、可观测、安全，这些需要额外工作。

踩过的坑及教训：

流式端点内存泄漏
现象：服务运行数小时后，内存持续上涨，最终 OOMKilled。
原因：在 stream 回调中，我们将每个 token 写入一个全局列表用于后续审计，但流式连接断开时，该列表未被清理。
解决：永远不要为每个请求维护全局状态。改用异步队列，并在连接断开时清理。对于审计，改为异步写入消息队列。
Playground 暴露到公网被滥用
教训：测试时把 Playground 开到了公网，几天后发现产生了巨额 API 账单。有人写脚本自动调用。
解决：Playground 必须放在认证后面，或者只在内部网络开放。生产环境默认关闭。
add_routes 自动生成的 schema 不准确
现象：客户端调用时报错 ValidationError，因为 LangServe 生成的 OpenAPI 中字段类型与实际不符。
原因：链没有显式定义 input_type 和 output_type，LangServe 靠推测，结果错误。
解决：务必为链使用 .with_types(input_type=MyModel, output_type=MyOutModel)，确保 Schema 正确。
Gunicorn Worker 超时导致请求中断
现象：Agent 执行超过 30 秒就返回 502。
原因：Gunicorn 默认 timeout 是 30 秒，超时后强制杀死 Worker。
解决：将 Gunicorn 的 --timeout 设为 300 秒，同时为 Agent 设置 max_iterations 并优化工具调用。流式请求建议用 Uvicorn 而非 Gunicorn，因为 Uvicorn 本身没有强制 Worker 超时。
CORS 配置不当导致前端无法调用
现象：浏览器报 Access-Control-Allow-Origin 错误。
原因：忘记添加 CORS 中间件，或者只允许了 GET 方法，而 LangServe 的 /invoke 是 POST。
解决：在 add_middleware 中允许所有来源（开发阶段），生产环境指定具体域名，并允许 Content-Type 和 Authorization 头。
依赖共享状态导致扩展失败
现象：将聊天记忆存储在本地内存，扩展到多个实例后，用户对话经常丢失。
解决：将所有 Memory 改为 RedisChatMessageHistory，确保所有实例共享同一 Redis。
日志爆炸
现象：开启了 verbose=True 并记录每个 token，磁盘迅速写满。
解决：生产环境关闭详细日志，只记录请求摘要和错误；使用日志采样和轮转。

总结：LangServe 是原型到产品的加速器，但生产化需要你补齐监控、安全、高可用、资源管理等方面的能力。它是一种“带着镣铐的便利”，理解其底层 FastAPI 的运作机制，是避免踩坑的关键。