LangChain 上线部署：把链和 Agent 变成能对外服务的 API

LangChain 上线部署：把链和 Agent 变成能对外服务的 API

链在本地跑通只是第一步。要给别人用，得变成 HTTP 服务，
还得处理流式、并发、超时、日志这几件事。

这篇讲两条路线：LangServe（官方，快）和 FastAPI 自己包（可控），
以及上线前必须做的检查项。

一、两条路线怎么选

我的建议：原型阶段用 LangServe 快速验证，
真正上线时如果团队已有 Python 后端，用 FastAPI 自己包更容易和现有体系对齐。

二、LangServe：五分钟出一个服务

pipinstall"langserve[all]"fastapi uvicorn

# server.pyfromfastapiimportFastAPIfromlangserveimportadd_routesfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser app=FastAPI(title="我的 LLM 服务",version="1.0")chain=(ChatPromptTemplate.from_template("用一句话解释：{topic}")|ChatOpenAI(model="gpt-4o-mini")|StrOutputParser())add_routes(app,chain,path="/explain")if__name__=="__main__":importuvicorn uvicorn.run(app,host="0.0.0.0",port=8000)

启动后自动获得这些接口：

2.1 调用

fromlangserveimportRemoteRunnable remote=RemoteRunnable("http://localhost:8000/explain")print(remote.invoke({"topic":"向量数据库"}))forchunkinremote.stream({"topic":"向量数据库"}):print(chunk,end="",flush=True)

或者直接用 curl：

curl-XPOST http://localhost:8000/explain/invoke\-H"Content-Type: application/json"\-d'{"input": {"topic": "向量数据库"}}'

2.2 Playground 很实用

浏览器打开http://localhost:8000/explain/playground，
有个现成的调试界面，能直接看到流式输出和中间步骤。
给产品、测试同学演示时特别省事，不用自己写前端。

2.3 部署带记忆的 Agent

fromlangserveimportadd_routesfromlanggraph.checkpoint.sqliteimportSqliteSaverfromlanggraph.prebuiltimportcreate_react_agent saver=SqliteSaver.from_conn_string("checkpoints.db")agent=create_react_agent(llm,tools,checkpointer=saver)add_routes(app,agent,path="/agent")

调用时带上thread_id：

remote=RemoteRunnable("http://localhost:8000/agent")remote.invoke({"messages":[("user","我叫小明")]},config={"configurable":{"thread_id":"u-1"}},)

三、FastAPI 自己包：完全可控

生产环境我更推荐这条路线。核心是三件事：请求体、流式、配置透传。

fromfastapiimportFastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModel,Fieldimportjson app=FastAPI()classAskRequest(BaseModel):question:str=Field(min_length=1,max_length=2000)session_id:str=Field(default="default")stream:bool=Field(default=False)@app.post("/ask")asyncdefask(req:AskRequest):config={"configurable":{"thread_id":req.session_id},"recursion_limit":12}try:result=awaitchain.ainvoke({"question":req.question},config=config)return{"answer":result}exceptExceptionase:# 别把内部异常直接抛给用户logger.exception("ask failed")raiseHTTPException(status_code=500,detail="服务暂时不可用")

3.1 流式输出（SSE）

这是用户体验的关键。两种粒度：

按 token 流（适合纯文本生成）：

asyncdeftoken_stream(question:str,config:dict):asyncforchunkinchain.astream({"question":question},config=config):# chunk 是字符串片段（链最后接了 StrOutputParser）yieldf"data:{json.dumps({'content':chunk},ensure_ascii=False)}\n\n"yield"data: [DONE]\n\n"@app.post("/ask/stream")asyncdefask_stream(req:AskRequest):returnStreamingResponse(token_stream(req.question,config),media_type="text/event-stream",headers={"Cache-Control":"no-cache","X-Accel-Buffering":"no"},)

按步骤流（适合 Agent，让用户看到"在查什么"）：

