lang: zh slug: langchain title: ‘LangChain：部署生产就绪型人工智能代理的 3 种方法’ description: ‘LangChain (LC) 是一个 Python/JS 框架，用于构建由 LLM 驱动的应用程序，具有 700 多个集成。了解如何安装 LangChain、使用 Docker 进行部署、与 OpenAI、Anthropic、Ollama 集成，以及如何使用 LangSmith 可观测性、LangGraph 代理和 Kubernetes 扩展到生产环境。’ tags: [“ai-agent”, “automation”, “guide”, “open-source”, “reference”, “tutorial”] date: 2026-05-19 00:00:00+08:00 lastmod: 2026-05-19 00:00:00+08:00 tech_stack: [] application_domain: Llm Frameworks source_version: ’' licensing_model: Open Source license_type: MIT file_size: ’' file_md5: ’' download_url: ’' backup_url: ’' github_repo: ‘https://github.com/langchain-ai/langchain' last_maintained: ‘2026-05-19’ draft: false categories: [’llm-frameworks’] aliases:- /帖子/langchain/

/resources/llm-frameworks/langchain-complete-guide/
/posts/langchain-complete-guide/ 常见问题解答：
q: ‘LangChain 和 LangGraph 有什么区别？’ a：“LangChain 是用于构建具有链、提示和模型集成的 LLM 应用程序的核心框架，而 LangGraph 是一个扩展库，它为复杂的代理工作流程添加了基于图形的编排，包括循环、分支和人机交互审批。将 LangChain 视为组件库，将 LangGraph 视为工作流引擎。两者均由LangChain Inc.维护并具有相同的发布周期。
q: ‘LangChain可以免费用于商业用途吗？ ‘是的。 LangChain 获得 MIT 许可，免费供商业和个人使用，包括核心框架、LangGraph 和所有社区集成。 LangSmith observability 每月提供 5,000 条跟踪的免费套餐，付费计划起价为每位用户每月 39 美元，LLM API 费用由提供商单独计费。
q: ‘如何在LangChain中切换LLM提供商？’ a: ‘更改模型类导入； LangChain 的标准化 BaseChatModel 接口意味着为 OpenAI 编写的代码可以与 Anthropic、Google、Ollama 或任何受支持的提供商一起工作，只需进行最小的更改。 1.0+ 版本中的 .content_blocks 属性标准化了所有提供程序的消息格式，消除了特定于提供程序的解析代码。
q: ‘LangChain需要什么Python版本？’ a: ‘LangChain 需要 Python 3.10 或更高版本（自版本 1.0 起）。它可以在 60 秒内通过 pip 安装，并且可以作为 Python 和 TypeScript 框架使用。
q：“我可以自行托管 LangSmith 以实现可观察性吗？” a：“自托管 LangSmith 仅适用于具有自定义定价的企业计划。对于需要本地可观察性的团队，开源替代方案包括 Langfuse（MIT 许可证）、Arize 的 Phoenix 和 Helicone，它们通过 OpenTelemetry 或直接回调与 LangChain 集成。

Retrieval — Document loaders, text splitters, embedding models, and vector stores form the RAG pipeline. Load PDFs, HTML, or Notion pages, chunk them, embed, and query semantically.
Agents — The create_agent API (LangChain 1.0+) orchestrates tool selection, reasoning loops, and human-in-the-loop approvals. Agents decide which tools to call, in what order, and when to stop.
Chains — Composable workflows that link components sequentially. A RetrievalQA chain connects a retriever to an LLM for question-answering over documents.
Observability — LangSmith traces every call, measuring latency, token usage, and cost. Traces capture inputs, outputs, and intermediate steps for debugging.用户查询 → 代理/链 → [工具调用 → LLM 调用 → 检索] → 响应 ↓ LangSmith（跟踪、指标、评估）

