LangServe 部署与生产实践#

1
# pip install "langserve[all]" langchain-openai langchain-core
2

3
from fastapi import FastAPI
4
from langchain_openai import ChatOpenAI
5
from langchain_core.prompts import ChatPromptTemplate
6
from langchain_core.output_parsers import StrOutputParser
7
from langserve import add_routes
8

9
# 1. 构建 Chain
10
prompt = ChatPromptTemplate.from_messages([
11
    ("system", "你是一位专业翻译，请将用户输入的文本翻译成{target_language}。只输出翻译结果。"),
12
    ("human", "{text}"),
13
])
14
chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | StrOutputParser()
15

16
# 2. 创建 FastAPI 应用并一行发布
17
app = FastAPI(title="翻译服务", version="1.0")
18
add_routes(app, chain, path="/translate")
19

20
if __name__ == "__main__":
21
    import uvicorn
22
    uvicorn.run(app, host="0.0.0.0", port=8000)

1
{
2
  "output": "LangChain 是一个用于构建大语言模型应用的框架。",
3
  "metadata": { "run_id": "a1b2c3d4-..." }
4
}

1
event: data
2
data: {"content": "LangChain"}
3

4
event: data
5
data: {"content": " 是一个框架。"}
6

7
event: end

1
# pip install "langserve[client]"
2

3
from langserve import RemoteRunnable
4

5
translate = RemoteRunnable("http://localhost:8000/translate")
6

7
# 同步调用
8
result = translate.invoke({"text": "Hello world", "target_language": "中文"})
9

10
# 流式调用
11
for chunk in translate.stream({"text": "Hello", "target_language": "日语"}):
12
    print(chunk, end="", flush=True)
13

14
# 批量调用
15
results = translate.batch([
16
    {"text": "Hello", "target_language": "法语"},
17
    {"text": "Goodbye", "target_language": "法语"},
18
])
19

20
# 异步调用
21
result = await translate.ainvoke({"text": "Hi", "target_language": "中文"})

1
# pip install "langserve[all]"
2

3
from fastapi import FastAPI, Request, HTTPException
4
from fastapi.middleware.cors import CORSMiddleware
5
from starlette.middleware.base import BaseHTTPMiddleware
6

7
app = FastAPI()
8

9
# CORS —— 生产环境务必指定具体域名
10
app.add_middleware(
11
    CORSMiddleware,
12
    allow_origins=["https://your-frontend.com"],
13
    allow_credentials=True,
14
    allow_methods=["*"],
15
    allow_headers=["*"],
16
)
17

18
# API Key 认证
19
API_KEYS = {"sk-abc123", "sk-def456"}
20

21
class APIKeyMiddleware(BaseHTTPMiddleware):
22
    async def dispatch(self, request: Request, call_next):
23
        if request.url.path.endswith("/playground") or request.url.path == "/docs":
24
            return await call_next(request)
25
        if request.headers.get("X-API-Key") not in API_KEYS:
26
            raise HTTPException(status_code=403, detail="Invalid API Key")
27
        return await call_next(request)
28

29
app.add_middleware(APIKeyMiddleware)

1
# pip install "langserve[all]" langchain-openai langchain-core
2

3
from fastapi import FastAPI
4
from langserve import add_routes
5

6
app = FastAPI(title="AI 工具集")
7

8
# 自定义路由
9
@app.get("/health")
10
async def health_check():
11
    return {"status": "healthy"}
12

13
# 多个 Chain 挂载到不同路径
14
add_routes(app, translate_chain, path="/translate")
15
add_routes(app, summarize_chain, path="/summarize")
16
add_routes(app, explain_chain, path="/explain-code")

1
# pip install "langserve[all]"
2

3
from fastapi import Request
4
from langchain_core.runnables import RunnableConfig
5

6
def modify_config(config: RunnableConfig, request: Request) -> RunnableConfig:
7
    user_id = request.headers.get("X-User-Id", "anonymous")
8
    config["metadata"] = {**config.get("metadata", {}), "user_id": user_id}
9
    config["configurable"] = {**config.get("configurable", {}), "user_id": user_id}