asyncdefstep_stream(question:str,config:dict):asyncforeventinagent.astream_events({"messages":[("user",question)]},config=config,version="v2"):kind=event["event"]ifkind=="on_tool_start":yieldf"data:{json.dumps({'type':'tool','name':event['name'],'input':event['data'].get('input')},ensure_ascii=False)}\n\n"elifkind=="on_chat_model_stream":delta=event["data"]["chunk"].contentifdelta:yieldf"data:{json.dumps({'type':'token','content':delta},ensure_ascii=False)}\n\n"yield"data: [DONE]\n\n"

X-Accel-Buffering: no这个 header 很重要——
Nginx 默认会缓冲响应，不加这个 header 流式就变成了"等全部完成再返回"。

3.2 并发与超时

LLM 调用是 IO 密集，务必用 async：

# ✅ 异步，多个请求真正并发result=awaitchain.ainvoke(inputs)# ❌ 同步会阻塞事件循环，一个慢请求拖垮所有result=chain.invoke(inputs)

超时要设两层：

importasyncioasyncdefwith_timeout(coro,seconds=60):try:returnawaitasyncio.wait_for(coro,timeout=seconds)exceptasyncio.TimeoutError:return"抱歉，处理超时，请稍后再试"answer=awaitwith_timeout(chain.ainvoke(inputs),60)

外层还要在网关（Nginx / K8s ingress）设超时，
比应用层稍大一些，避免网关先断导致连接悬挂。

四、上线必做的六件事

1. 密钥管理

# ❌ 硬编码os.environ["OPENAI_API_KEY"]="sk-xxx"# ✅ 从环境变量 / 密钥服务读importos api_key=os.environ["OPENAI_API_KEY"]# 缺失就报错，别用默认值

永远不要把 key 写进代码或镜像。

2. 别把内部异常抛给用户

exceptExceptionase:logger.exception("chain failed")# 详细堆栈进日志raiseHTTPException(500,"服务暂时不可用")# 用户只看到这句

堆栈里有文件路径、内部接口，暴露出去是安全问题。

3. 请求体要校验

classAskRequest(BaseModel):question:str=Field(min_length=1,max_length=2000)

不设上限的话，用户传 10 万字进来，token 费用会教你做人。

4. 限流

按用户限流，防止单人刷爆配额：

fromslowapiimportLimiterfromslowapi.utilimportget_remote_address limiter=Limiter(key_func=get_remote_address)@app.post("/ask")@limiter.limit("20/minute")asyncdefask(req:AskRequest):...

5. 成本监控

每次调用记录 token 消耗：

fromlangchain_core.callbacksimportBaseCallbackHandlerclassCostTracker(BaseCallbackHandler):defon_llm_end(self,response,**kwargs):usage=response.llm_output.get("token_usage",{})metrics.record(prompt=usage.get("prompt_tokens",0),completion=usage.get("completion_tokens",0),)logger.info("token_usage",extra=usage)

上线不看成本，月底账单会很难看。

6. 日志要带 trace_id

importuuid trace_id=str(uuid.uuid4())config={"configurable":{"thread_id":session_id},"metadata":{"trace_id":trace_id}}

用户投诉"刚才答错了"时，有没有 trace_id 决定了你排查要花 5 分钟还是 5 小时。

五、部署形态

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

几个注意点：

• --workers数要小于 CPU 核数的 2 倍，
  LLM 调用是 IO 密集，但 Pydantic 校验和序列化是 CPU 活；
• 如果用InMemorySaver，多 worker 之间状态不共享，
  必须换 Sqlite/Postgres，否则用户请求打到不同 worker 会"失忆"；
• 健康检查接口要单独加（/health），别让探针去打 LLM。

六、小结

• LangServe 适合快速验证：一行add_routes得到 invoke/stream/batch + Playground；
• FastAPI 自己包适合生产：可控性强，容易和现有后端体系对齐；
• 流式用 SSE，记得加X-Accel-Buffering: no，否则 Nginx 会缓冲掉流式效果；
• 一律用ainvoke/astream，同步调用会阻塞事件循环；
• 上线六件事：密钥入环境变量、异常不外抛、请求体限长度、按用户限流、
  记录 token 成本、日志带 trace_id；
• 多 worker 部署时InMemorySaver会失效，必须换持久化 checkpointer。

下一篇是本系列最后一篇：效果评测与回归测试。