LangChain: 3种部署生产级AI智能体的方法 — 2026年完整部署指南

# 安装核心框架
pip install langchain-core==1.4.0 langchain

# 安装OpenAI集成
pip install langchain-openai

# 安装Anthropic集成
pip install langchain-anthropic

# 安装社区集成（向量存储、加载器、工具）
pip install langchain-community

# 安装LangGraph智能体工作流
pip install langgraph

# 安装LangSmith可观测性
pip install langsmith

import langchain_core
print(langchain_core.__version__)
# Output: 1.4.0

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 测试模型实例化
openai_model = ChatOpenAI(model="gpt-4o", temperature=0)
anthropic_model = ChatAnthropic(model="claude-3-5-sonnet-20241022")

print("LangChain installed successfully with OpenAI and Anthropic providers")

# .env 文件
OPENAI_API_KEY=sk-proj-xxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxx
LANGSMITH_API_KEY=ls-xxxxx
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=production-agents

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    python3-dev \
    && rm -rf /var/lib/apt/lists/*

# 复制并安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 非root用户保障安全
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# requirements.txt
langchain-core==1.4.0
langchain==1.3.0
langchain-openai==1.2.0
langchain-anthropic==1.4.0
langgraph==0.4.0
langsmith==0.7.0
uvicorn==0.34.0
fastapi==0.115.0
pydantic==2.10.0
python-dotenv==1.0.0
redis==5.2.0
httpx==0.28.0

# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - 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
    depends_on:
      - redis
      - chroma
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

  chroma:
    image: chromadb/chroma:latest
    volumes:
      - chroma_data:/chroma/chroma
    restart: unless-stopped

volumes:
  redis_data:
  chroma_data:

# 构建镜像
docker build -t langchain-production-app .

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

# 验证部署
curl http://localhost:8000/health

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 初始化模型
model = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    max_tokens=4096,
    timeout=30,
    max_retries=3,
)

# 创建链
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that answers questions about {topic}."),
    ("human", "{question}"),
])

chain = prompt | model

# 调用
response = chain.invoke({
    "topic": "machine learning",
    "question": "Explain backpropagation in 3 sentences."
})
print(response.content)

from langchain_anthropic import ChatAnthropic

claude = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0.1,
    timeout=30,
    max_retries=3,
)

# 相同的提示跨提供商通用
claude_chain = prompt | claude
response = claude_chain.invoke({
    "topic": "distributed systems",
    "question": "What is the CAP theorem?"
})
print(response.content)

from langchain_ollama import ChatOllama

local_model = ChatOllama(
    model="llama3.3",
    temperature=0.1,
    base_url="http://localhost:11434",
)

response = local_model.invoke("Explain quantum computing simply.")
print(response.content)

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA

# 加载并分割文档
loader = PyPDFLoader("./documents.pdf")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
)
chunks = splitter.split_documents(docs)

# 存入向量数据库
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
)

# 创建QA链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True,
)

# 查询
result = qa_chain.invoke({"query": "What are the key findings?"})
print(result["result"])

from langchain import hub
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# 定义自定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base for technical documentation."""
    return f"Results for '{query}': Found 3 relevant documents."

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

# 创建智能体
tools = [search_knowledge_base, calculate]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行智能体
result = agent_executor.invoke({
    "input": "What is 1250 * 37 and search for deployment docs?"
})
print(result["output"])

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
import operator

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_step: str

# 定义节点
def agent_node(state: AgentState):
    model = ChatOpenAI(model="gpt-4o")
    response = model.invoke(state["messages"])
    return {"messages": [response], "next_step": "human_review"}

def human_review(state: AgentState):
    # 生产中这里会暂停等待人工审批
    last_msg = state["messages"][-1].content
    if "DELETE" in last_msg.upper() or "DROP" in last_msg.upper():
        return {"next_step": "reject"}
    return {"next_step": "execute"}

def execute_tool(state: AgentState):
    return {"messages": [AIMessage(content="Action executed successfully.")], "next_step": END}

def reject_action(state: AgentState):
    return {"messages": [AIMessage(content="Action rejected by policy.")], "next_step": END}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("human_review", human_review)
workflow.add_node("execute", execute_tool)
workflow.add_node("reject", reject_action)

workflow.set_entry_point("agent")
workflow.add_edge("agent", "human_review")
workflow.add_conditional_edges(
    "human_review",
    lambda x: x["next_step"],
    {"execute": "execute", "reject": "reject"}
)
workflow.add_edge("execute", END)
workflow.add_edge("reject", END)

# 编译并运行
app = workflow.compile()
result = app.invoke({"messages": [HumanMessage(content="Delete all user records from the database.")]})
print(result["messages"][-1].content)

from langchain_core.runnables import RunnableConfig
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True
)
def invoke_with_retry(chain, inputs, config: RunnableConfig = None):
    try:
        return chain.invoke(inputs, config=config)
    except Exception as e:
        # 记录到LangSmith进行分析
        print(f"Invocation failed: {e}. Retrying...")
        raise

# 用法
config = RunnableConfig(tags=["production", "customer-facing"])
result = invoke_with_retry(qa_chain, {"query": "What are the terms?"}, config)

from langchain_core.rate_limiters import InMemoryRateLimiter
import time

# 速率限制：每分钟10次请求
rate_limiter = InMemoryRateLimiter(
    requests_per_second=10/60,
    check_every_n_seconds=1,
    max_bucket_size=5,
)

model = ChatOpenAI(
    model="gpt-4o",
    rate_limiter=rate_limiter,
    max_tokens=2000,  # 输出token硬上限
)

# 追踪每次请求的成本
from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    response = model.invoke("Summarize this 50-page report.")
    print(f"Tokens: {cb.total_tokens}, Cost: ${cb.total_cost:.4f}")

import os

# 启用追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-xxxxx"
os.environ["LANGSMITH_PROJECT"] = "production-agents"

from langsmith import Client
client = Client()

# 程序化评估
from langsmith.evaluation import evaluate

def accuracy_evaluator(run, example):
    prediction = run.outputs["output"]
    expected = example.outputs["expected_answer"]
    score = 1.0 if expected.lower() in prediction.lower() else 0.0
    return {"key": "accuracy", "score": score}

results = evaluate(
    lambda x: agent_executor.invoke(x),
    data="my-dataset-name",
    evaluators=[accuracy_evaluator],
)

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-app
  labels:
    app: langchain-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain-app
  template:
    metadata:
      labels:
        app: langchain-app
    spec:
      containers:
      - name: app
        image: langchain-production-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: openai-key
        - name: LANGSMITH_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: langsmith-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: langchain-service
spec:
  selector:
    app: langchain-app
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000
  type: ClusterIP

# 部署到Kubernetes
kubectl apply -f k8s-deployment.yaml
kubectl get pods -l app=langchain-app
kubectl logs -f deployment/langchain-app

import redis
import json
import hashlib
from langchain.globals import set_llm_cache
from langchain_community.cache import RedisCache

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

# 基于输入哈希的缓存键
def get_cache_key(prefix: str, text: str) -> str:
    hash_val = hashlib.md5(text.encode()).hexdigest()
    return f"{prefix}:{hash_val}"

# 昂贵LLM调用前检查缓存
def cached_invoke(chain, inputs: dict, ttl: int = 3600):
    cache_key = get_cache_key("llm", json.dumps(inputs, sort_keys=True))
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    result = chain.invoke(inputs)
    redis_client.setex(cache_key, ttl, json.dumps({"output": result.content}))
    return result