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

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. 安全最佳实践

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

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

9. 常见问题排查

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

总结

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