LangChain 生产级实战（七）：部署与监控 —— 从 Demo 到产品

1. 痛点：Jupyter Notebook 不是产品

我们在前几篇写的代码，虽然逻辑通顺，但都有一个共同的问题：它们很难被集成。

如果你想把 Agent 给前端 React 开发，或者集成到微信机器人里，你需要：

写一个 FastAPI 服务。
手动定义 Request 和 Response 的 JSON 结构（Pydantic Models）。
最头疼的是：实现流式输出 (Streaming)。手动用 FastAPI 写 Server-Sent Events (SSE) 来推送 LLM 的一个个 Token，是一件非常繁琐且容易出错的事情。

LangServe 完美解决了这个问题。它能自动把任何 LangChain 对象（Chain, Agent, Runnable）转换成标准的 FastAPI 接口。

2. 环境准备

LangServe 是一个独立的包，基于 FastAPI。

# 安装 LangServe (包含服务端和客户端)
pip install "langserve[all]"

# 安装 web 服务器
pip install fastapi uvicorn sse_starlette

3. 实战一：三行代码部署 API

假设我们要部署第一篇博客里的“冷笑话生成器”。

我们需要创建一个新的 Python 文件，命名为 server.py。

服务端代码 (`server.py`)

#!/usr/bin/env python
from fastapi import FastAPI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langserve import add_routes # <--- 核心主角

# 1. 定义我们的 Chain (和之前一模一样)
model = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("请给我讲一个关于 {topic} 的冷笑话。")
chain = prompt | model | StrOutputParser()

# 2. 初始化 FastAPI 应用
app = FastAPI(
    title="我的冷笑话 API",
    version="1.0",
    description="这是一个基于 LangChain 的智能笑话生成服务",
)

# 3. 添加路由 (Magic happens here)
# add_routes 会自动分析 chain 的输入输出结构
# 并生成 /invoke, /stream, /batch 等标准端点
add_routes(
    app,
    chain,
    path="/joke",
)

if __name__ == "__main__":
    import uvicorn
    # 启动服务，运行在 localhost:8000
    uvicorn.run(app, host="localhost", port=8000)

启动与体验

在终端运行：

1	python server.py

现在，神奇的事情发生了。打开浏览器访问 http://localhost:8000/joke/playground/。

你会看到一个自动生成的 UI 界面：

左边有输入框，让你输入 topic。
点击 "Start"，右边会实时流式显示生成的笑话。
它甚至显示了中间的 JSON 数据结构。

这就是 LangServe 的威力：零前端代码，直接获得调试面板。

4. 实战二：客户端调用 (`client.py`)

现在服务跑起来了，我们怎么在另一个 Python 脚本里调用它？不用 requests 手写 HTTP 请求，直接用 RemoteRunnable。

from langserve import RemoteRunnable

# 指向刚才定义的 path
remote_chain = RemoteRunnable("http://localhost:8000/joke/")

# 就像调用本地 chain 一样调用它！
print("--- 同步调用 ---")
result = remote_chain.invoke({"topic": "程序员"})
print(result)

print("\n--- 流式调用 ---")
for chunk in remote_chain.stream({"topic": "猫"}):
    print(chunk, end="|", flush=True)

RemoteRunnable 让远程 API 看起来就像在本地运行一样，无缝支持 invoke, stream, batch。

5. 实战三：开启上帝视角 (LangSmith)

当你的 API 上线后，用户反馈：“昨天那个机器人对我说了一句很奇怪的话。” 你问：“它说了什么？当时你问了什么？” 用户：“我忘了。”

这时候你需要 LangSmith。它是 LangChain 官方的监控平台。它能记录：

每一次 API 调用的完整链路。
每个步骤消耗的时间（Latency）。
消耗的 Token 数量和金钱成本。
具体的输入和输出。

开启方式：无需改代码

你只需要在运行 server.py 的终端里，设置环境变量即可。（需要先在 smith.langchain.com 注册账号并获取 API Key）

# Linux / Mac
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_ENDPOINT="[https://api.smith.langchain.com](https://api.smith.langchain.com)"
export LANGCHAIN_API_KEY="你的_langsmith_api_key"
export LANGCHAIN_PROJECT="My_Joke_App_Prod"

# Windows Powershell
# $env:LANGCHAIN_TRACING_V2="true" ...

设置完环境变量后，再次运行 python server.py 并调用几次接口。

你会看到什么？

Run: Joke Generator (2.5s)
├── ChatPromptTemplate (0.01s)
│   └── Input: {"topic": "程序员"}
│   └── Output: [SystemMessage(...), HumanMessage(...)]
├── ChatOpenAI (2.4s)
│   └── Input: [Messages...]
│   └── Output: AIMessage(content="为什么程序员总是分不清万圣节和圣诞节？因为 Oct 31 == Dec 25")
└── StrOutputParser (0.01s)
    └── Output: "为什么程序员..."

如果中间某一步报错了（比如 OpenAI API 超时），你会在这里看到红色的 Error 标记，并且能看到完整的堆栈信息。

6. 进阶：LangSmith 的数据集与测试

LangSmith 不仅仅是监控，它还是测试平台。

你可以把用户历史记录中那些“回答得很好”或“回答得很差”的例子，一键添加到 Dataset 中。

然后，当你修改了 Prompt 或者换了模型（比如从 GPT-3.5 换到 GPT-4）后，可以在 Dataset 上跑自动化测试，对比新旧版本的表现。

# 简单的评估代码示例 (伪代码)
from langsmith import Client

client = Client()
# 运行评估
results = client.run_on_dataset(
    dataset_name="joke-dataset",
    llm_or_chain_factory=my_new_chain,
    evaluation=evaluators # 比如检查是否包含冒犯性词汇
)

7. 总结

在这一篇中，我们补全了 DevOps 的拼图。

LangServe：解决了“怎么卖”的问题。它让你的 AI 逻辑变成标准的 HTTP API，能够被任何编程语言、任何前端框架调用。
LangSmith：解决了“怎么修”的问题。它提供了生产环境必须的可观测性（Observability）。

至此，你已经掌握了从写第一行代码到上线运维的全套流程。

在下一篇（第八篇），我们将讨论一个让很多开发者头疼的高级话题：如何让 LLM 稳定地输出 JSON 格式，以及基于语义的高级路由策略。