第20章 LangServe：部署为 REST API
用LangChain开发完链（Chain）、RAG系统或Agent后，最关键的一步就是“落地部署”——让你的LLM应用能被前端、其他服务调用，真正发挥业务价值。而LangServe，正是LangChain官方推出的“一键部署神器”，专门解决LLM应用的服务化难题。
本章全程贴合掘金读者“拿来就用、实战为王”的需求，从LangServe基础概念入手，一步步教你将Chain转为API服务、生成接口文档、标准化请求响应，再到容器化部署、云原生集成，最后通过实战将RAG系统发布为可复用微服务，所有代码均简短可运行，标注官方及权威来源，避免冗余理论。
20.1 什么是 LangServe？
LangServe 是 LangChain 生态官方推出的服务化框架，核心作用是将 LangChain 构建的可运行组件（Chain、Agent、RAG 等）快速、标准化地部署为 REST API 服务，无需手动编写大量 FastAPI 路由、请求校验等冗余代码[superscript:2]。
简单说，你用LangChain写好的对话链、文档问答链，通过LangServe只需几行代码，就能变成可通过HTTP请求调用的API，适配前端集成、跨系统调用等生产级场景。
20.1.1 LangServe 核心价值（掘金实战视角）
对于开发者而言，LangServe 最实用的价值的是“省时间、标准化、易扩展”，解决传统部署的3大痛点：
痛点1：手动写API繁琐 → 解决方案：自动将Chain/Agent转为REST API，无需手动编写路由、请求解析代码；
痛点2：接口格式不统一 → 解决方案：内置标准化请求/响应格式，支持输入输出校验，避免跨团队对接麻烦；
痛点3：生产部署复杂 → 解决方案：支持异步、并发、容器化，无缝对接Docker、K8s，直接适配生产环境[superscript:4]。
20.1.2 LangServe 核心组件（必懂）
LangServe 基于 FastAPI 构建，继承了FastAPI的高性能和易用性，核心由4个组件组成，协同实现服务化部署[superscript:1]：
| 组件 | 核心作用 | 技术实现 |
|---|---|---|
| FastAPI Server | 提供REST API服务端点，处理HTTP请求 | FastAPI框架 |
| LCEL Integration | 自动将LCEL链转换为API端点，无需手动配置路由 | LangChain路由生成器 |
| Playground | 交互式API测试界面，方便调试接口 | Swagger UI定制 |
| Schema Validation | 输入输出格式校验，确保接口调用规范 | Pydantic模型 |
20.1.3 LangServe 与传统 FastAPI 部署对比
很多开发者会问：“我直接用FastAPI写接口不行吗？” 当然可以，但LangServe能帮你节省80%的重复编码，对比更直观：
| 对比维度 | 传统 FastAPI 部署 | LangServe 部署 |
|---|---|---|
| 代码量 | 需手动写路由、请求模型、响应解析（约50+行） | 只需3-5行代码，自动生成所有配置[superscript:4] |
| 接口文档 | 需手动编写API文档注释 | 自动生成OpenAPI文档和Swagger UI[superscript:6] |
| 扩展性 | 需手动扩展批量调用、流式输出等功能 | 内置invoke、batch、stream等端点，开箱即用[superscript:6] |
| LangChain 适配 | 需手动对接Chain、处理上下文 | 与LangChain组件无缝集成，支持所有Runnable对象[superscript:2] |
20.1.4 LangServe 部署流程（图例）
LangServe 的部署流程非常简洁，核心分为3步，全程无复杂配置：
flowchart TD
A[准备LangChain组件（Chain/RAG/Agent）] --> B[用LangServe的add_routes注册路由]
B --> C[启动FastAPI服务]
C --> D[自动生成API文档+测试界面]
D --> E[前端/其他服务调用API]
```
### 20\.1\.5 环境准备（必做）
先安装LangServe及相关依赖，支持Python≥3\.8，命令如下（代码来源：LangServe官方文档\[superscript:1\]）：
```bash
# 完整安装（包含服务端+客户端+Playground）
pip install "langserve[all]"
# 极简安装（仅服务端，适合生产环境）
pip install langserve fastapi uvicorn langchain-openai20.2 将 Chain 转换为 FastAPI 服务
这是LangServe最核心的功能——无论你是简单的对话链、复杂的RAG链，还是Agent，都能通过LangServe快速转为FastAPI服务，全程只需3步，代码简短可直接运行。
20.2.1 核心原理
LangServe 提供了 add\_routes 函数，该函数会自动解析LangChain的Chain（或其他Runnable对象），生成对应的FastAPI路由、请求/响应模型，无需手动编写任何路由逻辑[superscript:4]。
核心逻辑：Chain → add_routes（注册路由） → FastAPI服务 → 可调用API。
20.2.2 实战示例1：简单对话链转为API
先创建一个简单的LangChain对话链，再通过LangServe转为API，代码简短，可直接运行（代码来源：LangServe官方示例[superscript:2]）：
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 1. 准备LangChain对话链（核心业务逻辑）
prompt = ChatPromptTemplate.from_template("你是友好的助手，回答用户问题：{input}")
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API Key", temperature=0.7)
chain = prompt | llm # LCEL链式调用
# 2. 创建FastAPI应用
app = FastAPI(title="LangServe Demo", version="1.0")
# 3. 用LangServe注册路由（核心步骤）
add_routes(
app,
chain,
path="/chat" # API端点路径，调用时用http://localhost:8000/chat/invoke
)
# 启动服务
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)20.2.3 启动与测试服务
启动服务：运行上述代码，控制台会提示服务启动成功（默认端口8000）；
测试接口：有两种方式，适合不同场景：
curl \-X POST http://localhost:8000/chat/invoke \\ \-H \&\#34;Content\-Type: application/json\&\#34; \\ \-d \&\#39;\{\&\#34;input\&\#34;: \&\#34;LangServe是什么？\&\#34;\}\&\#39;方式1：用curl命令调用（适合后端测试）：
方式2：访问交互式界面（适合调试）：打开浏览器访问
http://localhost:8000/chat/playground，输入问题即可测试。
20.2.4 实战示例2：多链注册（多API端点）
实际开发中，可能需要部署多个Chain（如对话链、翻译链），LangServe支持一次性注册多个路由，代码如下（代码来源：LangServe实战示例[superscript:5]）：
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
app = FastAPI(title="多Chain API服务", version="1.0")
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API Key")
# 1. 对话链
chat_prompt = ChatPromptTemplate.from_template("回答用户问题：{input}")
chat_chain = chat_prompt | llm
# 2. 翻译链
translate_prompt = ChatPromptTemplate.from_template("将{text}翻译成英文")
translate_chain = translate_prompt | llm
# 注册多个路由（不同端点）
add_routes(app, chat_chain, path="/chat")
add_routes(app, translate_chain, path="/translate")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)启动后，将拥有两个API端点：/chat/invoke（对话）、/translate/invoke（翻译），各自独立运行，互不影响。
20.2.5 关键注意点
Chain必须是LangChain的Runnable对象（如LCEL链、ConversationChain等），否则无法注册；
path参数需唯一，避免多个Chain注册到同一端点，导致冲突；
API Key建议用环境变量存储（如
os\.getenv\(\&\#34;OPENAI\_API\_KEY\&\#34;\)），避免硬编码，提升安全性[superscript:1]。
20.3 自动生成 OpenAPI 文档与 Swagger UI
LangServe 最实用的特性之一——无需手动编写API文档，会自动根据Chain的输入输出，生成标准化的OpenAPI文档和Swagger UI，方便团队协作、接口调试和前端对接[superscript:6]。
对于掘金读者而言，这意味着“写完代码，文档自动生成”，无需额外花时间维护接口文档。
20.3.1 如何访问自动生成的文档
启动LangServe服务后，只需访问两个地址，即可获取完整的API文档：
Swagger UI（交互式调试界面）：
http://localhost:8000/docs（最常用，可直接测试接口）；OpenAPI规范（JSON格式）：
http://localhost:8000/openapi\.json（可导入Postman、Apifox等工具）[superscript:1]。
20.3.2 Swagger UI 核心功能（实战演示）
访问 http://localhost:8000/docs 后，会看到自动生成的接口文档，核心功能如下：
查看所有API端点：清晰展示已注册的所有Chain对应的接口（如/chat/invoke、/translate/invoke）；
查看请求/响应格式：自动显示输入参数、输出参数的JSON Schema，无需手动说明；
在线调试接口：点击“Try it out”，输入参数，点击“Execute”即可查看接口响应，无需借助其他工具；
导出接口文档：支持导出JSON、YAML格式的OpenAPI规范，方便前端对接[superscript:4]。
20.3.3 自定义文档信息（优化体验）
默认的文档信息（标题、描述、版本）比较简单，可通过FastAPI的参数自定义，让文档更贴合业务，代码示例：
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 自定义文档信息（核心）
app = FastAPI(
title="LangServe 电商客服API",
version="1.0.0",
description="基于LangServe部署的电商客服对话API，支持订单查询、物流咨询等功能",
terms_of_service="http://example.com/terms/",
contact={"name": "开发者", "email": "dev@example.com"}
)
prompt = ChatPromptTemplate.from_template("电商客服：{input}")
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API Key")
chain = prompt | llm
add_routes(app, chain, path="/customer-service")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)重启服务后，访问 http://localhost:8000/docs，会看到自定义的文档信息，更专业、更易理解。
20.3.4 文档生成原理
LangServe 会自动解析Chain的输入输出类型，通过Pydantic模型生成JSON Schema，再基于FastAPI的OpenAPI生成能力，自动构建文档[superscript:6]。
例如：Chain的输入是字符串，文档会自动生成“input: string”的请求格式；输出是ChatMessage，文档会自动展示输出的结构（content、role等），无需手动配置。
20.4 请求/响应格式标准化
生产环境中，API的请求/响应格式必须标准化，否则会导致前端对接混乱、跨系统调用失败。LangServe 内置了标准化的请求/响应格式，无需手动定义，同时支持自定义格式，适配不同业务需求[superscript:6]。
20.4.1 默认请求/响应格式（核心）
LangServe 为所有Chain默认提供3种请求方式，每种方式对应标准化的格式，最常用的是 invoke（单次调用）：
1. invoke（单次调用，最常用）
适用于单次请求、获取单次响应的场景（如用户发送一条消息，获取机器人回复），格式如下：
请求方式：POST
请求地址：
http://localhost:8000/\[path\]/invoke（path是注册时的路径，如/chat）；请求体（JSON）：
{
"input": "你的请求内容", // 对应Chain的输入参数
"config": {} // 可选，用于配置LLM参数（如temperature）
}- 响应体（JSON）：
{
"output": "Chain的输出结果", // 对应Chain的输出
"metadata": {} // 可选，包含请求ID、执行时间等元数据
}2. batch（批量调用）
适用于批量处理多个请求（如批量翻译、批量生成摘要），请求地址：http://localhost:8000/\[path\]/batch，请求体格式：
{
"inputs": ["请求1", "请求2", "请求3"], // 批量输入，数组格式
"config": {}
}3. stream（流式输出）
适用于长文本生成（如聊天机器人、长文档摘要），支持实时返回结果，请求地址：http://localhost:8000/\[path\]/stream，请求体与invoke一致，响应为流式数据[superscript:6]。
20.4.2 自定义请求/响应格式（实战）
默认格式适用于大多数场景，若业务需要自定义格式（如添加额外参数、修改输出结构），可通过Pydantic模型实现，代码示例（代码来源：LangServe高级示例[superscript:1]）：
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel
# 1. 自定义请求模型（添加额外参数）
class CustomInput(BaseModel):
question: str # 用户问题
user_id: str # 额外参数：用户ID
style: str = "简洁" # 额外参数：回答风格（默认简洁）
# 2. 自定义输出模型
class CustomOutput(BaseModel):
answer: str # 回答内容
user_id: str # 回显用户ID
length: int # 回答长度
# 3. 构建Chain，适配自定义格式
prompt = ChatPromptTemplate.from_template("以{style}风格回答用户{user_id}的问题：{question}")
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API Key")
# 处理输入：将CustomInput转换为Chain需要的格式
def format_input(input_data: CustomInput):
return {
"question": input_data.question,
"user_id": input_data.user_id,
"style": input_data.style
}
# 处理输出：将Chain输出转换为CustomOutput格式
def format_output(output: str, input_data: CustomInput):
return CustomOutput(
answer=output,
user_id=input_data.user_id,
length=len(output)
)
# 构建完整Chain
chain = (format_input | prompt | llm | (lambda x: x.content))
# 4. 注册路由，指定自定义输入输出模型
app = FastAPI(title="自定义格式API")
add_routes(
app,
chain.with_types(input_type=CustomInput, output_type=CustomOutput),
path="/custom-chat"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)启动服务后，访问Swagger UI，会看到请求/响应格式已变为自定义的结构，完美适配业务需求。
20.4.3 格式校验（避免非法请求）
LangServe 基于Pydantic模型实现自动格式校验，当请求格式不符合要求时，会返回清晰的错误信息，无需手动编写校验代码[superscript:6]。
示例：若自定义请求模型要求user\_id为字符串，而请求时传入数字，会返回如下错误：
{
"detail": [
{
"loc": ["body", "user_id"],
"msg": "value is not a valid string",
"type": "type_error.string"
}
]
}20.5 并发与异步支持
生产环境中，API需要支持高并发请求（如同时有上百个用户调用客服接口），LangServe 基于FastAPI和Starlette，天生支持异步和高并发，无需额外配置，只需简单调整代码，即可应对高并发场景[superscript:1]。
20.5.1 异步Chain与异步服务
LangChain 提供了异步版本的Chain（如AsyncChatOpenAI、异步RAG链），LangServe 可直接适配，实现异步处理请求，提升并发能力，代码示例（代码来源：LangServe异步示例[superscript:4]）：
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import AsyncChatOpenAI # 异步LLM
from langchain_core.prompts import ChatPromptTemplate
app = FastAPI(title="异步LangServe服务")
# 1. 构建异步Chain
prompt = ChatPromptTemplate.from_template("回答用户问题：{input}")
async_llm = AsyncChatOpenAI(model="gpt-3.5-turbo", api_key="你的API Key")
async_chain = prompt | async_llm # 异步链式调用
# 2. 注册路由（自动适配异步）
add_routes(app, async_chain, path="/async-chat")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)关键说明：使用AsyncChatOpenAI（异步LLM），Chain会自动变为异步，LangServe 会自动生成异步API端点，支持高并发请求。
20.5.2 并发配置优化（生产级）
默认启动方式适合开发调试，生产环境中需优化uvicorn的并发配置，提升并发处理能力，启动命令如下（代码来源：LangServe生产部署指南[superscript:5]）：
# 生产环境启动命令（优化并发）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop参数说明：
--workers 4：启动4个工作进程，建议设置为服务器CPU核心数的2倍；
--loop uvloop：使用uvloop事件循环，提升异步性能，比默认loop快30%以上。
20.5.3 并发测试（验证性能）
可使用locust工具测试并发能力，步骤如下：
安装locust：
pip install locust；创建测试文件（locustfile.py）：
from locust import HttpUser, task, between
class LangServeUser(HttpUser):
wait_time = between(1, 3) # 每个用户请求间隔1-3秒
@task
def invoke_chat(self):
self.client.post(
"/async-chat/invoke",
json={"input": "LangServe并发性能如何？"}
)启动测试：
locust \-f locustfile\.py \-\-host=http://localhost:8000；访问
http://localhost:8089，设置并发用户数和每秒请求数，即可查看并发测试结果。
测试结果说明：在4核8G服务器上，配置--workers 4，LangServe可轻松支持每秒100+请求，完全满足生产级并发需求。
20.6 Docker 容器化部署
生产环境中，容器化部署是标配——可确保开发、测试、生产环境一致，避免“在我电脑上能运行，部署到服务器就报错”的问题。LangServe 支持无缝Docker容器化，只需编写2个文件，即可完成打包部署[superscript:3]。
20.6.1 容器化部署流程（图例）
flowchart TD
A[编写LangServe代码（main.py）] --> B[编写Dockerfile]
B --> C[编写requirements.txt]
C --> D[构建Docker镜像]
D --> E[启动Docker容器]
E --> F[访问API服务]
```
### 20\.6\.2 实战：Dockerfile 编写
假设我们有一个简单的LangServe服务（main\.py），编写Dockerfile，将服务打包为镜像，代码如下（代码来源：LangServe容器化示例\[superscript:3\]）：
```dockerfile
# 基础镜像（Python 3.10，轻量且稳定）
FROM python:3.10-slim
# 设置工作目录
WORKDIR /app
# 复制依赖文件
COPY requirements.txt .
# 安装依赖（国内源优化，加快安装速度）
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
# 复制应用代码
COPY main.py .
# 暴露服务端口（与代码中uvicorn端口一致）
EXPOSE 8000
# 启动命令（生产级配置）
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4", "--loop", "uvloop"]20.6.3 requirements.txt 编写
列出所有依赖包，确保版本兼容，内容如下：
langserve>=0.0.19
fastapi>=0.95.0
uvicorn>=0.23.2
langchain-openai>=0.1.0
python-dotenv>=1.0.0
uvloop>=0.19.020.6.4 构建与启动Docker容器
在代码目录（main.py、Dockerfile、requirements.txt同目录）下，执行以下命令，完成构建与启动：
# 1. 构建Docker镜像（镜像名：langserve-app，版本：v1）
docker build -t langserve-app:v1 .
# 2. 启动Docker容器（端口映射：宿主机8000 → 容器8000，后台运行）
docker run -d -p 8000:8000 --name langserve-container langserve-app:v1
# 3. 查看容器运行状态
docker ps
# 4. 查看容器日志（排查问题）
docker logs -f langserve-container启动成功后，访问 http://服务器IP:8000/docs，即可正常使用API服务。
20.6.5 容器化注意事项
环境变量：API Key等敏感信息，建议通过Docker环境变量传入，避免硬编码（如
docker run \-e OPENAI\_API\_KEY=你的密钥 \.\.\.）；镜像优化：使用
python:3\.10\-slim轻量镜像，减少镜像体积；使用\-\-no\-cache\-dir避免缓存依赖，进一步减小体积；端口映射：确保宿主机端口未被占用，若需修改端口，可调整
\-p 新端口:8000；持久化：若服务需要存储数据（如向量库），需配置Docker数据卷，避免容器删除后数据丢失[superscript:3]。
20.7 与 Nginx、Kubernetes 集成
Docker容器化适合单节点部署，若需要高可用、负载均衡、自动扩缩容，需集成Nginx和Kubernetes（K8s）——Nginx作为反向代理，处理请求分发；K8s负责容器编排，实现高可用部署[superscript:7]。
本节讲解生产级集成方案，贴合企业实际部署需求，步骤清晰可落地。
20.7.1 与 Nginx 集成（反向代理）
Nginx 主要作用：反向代理、负载均衡、SSL终止（HTTPS）、静态资源服务，集成步骤如下：
1. 准备Nginx配置文件（nginx.conf）
http {
upstream langserve_servers {
# 多个LangServe容器实例（负载均衡）
server 172.17.0.2:8000; # 容器1IP:端口
server 172.17.0.3:8000; # 容器2IP:端口
# 可新增更多容器实例，实现负载均衡
}
server {
listen 80;
server_name api.example.com; # 你的域名
# 反向代理到LangServe服务
location / {
proxy_pass http://langserve_servers;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# 静态资源缓存（如Swagger UI静态文件）
location /static/ {
alias /usr/share/nginx/html/static/;
expires 1d;
}
}
}
events {
worker_connections 1024;
}配置说明：upstream 中配置多个LangServe容器实例，实现负载均衡；proxy\_pass 将请求转发到LangServe服务[superscript:7]。
2. 启动Nginx容器（与LangServe容器联动）
# 1. 启动2个LangServe容器（负载均衡示例）
docker run -d --name langserve-1 -p 8001:8000 langserve-app:v1
docker run -d --name langserve-2 -p 8002:8000 langserve-app:v1
# 2. 启动Nginx容器（挂载配置文件）
docker run -d -p 80:80 --name nginx-langserve \
-v /root/nginx.conf:/etc/nginx/nginx.conf \
--link langserve-1:langserve-1 \
--link langserve-2:langserve-2 \
nginx:alpine启动后，访问 http://api\.example\.com/docs，Nginx会自动将请求分发到两个LangServe容器，实现负载均衡。
20.7.2 与 Kubernetes 集成（容器编排）
Kubernetes 用于实现LangServe服务的高可用、自动扩缩容、故障自愈，适合大规模部署，核心步骤如下（代码来源：K8s与LangServe集成示例[superscript:7]）：
1. 编写Deployment配置（langserve-deployment.yaml）
用于部署LangServe容器，实现多副本、自动重启：
apiVersion: apps/v1
kind: Deployment
metadata:
name: langserve-deployment
namespace: llm-app # 自定义命名空间
spec:
replicas: 3 # 3个副本，实现高可用
selector:
matchLabels:
app: langserve
template:
metadata:
labels:
app: langserve
spec:
containers:
- name: langserve
image: langserve-app:v1 # 之前构建的Docker镜像
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: llm-secrets
key: openai-api-key # 从K8s Secret中获取API Key（安全）
resources:
requests:
cpu: "100m"
memory: "256Mi"
limits:
cpu: "500m"
memory: "512Mi"
livenessProbe: # 健康检查，故障自动重启
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 102. 编写Service配置（langserve-service.yaml）
用于暴露LangServe服务，供Nginx或其他服务访问：
apiVersion: v1
kind: Service
metadata:
name: langserve-service
namespace: llm-app
spec:
selector:
app: langserve
ports:
- port: 80
targetPort: 8000
type: ClusterIP # 集群内部访问，配合Nginx Ingress对外暴露3. 编写Ingress配置（langserve-ingress.yaml）
用于对外暴露服务，配合Nginx Ingress Controller实现反向代理和负载均衡[superscript:7]：
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: langserve-ingress
namespace: llm-app
annotations:
nginx.ingress.kubernetes.io/rewrite-target: /
nginx.ingress.kubernetes.io/proxy-body-size: "50m"
spec:
ingressClassName: nginx
rules:
- host: api.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: langserve-service
port:
number: 804. 部署到K8s集群
# 1. 创建命名空间
kubectl create namespace llm-app
# 2. 创建Secret（存储API Key，安全保密）
kubectl create secret generic llm-secrets -n llm-app --from-literal=openai-api-key=你的API Key
# 3. 部署Deployment、Service、Ingress
kubectl apply -f langserve-deployment.yaml -n llm-app
kubectl apply -f langserve-service.yaml -n llm-app
kubectl apply -f langserve-ingress.yaml -n llm-app
# 4. 查看部署状态
kubectl get pods -n llm-app
kubectl get svc -n llm-app
kubectl get ingress -n llm-app部署成功后，访问 http://api\.example\.com/docs，即可通过K8s集群访问LangServe服务，实现高可用、自动扩缩容。
20.7.3 集成核心优势
负载均衡：Nginx+K8s实现请求分发，避免单节点压力过大；
高可用：K8s自动检测容器故障，重启故障容器，确保服务不中断；
自动扩缩容：根据CPU、内存使用率，自动增加或减少容器副本，适配流量波动；
安全可靠：API Key通过K8s Secret存储，避免泄露；Nginx实现SSL终止，支持HTTPS[superscript:7]。
20.8 【实战】将 RAG 系统发布为微服务
结合本章所学知识点，实战将一个完整的RAG（文档问答）系统，通过LangServe部署为微服务，实现“上传文档→检索问答→API调用”的完整流程，支持Docker容器化，可直接对接前端或其他服务，代码可复用、步骤清晰。
20.8.1 实战需求与技术栈
核心需求
基础功能：构建RAG系统，支持加载本地文档，用户通过API提问，返回基于文档的精准回答；
服务化：通过LangServe部署为REST API，支持invoke（单次查询）、stream（流式输出）；
容器化：支持Docker打包部署，可直接运行在服务器；
可扩展：支持新增文档，自动更新向量库，适配企业知识库场景。
技术栈
核心框架：LangChain、LangServe、FastAPI；
RAG组件：Chroma（向量库）、OpenAIEmbeddings（嵌入模型）、ChatOpenAI（LLM）；
部署工具：Docker、uvicorn；
依赖包：langchain、langserve、chromadb、langchain-openai、fastapi、uvicorn。
20.8.2 完整代码实现（分模块）
代码分为3个模块：rag_chain.py（RAG核心逻辑）、main.py（LangServe服务）、requirements.txt（依赖），结构清晰，便于维护。
1. RAG核心逻辑（rag_chain.py）
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os
# 初始化LLM和嵌入模型
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key=os.getenv("OPENAI_API_KEY"), temperature=0.3)
embeddings = OpenAIEmbeddings(api_key=os.getenv("OPENAI_API_KEY"))
# 构建RAG链（加载文档→分割→向量库→检索问答链）
def create_rag_chain():
# 1. 加载本地文档（可替换为PDF、Word等文档）
loader = TextLoader("knowledge_base.txt") # 企业知识库文档
documents = loader.load()
# 2. 分割文档（避免单段过长）
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = text_splitter.split_documents(documents)
# 3. 构建向量库（持久化存储，避免每次重启重新加载）
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db"
)
vectorstore.persist()
# 4. 构建检索问答链
rag_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(k=2), # 检索前2条相关片段
return_source_documents=True # 返回检索到的文档片段，便于调试
)
return rag_chain2. LangServe服务（main.py）
from fastapi import FastAPI
from langserve import add_routes
from rag_chain import create_rag_chain
from pydantic import BaseModel
from dotenv import load_dotenv
# 加载环境变量（API Key从.env文件读取，避免硬编码）
load_dotenv()
# 创建FastAPI应用（自定义文档信息）
app = FastAPI(
title="RAG文档问答微服务",
version="1.0.0",
description="基于LangServe部署的RAG文档问答API，支持企业知识库查询",
contact={"name": "开发者", "email": "dev@example.com"}
)
# 初始化RAG链
rag_chain = create_rag_chain()
# 自定义请求/响应模型（适配RAG场景）
class RAGInput(BaseModel):
question: str # 用户查询问题
user_id: str # 用户ID（用于跟踪查询记录）
class RAGOutput(BaseModel):
answer: str # 回答内容
source_documents: list # 检索到的相关文档片段
user_id: str # 回显用户ID
query_time: str = "即时" # 查询时间（简化版，实际可添加时间戳）
# 处理输出：将RAG链输出转换为自定义格式
def format_rag_output(output, input_data: RAGInput):
return RAGOutput(
answer=output["result"],
source_documents=[doc.page_content for doc in output["source_documents"]],
user_id=input_data.user_id
)
# 构建完整Chain（适配自定义输入输出）
custom_rag_chain = (
lambda x: rag_chain.invoke({"query": x.question})
| (lambda output, x=x: format_rag_output(output, x))
)
# 注册路由（核心步骤）
add_routes(
app,
custom_rag_chain.with_types(input_type=RAGInput, output_type=RAGOutput),
path="/rag-qa",
enable_feedback_endpoint=True # 启用反馈端点，便于收集用户反馈
)
# 启动服务
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000, workers=4, loop="uvloop")3. 依赖文件（requirements.txt）
langserve>=0.0.19
fastapi>=0.95.0
uvicorn>=0.23.2
langchain>=0.1.0
langchain-openai>=0.1.0
chromadb>=0.4.20
python-dotenv>=1.0.0
uvloop>=0.19.0
pydantic>=2.0.04. 环境变量文件（.env）
OPENAI_API_KEY=你的API Key5. 知识库文档（knowledge_base.txt）
创建简单的企业知识库文档（示例），用于测试RAG功能：
企业名称：XX科技有限公司
成立时间：2020年10月
核心业务：人工智能、大模型应用开发、LangChain生态部署
产品服务