基于 LangChain 开发并部署简单大模型应用

引言

在人工智能快速发展的今天，构建基于大语言模型（LLM）的应用已成为开发者的核心技能之一。本教程将指导您使用 LangChain 框架构建一个简单的翻译应用程序。虽然该应用功能相对基础——仅涉及 LLM 调用与提示词处理，但它涵盖了 LangChain 的核心概念，是入门 LangChain 生态的理想起点。通过本指南，您将掌握从环境配置、模型集成、链式编排到服务化部署的完整流程。

环境安装

Python 环境准备

本指南假设您已安装 Python 3.8 或更高版本。建议使用虚拟环境隔离依赖，避免污染全局环境。

python -m venv langchain_env
source langchain_env/bin/activate  # Linux/Mac
# langchain_env\Scripts\activate   # Windows

安装 LangChain 核心库

pip install langchain langchain-core langchain-community

安装 LangSmith

LangSmith 是 LangChain 官方提供的可观测性平台，对于复杂应用的调试至关重要。注册后需设置环境变量以启用追踪。

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="your_api_key_here"

在 Jupyter Notebook 中可通过以下方式设置：

import getpass
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = getpass.getpass("Enter LangSmith API Key: ")

详细步骤

1. 模型集成

首先学习如何单独调用语言模型。LangChain 支持多种模型提供商，此处以 OpenAI 为例。

pip install langchain-openai

import getpass
import os
from langchain_openai import ChatOpenAI

# 设置 API Key
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OpenAI API Key: ")

# 初始化模型
model = ChatOpenAI(model="gpt-4", temperature=0)

ChatModel 实现了 Runnable 接口，可通过 .invoke 方法交互。

from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage(content="Translate the following from English into Italian"),
    HumanMessage(content="hi!")
]

response = model.invoke(messages)
print(response.content)

若启用了 LangSmith，可在控制台查看此次调用的详细追踪信息，包括 Token 消耗与延迟。

2. 输出解析器

模型返回的是 AIMessage 对象，包含内容及其他元数据。通常我们只需要字符串内容。

from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()

result = parser.invoke(response)
print(result)  # 输出：Ciao!

更高效的用法是将模型与解析器链接：

chain = model | parser
output = chain.invoke(messages)
print(output)

此链包含两个步骤：模型调用与结果解析。LangSmith 会自动记录这两个步骤的追踪。

3. 提示词模板

直接将消息列表传递给模型不够灵活。PromptTemplate 允许动态构建提示，适应不同场景。

from langchain_core.prompts import ChatPromptTemplate

system_template = "Translate the following into {language}:"
prompt_template = ChatPromptTemplate.from_messages(
    [("system", system_template), ("user", "{text}")]
)

# 测试模板
result = prompt_template.invoke({"language": "italian", "text": "hi"})
print(result.to_messages())

现在将模板、模型和解析器组合成完整链条：

chain = prompt_template | model | parser
output = chain.invoke({"language": "italian", "text": "hello world"})
print(output)

4. 高级提示工程

在实际应用中，单一指令可能无法满足精度要求。建议引入 Few-Shot Prompting（少样本提示）以提升效果。

few_shot_prompt = """Translate the following text to {language}.
Examples:
English: Hello -> {language}: Ciao
English: Good morning -> {language}: Buongiorno
Text: {text}"""

prompt_with_fewshot = ChatPromptTemplate.from_messages([
    ("system", few_shot_prompt),
    ("user", "{text}")
])

5. 错误处理

生产环境中必须考虑网络异常或模型超时。建议添加重试机制。

from langchain_core.runnables import RunnableWithRetry

retry_chain = RunnableWithRetry(chain, retries=3)
try:
    output = retry_chain.invoke({"language": "spanish", "text": "Goodbye"})
except Exception as e:
    print(f"Error occurred: {e}")

6. 服务化部署

使用 LangServe 将链封装为 FastAPI 服务。

安装依赖

pip install "langserve[all]" uvicorn

创建服务器

新建 serve.py 文件：

#!/usr/bin/env python
from typing import List
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
import os

# 1. 定义提示模板
system_template = "Translate the following into {language}:"
prompt_template = ChatPromptTemplate.from_messages([
    ("system", system_template),
    ("user", "{text}")
])

# 2. 初始化模型
model = ChatOpenAI()

# 3. 初始化解析器
parser = StrOutputParser()

# 4. 创建链
chain = prompt_template | model | parser

# 5. 定义 FastAPI 应用
app = FastAPI(
    title="LangChain Translation Server",
    version="1.0",
    description="A simple API server using LangChain's Runnable interfaces"
)

# 6. 添加路由
add_routes(app, chain, path="/chain")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

启动服务

python serve.py

服务将在 http://localhost:8000 运行。

7. 客户端调用与测试

Playground 界面

访问 http://localhost:8000/chain/playground 即可在浏览器中测试服务，支持流式输出与中间步骤查看。

程序化调用

使用 RemoteRunnable 进行远程调用：

from langserve import RemoteRunnable
remote_chain = RemoteRunnable("http://localhost:8000/chain/")
response = remote_chain.invoke({"language": "french", "text": "Hello"})
print(response)

Curl 测试

curl -X POST http://localhost:8000/chain/ \n-H "Content-Type: application/json" \n-d '{"input": {"language": "german", "text": "Hi"}}'

8. 安全最佳实践

在生产部署时，请注意以下事项：

API 密钥管理：不要硬编码密钥，使用环境变量或密钥管理服务。
HTTPS 加密：生产环境务必启用 HTTPS 保护数据传输。
限流策略：防止恶意请求耗尽资源，建议在网关层配置限流。
输入验证：对用户输入的长度和内容进行校验，防止注入攻击。

9. 常见问题排查

Token 限制：注意模型的上下文窗口限制，长文本需分段处理。
响应延迟：若响应过慢，可尝试降低温度参数或使用更快的模型。
依赖冲突：确保所有包版本兼容，必要时锁定 requirements.txt。

总结

本教程详细介绍了如何使用 LangChain 构建并部署一个翻译应用。我们学习了模型调用、提示词设计、链式编排、服务化部署及安全性考量。这些技能构成了大模型应用开发的基础。后续您可以进一步探索结构化输出、多模态处理及 Agent 代理等高级主题，逐步构建更复杂的智能系统。

基于 LangChain 开发并部署简单大模型应用