10
    return config
11

12
add_routes(app, my_chain, path="/chat", per_req_config_modifier=modify_config)

1
# ============ 阶段 1：构建依赖 ============
2
FROM python:3.11-slim AS builder
3
WORKDIR /app
4
COPY requirements.txt .
5
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
6

7
# ============ 阶段 2：运行时 ============
8
FROM python:3.11-slim AS runtime
9
WORKDIR /app
10
COPY --from=builder /install /usr/local
11
COPY . .
12
RUN adduser --disabled-password --gecos '' appuser
13
USER appuser
14
EXPOSE 8000
15

16
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
17
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
18

19
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

1
version: "3.9"
2
services:
3
  langserve:
4
    build: .
5
    ports: ["8000:8000"]
6
    environment:
7
      - OPENAI_API_KEY=${OPENAI_API_KEY}
8
      - LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
9
      - LANGSMITH_TRACING=true
10
      - LANGSMITH_PROJECT=my-langserve-app
11
      - CHROMA_HOST=chroma
12
      - OLLAMA_BASE_URL=http://ollama:11434
13
    depends_on:
14
      chroma: { condition: service_healthy }
15
    restart: unless-stopped
16
    healthcheck:
17
      test: ["CMD", "python", "-c",
18
             "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
19
      interval: 30s
20
      timeout: 5s
21
      retries: 3
22

23
  chroma:
24
    image: chromadb/chroma:0.5.23
25
    ports: ["8001:8000"]
26
    volumes: [chroma_data:/chroma/chroma]
27
    healthcheck:
28
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/heartbeat"]
29
      interval: 10s
30
      timeout: 5s
31
      retries: 5
32

33
  ollama:
34
    image: ollama/ollama:latest
35
    ports: ["11434:11434"]
36
    volumes: [ollama_data:/root/.ollama]
37
    deploy:
38
      resources:
39
        reservations:
40
          devices: [{ driver: nvidia, count: all, capabilities: [gpu] }]
41

42
volumes:
43
  chroma_data:
44
  ollama_data:

1
# pip install langsmith
2
import os
3
os.environ["LANGSMITH_TRACING"] = "true"
4
os.environ["LANGSMITH_API_KEY"] = "lsv2_your-key-here"
5
os.environ["LANGSMITH_PROJECT"] = "production-translate-service"

1
# pip install "langserve[all]" langchain-openai
2

3
# Runnable 内置容错：with_retry + with_fallbacks
4
chain_with_retry = chain.with_retry(
5
    retry_if_exception_type=(TimeoutError, ConnectionError),
6
    stop_after_attempt=3,
7
    wait_exponential_jitter=True,
8
)
9

10
fallback_chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
11
chain_safe = chain_with_retry.with_fallbacks([fallback_chain])
12

13
add_routes(app, chain_safe, path="/translate")

1
# pip install python-json-logger
2
import logging, sys
3
from pythonjsonlogger import jsonlogger
4

5
handler = logging.StreamHandler(sys.stdout)
6
handler.setFormatter(jsonlogger.JsonFormatter(
7
    fmt="%(asctime)s %(name)s %(levelname)s %(message)s"
8
))
9
logging.getLogger().addHandler(handler)
10
logging.getLogger().setLevel(logging.INFO)

1
# pip install slowapi pydantic "langserve[all]"
2

3
# 速率限制示例
4
from slowapi import Limiter
5
from slowapi.util import get_remote_address
6
limiter = Limiter(key_func=get_remote_address)
7

8
# 输入验证示例
9
from pydantic import BaseModel, Field, field_validator
10
class TranslateInput(BaseModel):
11
    text: str = Field(..., min_length=1, max_length=10000)
12
    target_language: str
13

14
    @field_validator("target_language")
15
    @classmethod
16
    def check_lang(cls, v):
17
        allowed = {"中文", "英文", "日语", "法语", "德语", "韩语"}
18
        if v not in allowed:
19
            raise ValueError(f"不支持: {v}，可选: {allowed}")
20
        return v
21

22
add_routes(app, chain, path="/translate", input_type=TranslateInput)