### 核心概念Runnable接口。 LangChain中的每个组件都通过“.invoke()”、“.batch()”和“.stream()”方法实现了“Runnable”协议。这种统一的界面使您能够以相同的方式处理单个提示、十个组件链或多代理图。内容块。 LangChain 1.0 在消息上引入了“.content_blocks”——一种跨所有提供商的文本、图像、工具调用和推理跟踪的统一格式。不再需要特定于提供者的消息解析。模型配置文件。 聊天模型通过“.profile”属性公开功能，从而实现动态功能检测。您的代码可以在尝试之前检查模型是否支持工具调用或视觉。## 安装和设置### 基本安装LangChain 通过 pip 在 60 秒内完成安装。从版本 1.0 开始，需要 Python 3.10+。```` bas h

安装核心框架 #

pip install langchain-core==1.4.0 langchain# 安装 OpenAI 集成 pip 安装 langchain-openai# 安装``` bas h

安装核心框架 #

pip install langchain-core==1.4.0 langchain

安装 OpenAI 集成 #

pip 安装 langchain-openai

安装 Anthropic 集成 #

pip 安装 langchain-anthropic

安装社区集成（向量存储、加载器、工具） #

pip install langchain-社区

为代理工作流程安装 LangGraph #

pip 安装语言图

安装 LangSmith 以实现可观察性 #

pip 安装 langsmith ``enAI 从 langchain_anthropic 导入 ChatAnthropic# 测试模型实例化 openai_model = ChatOpenAI(模型=“gpt-4o”，温度=0) anthropic_model = ChatAnthropic(model=“claude-3-5-sonnet-20241022”)print(“LangChain 通过 OpenAI 和 Anthropic 提供商成功安装”) ### 环境配置 bas h

.env 文件 #

OPENAI_API_KEY=sk-proj-xxxxx ANTHROPIC_API_KEY=sk-ant-xxxxx LANGSMITH_API_KEY=ls-xxxxx LANGSMITH_TRACING=tr``` pytho n 导入langchain_core 打印（langchain_core.version）

输出：1.4.0 #

从 langchain_openai 导入 ChatOpenAI 从 langchain_anthropic 导入 ChatAnthropic

测试模型实例化 #

openai_model = ChatOpenAI(模型=“gpt-4o”，温度=0) anthropic_model = ChatAnthropic(model=“claude-3-5-sonnet-20241022”)

print(“LangChain 通过 OpenAI 和 Anthropic 提供商成功安装”)

t
# 复制应用程序代码
复制。 。# 为了安全起见，非 root 用户
运行 useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER 应用程序用户# 健康检查
HEALTHCHECK --间隔=30s --超时=10s --start-period=5s --retries=3 \
CMD python -c“导入 urllib.request; urllib.request.urlopen('http://localhost:8000/health')”曝光 8000
CMD [“uvicorn”，“main：app”，“--host”，“0.0.0.0”，“--port”，“8000”，“--workers”，“4”]

bas h

.env 文件 #

OPENAI_API_KEY=sk-proj-xxxxx ANTHROPIC_API_KEY=sk-ant-xxxxx LANGSMITH_API_KEY=ls-xxxxx LANGSMITH_TRACING=true LANGSMITH_PROJECT=生产代理

fastapi
==0.115.0
派丹蒂克==2.10.0
python-dotenv==1.0.0
redis==5.2.0
httpx==0.28.0

yam l

docker-compose.yml #

版本：‘3.8’服务：应用程序：构建： . 端口：

“8000:8000” 环境：

dockerfil
e
# Dockerfile
来自 python:3.12-slim

工作目录/应用程序

# 安装系统依赖
运行 apt-get update && apt-get install -y \
海湾合作委员会\
python3-dev \
&& rm -rf /var/lib/apt/lists/*

# 复制并安装Python依赖项
复制requirements.txt。 
运行 pip install --no-cache-dir -rrequirements.txt

# 复制应用程序代码
复制。 。 

# 为了安全起见，非 root 用户
运行 useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER 应用程序用户

# 健康检查
HEALTHCHECK --间隔=30s --超时=10s --start-period=5s --retries=3 \
CMD python -c“导入 urllib.request; urllib.request.urlopen('http://localhost:8000/health')”

曝光 8000
CMD [“uvicorn”，“main：app”，“--host”，“0.0.0.0”，“--port”，“8000”，“--workers”，“4”]
``Egration 与 OpenAI、Anthropic、Ollama 和 Vector Store### OpenAI GPT-4o 集成````蟒蛇
从 langchain_openai 导入 ChatOpenAI
从 langchain_core.prompts 导入 ChatPromptTemplate# 初始化模型
模型 = ChatOpenAI(
型号=“gpt-4o”，
温度=0.1，
最大令牌=4096，
超时=30，
最大重试次数=3，
）# 创建一条链
提示 = ChatPromptTemplate.from_messages([
("system", "你是一位乐于助人的助手，解答有关{topic}的问题。"),
（“人类”，“{问题}”），
]）链=提示| 模型# 调用
响应 = chain.invoke({
“主题”：“机器学习”，
"question": "用 3 句话解释反向传播。"
})
打印（响应.内容）
````### 人类克劳德积分``python``txt
# 需求.txt
langchain-core==1.4.0
语言链==1.3.0
langchain-openai==1.2.0
langchain-anthropic==1.4.0
语言图==0.4.0
朗史密斯==0.7.0
uvicorn==0.34.0
fastapi==0.115.0
派丹蒂克==2.10.0
python-dotenv==1.0.0
redis==5.2.0
httpx==0.28.0

laude _chain.invoke({ “topic”: “分布式系统”, “question”: “什么是 CAP 定理？” }) 打印（响应.内容） ### 奥拉马当地模特蟒蛇从 langchain_ollama 导入 ChatOllamalocal_model = ChatOllama( 模型=“l``` yam l

