07. LangServe 实战:一键将 LangChain 应用封装为高性能 API
摘要:LangServe是LangChain官方推出的高性能服务框架,基于FastAPI构建,可一键将LangChain应用部署为RESTful API。它通过add_routes自动生成接口与Swagger文档,支持同步调用(/invoke)和流式输出(/stream),同时保留FastAPI的灵活性,可添加认证或自定义路由。开发者只需专注AI逻辑设计,无需手动封装接口,极大简化了从开发到部署的
摘要: LangServe是LangChain官方推出的高性能服务框架,基于FastAPI构建,可一键将LangChain应用(如Agent、Chain)部署为RESTful API。通过add_routes自动生成接口与Swagger文档,支持同步调用(/invoke)和流式输出(/stream),同时保留FastAPI的灵活性,可添加认证(如Token验证)或自定义路由。适用于快速部署AI服务,尤其适合学生或开发者专注AI逻辑而无需手动封装接口。
前言
上一篇文章中,我们刚刚领略了 FastAPI 的高性能与优雅,学会了如何利用 Python 类型提示构建现代化的 Web 接口。
然而,在 AI 开发的道路上,我们不仅需要构建接口,更需要将复杂的 LangChain 应用(如 Agent、Chain)快速部署为生产环境可用的服务。如果手动编写 FastAPI 代码来封装每一个 Chain,不仅繁琐,还容易出错。
今天,就让我们进入 LangServe 的世界。它是 LangChain 官方推出的高性能服务框架,能够让我们仅用一行代码,就将本地的 LLM 应用转换为带有交互式文档的 RESTful API。
一、 核心回顾:从 FastAPI 到 LangServe
在上一章中,我们通过 FastAPI 手动构建了路由和 Pydantic 模型。虽然那能让我们深刻理解底层原理,但在实际开发中,LangChain 的组件(如 Chain、Agent)结构复杂,手动封装序列化逻辑是一项巨大的工程。
LangServe 的价值在于:
- 自动化:它基于 FastAPI 构建,自动处理了输入/输出的序列化与反序列化。
- 开箱即用:自动生成 Swagger UI 文档,原生支持流式输出(Streaming)。
- 高性能:继承了 FastAPI 的异步特性,支持高并发。
简而言之,LangServe = LangChain + FastAPI + 自动化魔法。
二、 环境准备
LangServe 的安装非常简单,它是一个独立的 Python 包。如果你已经安装了 LangChain 和 FastAPI,只需要补充安装 langserve 即可。
pip install langserve
前置要求:
- Python 3.8+
- 已安装 FastAPI 和 Uvicorn(上一篇已安装)
- 已安装 LangChain 相关组件(如 OpenAI 等)
三、 核心实战:三步部署你的第一个 AI 服务
让我们通过一个简单的例子,看看如何将一个 LLMChain 直接变成 Web API。
1. 编写 main.py
这一步你会感到非常熟悉,因为我们直接复用了 FastAPI 的 App,但不需要手动写路由逻辑。
from fastapi import FastAPI
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI # 请确保已安装 langchain_openai
from langchain.chains import LLMChain
from langserve import add_routes # 核心导入
# --- 1. 定义你的 LangChain 组件 ---
prompt = PromptTemplate.from_template("请用通俗的语言解释:{topic}")
llm = OpenAI(temperature=0.7)
chain = LLMChain(llm=llm, prompt=prompt)
# --- 2. 创建 FastAPI 应用 ---
app = FastAPI(
title="LangChain Server",
version="1.0",
description="使用 LangServe 部署的 AI 服务",
)
# --- 3. 关键一步:使用 add_routes 自动挂载 ---
# 这行代码会自动将 chain 挂载到 /rag 路径下,并生成所有端点
add_routes(app, chain, path="/rag")
# 启动命令:uvicorn main:app --reload
2. 运行服务
uvicorn main:app --reload
3. 零代码体验交互式文档
启动后,你不需要写任何 @app.get() 代码,直接访问 http://127.0.0.1:8000/docs。
你会发现,LangServe 已经根据你的 Chain 自动生成了 API 文档!你可以直接在网页上点击“Try it out”进行调试。
四、 深入理解:LangServe 自动创建的端点
LangServe 为每一个挂载的组件自动生成了两个核心端点,这与我们手动写 FastAPI 逻辑是一致的:
| 端点路径 | 功能 | 适用场景 |
|---|---|---|
/rag/invoke |
同步调用 | 获取完整的最终结果(阻塞直到生成结束) |
/rag/stream |
流式输出 | 获取生成的 Token 流,适合构建类似 ChatGPT 的逐字输出体验 |
💡 流式调用的魅力:
在前端开发中,直接调用 /stream 接口,你可以实现文字像打字机一样逐个出现的效果,而不需要等待漫长的思考时间。
五、 进阶用法:添加安全认证与自定义路由
虽然 LangServe 帮我们省去了繁琐的路由编写,但它并没有限制我们对 FastAPI 的控制权。我们可以像操作普通 FastAPI 应用一样,添加中间件和依赖。
1. 添加 Token 认证
为了防止你的 API 被滥用,可以轻松添加依赖注入:
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer
security = HTTPBearer()
def verify_token(credentials: HTTPBearer = Depends(security)):
if credentials.credentials != "MySecretToken123":
raise HTTPException(status_code=401, detail="Unauthorized")
return credentials.credentials
# 在 add_routes 中传入 dependencies 参数
add_routes(
app,
chain,
path="/secure-rag",
# 这里的依赖会应用到 /secure-rag 下的所有子路径
dependencies=[Depends(verify_token)]
)
2. 混合自定义路由
你依然可以在同一个文件中,既使用 LangServe 部署 AI 能力,又手动编写普通业务逻辑:
@app.get("/health")
async def health_check():
return {"status": "ok"}
六、 总结
通过这篇文章,我们完成了从 LangChain 逻辑代码 到 生产级 API 服务 的跨越。
LangServe 的核心价值在于:
- 极简开发:
add_routes一行代码解决了一切。 - 类型安全:利用 Pydantic 确保了数据的完整性。
- 无缝集成:它不是替代 FastAPI,而是 FastAPI 的强力插件。
对于大学生做毕设或大作业来说,LangServe 是完美的解决方案。它让你专注于核心的 AI 逻辑(Prompt Engineering、Chain 设计),而不用在后端繁琐的接口封装上浪费时间。
更多推荐



所有评论(0)