docker-compose.yml #

版本：‘3.8’

服务：应用程序：构建： . 端口：

“8000:8000” 环境：
OPENAI_API_KEY=${OPENAI_API_KEY}
ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
LANGSMITH_TRACING=true
REDIS_URL=redis://redis:6379 取决于：
雷迪斯
色度重新启动：除非停止

雷迪斯：图片：redis:7-alpine 卷：

redis_data:/数据重新启动：除非停止

色度：图片：chromadb/chroma：最新卷：

色度数据：/色度/色度重新启动：除非停止

卷： redis_数据：色度数据：

action
=len,
）
块= splitter.split_documents（文档）# 存储在向量数据库中
嵌入 = OpenAIEmbeddings(model="text-embedding-3-large")
矢量存储 = Chroma.from_documents(
文档=块，
嵌入=嵌入，
persist_directory="./chroma_db",
）# 创建 QA 链
qa_chain = RetrievalQA.from_chain_type(
llm=ChatOpenAI(型号=“gpt-4o”，温度=0),
chain_type=“东西”，
检索器=vectorstore.as_retriever(search_kwargs={"k": 5}),
return_source_documents=真，
）# 查询
result = qa_chain.invoke({"query": "主要发现是什么？"})
打印（结果[“结果”]）
````### 代理与工具````蟒蛇
从 langchain import hub
来自 langchain。```
bas
h
# 构建图像
docker build -t langchain-生产-app 。 

# 使用 docker-compose 运行
docker-compose up -d

# 验证部署
卷曲 http://localhost:8000/health

owledge _base(查询：str) -> str: “““在内部知识库中搜索技术文档。””” return f“‘{query}’的结果：找到 3 个相关文档。”@工具 def 计算(表达式: str) -> str: “”“评估数学表达式。”“” 尝试``` pytho n 从 langchain_openai 导入 ChatOpenAI 从 langchain_core.prompts 导入 ChatPromptTemplate

初始化模型 #

模型 = ChatOpenAI( 型号=“gpt-4o”，温度=0.1，最大令牌=4096，超时=30，最大重试次数=3，）

创建一条链 #

提示 = ChatPromptTemplate.from_messages([ (“system”, “你是一位乐于助人的助手，解答有关{topic}的问题。”), （“人类”，“{问题}”）， ]）

链=提示| 型号

调用 #

响应 = chain.invoke({ “主题”：“机器学习”， “question”: “用 3 句话解释反向传播。” }) 打印（响应.内容） ``nce 基准定性比较基于典型部署（未发布的官方基准）。性能因工作负载、配置和硬件而异：| 公制| 浪链 | 骆驼索引 | 干草堆 | 语义内核| |——–|———|————|———|—————–| | 吞吐量| 中等| 高| 最高| 中等| | 内存使用情况| 中高| 中等| 降低| 中等| | 延迟| 中等| 降低| 最低| 中等| | 积分计数 | 700+ | 300+ | 150+ | 80+ | | GitHub 之星 | 137,165 | 137,165 39,200 | 39,200 17,900 | 17,900 26,300 | 26,300 | 生产时间| 2-3天| 1-2 天 | 3````蟒蛇从 langchain_anthropic 导入 ChatAnthropic

克劳德 = 聊天人类（模型=“claude-3-5-sonnet-20241022”，温度=0.1，超时=30，最大重试次数=3，）

相同的提示适用于不同的提供商 #

claude_chain = 提示 | 克劳德响应 = claude_chain.invoke({ “topic”: “分布式系统”, “question”: “什么是 CAP 定理？” }) 打印（响应.内容） ``琐碎但缺乏代理能力。 LlamaIndex 提供最快的 RAG 时间，但检索范围更窄。 Semantic Kernel 与 Azure 原生集成，但开源生态系统较小。### 生产案例研究客户支持自动化（SaaS，50 万用户）。 一家 B2B SaaS 公司用集成了知识库、CRM 和 tic python 的 LangChain 代理取代了基于规则的支持机器人从 langchain_ollama 导入 ChatOllama

local_model = ChatOllama( 型号=“llama3.3”，温度=0.1， base_url =“http://localhost:11434”，）

response = local_model.invoke(“简单解释一下量子计算。”) 打印（响应.内容） ``* 诉讼团队使用LangChain RAG超过50,000份案件文件。该管道加载 PDF，通过语义分块进行分割，将嵌入存储在 Pinecone 中，并生成带有引文的备忘录草稿。每个案件的律师研究时间减少了 60%。**代码生成助手（金融科技，200 engi``` pytho n 从 langchain_community.document_loaders 导入 PyPDFLoader 从 langchain.text_splitter 导入 RecursiveCharacterTextSplitter 从 langchain_openai 导入 OpenAIEmbeddings 从 langchain_chroma 导入 Chroma 从 langchain.chains 导入 RetrievalQA

加载并分割文档 #

loader = PyPDFLoader("./documents.pdf") 文档 = loader.load()

分割器 = RecursiveCharacterTextSplitter( 块大小=1000，块重叠=200，长度函数=len，）块= splitter.split_documents（文档）

存储在向量数据库中 #

嵌入 = OpenAIEmbeddings(model=“text-embedding-3-large”) 矢量存储 = Chroma.from_documents( 文档=块，嵌入=嵌入， persist_directory="./chroma_db", ）

创建 QA 链 #

qa_chain = RetrievalQA.from_chain_type( llm=ChatOpenAI(型号=“gpt-4o”，温度=0), chain_type=“东西”，检索器=vectorstore.as_retriever(search_kwargs={“k”: 5}), return_source_documents=真，）

查询 #

result = qa_chain.invoke({“query”: “主要发现是什么？”}) 打印（结果[“结果”]）

essages
": [response], "next_step": " human_review"}def human_review(状态: AgentState):
# 在生产中，这会暂停以供人工批准
last_msg = 状态["消息"][-1].内容
如果在last_msg.upper()中“DELETE”或在last_msg.upper()中“DROP”：
返回{“next_step”：“拒绝”}
返回{“next_step”：“执行”}defexecute_tool(状态: AgentState):
return {"messages": [AIMessage(content="操作执行成功。")], "next_step": END}defreject_action(状态: AgentState):
return {"messages": [AIMessage(content="操作被策略拒绝。")], "next_step": END}# 构建图表
工作流程 = StateGraph(AgentState)
工作流.add_node("代理", agent_node)
工作流程.add_node（“人类评论”，人类评论）
工作流.add_node(“执行”,execute_tool)
工作流.add_node(“拒绝”,reject_action)工作流程.set_entry_point(“代理”)
工作流程.add_edge(“代理”,“人类评论”)
工作流程.add_conditional_edges(
“人类评论”，
拉姆达 x: x["next_step"],
{“执行”：“执行”，“拒绝”：“拒绝”}
）
workflow.add_edge("执行", END)
工作流程.add```
pytho
n
从 langchain import hub
从 langchain.agents 导入 create_tool_calling_agent, AgentExecutor
从 langchain_core.tools 导入工具
从 langchain_openai 导入 ChatOpenAI

# 定义自定义工具
@工具
def search_knowledge_base(查询: str) -> str:
"""在内部知识库中搜索技术文档。"""
return f“‘{query}’的结果：找到 3 个相关文档。”

@工具
def 计算(表达式: str) -> str:
“”“评估数学表达式。”“”
尝试：
结果 = eval(表达式)
返回str(结果)
除了异常 e：
返回 f“错误：{str(e)}”

# 创建代理
工具= [搜索知识库，计算]
llm = ChatOpenAI(模型=“gpt-4o”，温度=0)
提示 = hub.pull("hwchase17/openai-functions-agent")
代理 = create_tool_calling_agent(llm, 工具, 提示)
agent_executor = AgentExecutor（代理=代理，工具=工具，详细= True）

# 运行代理
结果 = agent_executor.invoke({
"input": "1250 * 37 是什么并搜索部署文档？"
})
打印（结果[“输出”]）
```内存中速率限制器
导入时间# 速率限制：每分钟 10 个请求
rate_limiter = InMemoryRateLimiter(
每秒请求数=10/60，
check_every_n_seconds=1,
最大桶大小=5,
）模型 = ChatOpenAI(
型号=“gpt-4o”，
速率限制器=速率限制器，
max_tokens=2000, # 输出代币的硬性上限
）# 跟踪每个请求的成本
从 langchain.callbacks 导入 get_openai_callback将 get_openai_callback() 作为 cb：
response = model.invoke("总结这份 50 页的报告。")
print(f"代币: {cb.total_tokens}, 成本: ${cb.total_cost:.4f}")
````### 使用 LangSmith 进行监控````蟒蛇
导入操作系统# 启用跟踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-xxxxx"
os.environ["LANGSMITH_PROJECT"] = "生产代理"来自 langsmith 导入客户端
客户端 = 客户端()# 程序化评估
从 langsmith.evaluation 导入评估def precision_evaluator（运行，示例）：
预测 = run.outputs["输出"]
预期 = example.outputs["expected_answer"]
如果在 Prediction.lower() 中预期.lower()，则得分 = 1.0，否则 0.0
return {"key": "准确率", "score": 分数}结果=评估（
lambda x：agent_executor.invoke(x)，
数据=“我的数据集名称”，
评估者=[accuracy_evaluator],
）
````### Kubernetes 部署````
yam
l
# k8s-deployment.yaml
api版本：apps/v1
种类：部署
元数据：
名称：langchain-app
标签：
应用程序：langchain-app
规格：
副本：3
选择器：
匹配标签：
应用程序：langchain-app
模板：
元数据：
标签：
应用程序：langchain-app
规格：
容器：
- 名称：应用程序
图片：langchain-生产-应用程序：最新
端口：
- 集装箱端口：8000
环境：
- 名称：OPENAI_API_KEY
值来自：
秘密密钥参考：
名称：api 秘密
密钥： openai-key
- 名称：LANGSMITH_API_KEY
值来自：
秘密密钥参考：
名称：api 秘密
密钥： langsmith 密钥
资源：
要求：
内存：“512Mi”
中央处理器：“500m”
限制：
内存：“2Gi”
中央处理器：“2000米”
活性探针：
http获取：
路径：/健康
端口：8000
初始延迟秒数：10
周期秒：30
准备情况探针：
http获取：
路径：/准备好
端口：8000
初始延迟秒数：5
周期秒：10
---
api版本：v1
种类： 服务
元数据：
名称：langchain-服务
规格：
选择器：
应用程序：langchain-app
端口：
- 协议：TCP
端口：80
目标端口：8000
类型：集群IP

bas h

部署到 Kubernetes #

kubectl apply -f k8s-deployment.yaml kubectl get pods -l app=langchain-app kubectl 日志 -f 部署/langchain-app ### Redis 缓存频繁查询蟒蛇导入redis 导入 json 导入哈希库从 langchain.globals 导入 set_llm_cache 从 langchain_community.cache 导入 RedisCache# 连接到Redis redis_client = redis.Redis.from_url(“redis://localhost:6379”) set_llm_cache(RedisCache(redis_client=redis_client))# 根据输入哈希缓存键 def get_cache_key(前缀: str, 文本: str) -> str: hash_val = hashlib.md5(text.encode()).hexdigest() 返回 f"{prefix}:{hash_val}"# 在昂贵的 LLM 调用之前检查缓存 def cached_invoke（链，输入：dict，ttl：int = 3600）： cache_key = get_cache_key(“llm”, json.dumps(inputs, sort_keys=True)) 缓存 = redis_client.get(cache_key) 如果缓存：返回 json.loads(缓存)结果 = chain.invoke(输入) redis_c``` pytho n 从 langgraph.graph 导入 StateGraph，END 从输入导入 TypedDict、注释、序列从 langchain_core.messages 导入 BaseMessage、HumanMessage、AIMessage 从 langchain_openai 导入 ChatOpenAI 进口经营者

定义状态 #

类 AgentState(TypedDict): 消息：带注释的[序列[BaseMessage]，operator.add] 下一步：str

定义节点 #

def agent_node(状态: AgentState): 模型= ChatOpenAI(模型=“gpt-4o”) 响应 = model.invoke(state[“messages”]) 返回 {“messages”: [response], “next_step”: " human_review"}

def human_review(状态: AgentState):

在生产中，这会暂停以供人工批准 #

last_msg = 状态[“消息”][-1].内容如果在last_msg.upper()中“DELETE”或在last_msg.upper()中“DROP”：返回{“next_step”：“拒绝”} 返回{“next_step”：“执行”}

defexecute_tool(状态: AgentState): return {“messages”: [AIMessage(content=“操作执行成功。”)], “next_step”: END}

defreject_action(状态: AgentState): return {“messages”: [AIMessage(content=“操作被策略拒绝。”)], “next_step”: END}

构建图表 #

工作流程 = StateGraph(AgentState) 工作流.add_node(“代理”, agent_node) 工作流程.add_node（“人类评论”，人类评论）工作流.add_node(“执行”,execute_tool) 工作流.add_node(“拒绝”,reject_action)

工作流程.set_entry_point(“代理”) 工作流程.add_edge(“代理”,“人类评论”) 工作流程.add_conditional_edges( “人类评论”，拉姆达 x: x[“next_step”], {“执行”：“执行”，“拒绝”：“拒绝”} ） workflow.add_edge(“执行”, END) 工作流程.add_edge(“拒绝”, END)

编译并运行 #

应用程序=工作流程.编译() result = app.invoke({“messages”: [HumanMessage(content=“从数据库中删除所有用户记录。”)]})\nprint(result[“messages”][-1].content) 准确性基准。 LangChain的优势在于编排，而不是原始检索速度。**简单用例的学习曲线更陡。** 基本的“与 PDF 聊天”应用程序需要了解加载器、拆分器、嵌入、向量存储和链。 RAGFlow 或 Verba 等工具为非开发人员提供了更快的路径。**快速演进造成版本漂移。** 尽管做出了 1.0 LTS 的承诺，但生态系统的发展速度很快。社区集成（“langchain-community”）可以在次要版本中引入重大更改。在生产中固定精确版本。**LangSmith 成本随使用情况而变化。** 免费套餐每月覆盖 5,000 个跟踪 — 足以用于原型设计，但不足以用于生产。每月处理 100,000 条跟踪的 5 人团队仅为 LangSmith 支付的费用约为 220 美元/月，不包括 LLM API 成本。**过度设计风险。** LangChain的灵活性吸引开发人员构建复杂的代理图，其中简单的提示+API调用就足够了。从简单开始，只有在指标证明合理的情况下才增加复杂性。**有限的 C# 和 Java 生态系统。** 以 Microsoft 为中心的环境中的团队可能会发现 Semantic Kernel 一流的 .NET 支持比 LangChain 的 Python 优先方法更自然。## 常见问题### LangChain 和 LangGraph 有什么区别？LangChain是构建LLM应用程序的核心框架，具有链、提示和模型集成。 LangGraph 是一个扩展库，它为复杂的代理工作流程添加了基于图形的编排，包括循环、分支和人机交互批准。将 LangChain 视为组件库，将 LangGraph 视为工作流引擎。两者均由LangChain Inc.维护并具有相同的发布周期。### 如何在LangChain中切换LLM提供商？更改模型类导入。 LangChain的标准化“BaseChatModel”接口意味着为OpenAI wopython编写的代码从 langchain_core.runnables 导入 RunnableConfig from tenacity import retry, stop_after_attempt, wait_exponential

@重试（停止=尝试后停止(3), 等待=wait_exponential（乘数=1，最小值=2，最大值=10），重新加注=真） def invoke_with_retry（链，输入，配置：RunnableConfig = None）：尝试：返回 chain.invoke(输入, config=config) 除了异常 e：

登录LangSmith进行分析 #

print(f"调用失败：{e}。正在重试…") 提高

用法 #

config = RunnableConfig(tags=[“生产”, “面向客户”]) result = invoke_with_retry(qa_chain, {“query”: “条款是什么？”}, config)

LangChai
n
在生产环境中的部署堆栈？对于生产部署，请使用带有 WSGI/ASGI 服务器（Uvicorn 或 Gunicorn）的 Docker 容器、用于缓存和会话状态的 Redis、矢量存储（用于小规模的 Chroma、用于大规模的 Pinecone 或 Weaviate）以及用于可观察性的 LangSmith。 在 Kubernetes 上部署以进行水平扩展。 设置资源限制、运行状况检查和速率限制。 固定所有依赖项版本并在每次部署之前运行评估。### LangChain如何处理错误和重试？LangChain 通过模型类上的 max_retries 参数提供内置的具有指数退避的重试逻辑。 对于生产，用 Tenacity 包装关键路径以实现良好的Python
从 langchain_core.rate_limiters 导入 InMemoryRateLimiter
导入时间

# 速率限制：每分钟 10 个请求
rate_limiter = InMemoryRateLimiter(
每秒请求数=10/60，
check_every_n_seconds=1,
最大桶大小=5,
）

模型 = ChatOpenAI(
型号=“gpt-4o”，
速率限制器=速率限制器，
max_tokens=2000, # 输出代币的硬性上限
）

# 跟踪每个请求的成本
从 langchain.callbacks 导入 get_openai_callback

将 get_openai_callback() 作为 cb：
response = model.invoke("总结这份 50 页的报告。")
print(f"代币: {cb.total_tokens}, 成本: ${cb.total_cost:.4f}")

al e LangChain 代理可以处理 1000+ 并发用户吗？通过在负载均衡器后面运行多个容器实例来水平扩展。使用异步模式（“ainvoke”、“astream”）来最大化每个工作线程的吞吐量。对常见查询实施 Redis 缓存。为数据库和外部 API 设置连接池。通过 LangSmith 监控令牌使用情况和每个请求的成本。考虑使用队列系统（Celery、RQ）来执行长时间运行的代理任务，而不是同步 HTTP 请求。＃＃结论![LangChain 集成图](https://raw.githubusercontent.com/langchain-ai/langchain/master/docs/static/img/social_share.png``` pytho n 导入操作系统

启用跟踪 #

os.environ[“LANGSMITH_TRACING”] = “true” os.environ[“LANGSMITH_API_KEY”] = “ls-xxxxx” os.environ[“LANGSMITH_PROJECT”] = “生产代理”

来自 langsmith 导入客户端客户端 = 客户端()

程序化评估 #

从 langsmith.evaluation 导入评估

def precision_evaluator（运行，示例）：预测 = run.outputs[“输出”] 预期 = example.outputs[“expected_answer”] 如果在 Prediction.lower() 中预期.lower()，则得分 = 1.0，否则 0.0 return {“key”: “准确率”, “score”: 分数}

结果=评估（ lambda x：agent_executor.invoke(x)，数据=“我的数据集名称”，评估者=[accuracy_evaluator], ）

4. 加入 [Discord 上的 LangChain 社区](https://discord.gg/langchain) 和 [Telegram AI 开发小组](https://t.me/ai_source_code_hub) 进行生产部署讨论







## 推荐的托管和基础设施在将上述任何工具部署到生产环境之前，您需要坚实的基础设施。 dibi8实际使用和推荐的两个选项：- **{< aff "digitalocean" "footer-cta-legacy" "DigitalOcean" >}}** — 200 美元免费赠金，为期 60 天，覆盖全球 14 个以上区域。 运行开源人工智能工具的独立开发者的默认选项。 
- **HTStack
** — 具有低 la```
yam
l
的香港 VPS
# k8s-deployment.yaml
api版本：apps/v1
种类：部署
元数据：
名称：langchain-app
标签：
应用程序：langchain-app
规格：
副本：3
选择器：
匹配标签：
应用程序：langchain-app
模板：
元数据：
标签：
应用程序：langchain-app
规格：
容器：
- 名称：应用程序
图片：langchain-生产-应用程序：最新
端口：
- 集装箱端口：8000
环境：
- 名称：OPENAI_API_KEY
值来自：
秘密密钥参考：
名称：api 秘密
密钥： openai-key
- 名称：LANGSMITH_API_KEY
值来自：
秘密密钥参考：
名称：api 秘密
密钥： langsmith 密钥
资源：
要求：
内存：“512Mi”
中央处理器：“500m”
限制：
内存：“2Gi”
中央处理器：“2000米”
活性探针：
http获取：
路径：/健康
端口：8000
初始延迟秒数：10
周期秒：30
准备情况探针：
http获取：
路径：/准备好
端口：8000
初始延迟秒数：5
周期秒：10
---
api版本：v1
种类： 服务
元数据：
名称：langchain-服务
规格：
选择器：
应用程序：langchain-app
端口：
- 协议：TCP
端口：80
目标端口：8000
类型：集群IP

angGraph ](https://github.com/langchain-ai/langgraph)

bas
h
# 部署到 Kubernetes
kubectl apply -f k8s-deployment.yaml
kubectl get pods -l app=langchain-app
kubectl 日志 -f 部署/langchain-app
``````蟒蛇
导入redis
导入 json
导入哈希库
从 langchain.globals 导入 set_llm_cache
从 langchain_community.cache 导入 RedisCache

# 连接到Redis
redis_client = redis.Redis.from_url("redis://localhost:6379")
set_llm_cache(RedisCache(redis_client=redis_client))

# 根据输入哈希缓存键
def get_cache_key(前缀: str, 文本: str) -> str:
hash_val = hashlib.md5(text.encode()).hexdigest()
返回 f"{prefix}:{hash_val}"

# 在昂贵的 LLM 调用之前检查缓存
def cached_invoke（链，输入：dict，ttl：int = 3600）：
cache_key = get_cache_key("llm", json.dumps(inputs, sort_keys=True))
缓存 = redis_client.get(cache_key)
如果缓存：
返回 json.loads(缓存)

结果 = chain.invoke(输入)
redis_client.setex(cache_key, ttl, json.dumps({"output": result.content}))
返回结果

featureImage: /images/articles/langchain部署生产就绪型人工智能代理的-3-种方法.jpg

安装核心框架 #

安装 OpenAI 集成 #

安装 Anthropic 集成 #

安装社区集成（向量存储、加载器、工具） #

为代理工作流程安装 LangGraph #

安装 LangSmith 以实现可观察性 #

.env 文件 #

输出：1.4.0 #

测试模型实例化 #

.env 文件 #

docker-compose.yml #

docker-compose.yml #

初始化模型 #

创建一条链 #

调用 #

相同的提示适用于不同的提供商 #

加载并分割文档 #

存储在向量数据库中 #

创建 QA 链 #

查询 #

部署到 Kubernetes #

定义状态 #

定义节点 #

在生产中，这会暂停以供人工批准 #

构建图表 #

编译并运行 #

登录LangSmith进行分析 #

用法 #

启用跟踪 #

程序化评估 #

🔗 相关资源推荐

💬 留言讨论