LangChain 2026演进：从框架到平台

LangChain生态系统
├── LangChain (核心库)
│   ├── Models (模型抽象)
│   ├── Prompts (提示管理)
│   ├── Chains (链式调用)
│   ├── Agents (智能体)
│   ├── Memory (记忆)
│   ├── Retrievers (检索器)
│   └── Tools (工具集)
├── LangGraph (Agent编排)
│   ├── State Graph (状态图)
│   ├── Checkpointing (检查点)
│   └── Human-in-loop (人机协作)
├── LangSmith (可观测性)
│   ├── Tracing (追踪)
│   ├── Evaluation (评估)
│   └── Monitoring (监控)
└── LangServe (部署)
    ├── API Server
    └── Streaming (流式)

from langchain_community.llms import Ollama
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 统一接口，不同后端
models = {
    "gpt5": ChatOpenAI(model="gpt-5"),
    "claude4": ChatAnthropic(model="claude-4-opus"),
    "glm5": Ollama(model="glm-5:32b"),
}

# 统一调用
for name, model in models.items():
    response = model.invoke("你好")
    print(f"{name}: {response.content}")

from langchain.prompts import ChatPromptTemplate

# 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}。"),
    ("human", "{question}")
])

# 链式调用
chain = prompt | model | output_parser
response = chain.invoke({"role": "数学老师", "question": "1+1=?"})

from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 文档处理
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_text(document)

# 嵌入和存储
embeddings = OllamaEmbeddings(model="bge-large-zh")
vectorstore = Chroma.from_texts(chunks, embeddings)

# RAG链
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | output_parser
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    tool_results: list
    next_action: str

def call_model(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

app = workflow.compile()

from langgraph.checkpoint import MemorySaver

# 支持对话中断和恢复
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 使用线程ID，可以中断和恢复
config = {"configurable": {"thread_id": "thread-1"}}
response = app.invoke({"messages": [("user", "你好")]}, config)

def human_review(state: AgentState):
    """人工审核节点"""
    # 暂停执行，等待人工输入
    pass

workflow.add_node("human_review", human_review)
# 在关键步骤加入人工审核

import langsmith

# 自动追踪所有LangChain调用
with langsmith.trace("my_chain") as run:
    response = chain.invoke({"question": "什么是AI?"})
    # 所有中间步骤都被追踪

from langsmith import Client

client = Client()

# 创建评估数据集
dataset = client.create_dataset("QA评估集")
client.create_examples(
    inputs=[{"question": "1+1=?"}, {"question": "什么是AI?"}],
    outputs=[{"answer": "2"}, {"answer": "AI是人工智能"}],
    dataset_id=dataset.id
)

# 运行评估
results = client.run_on_dataset(
    dataset_name="QA评估集",
    llm_or_chain_factory=chain,
    evaluation=eval_config
)

from langserve import add_routes
from fastapi import FastAPI

app = FastAPI()

# 添加API路由
add_routes(app, chain, path="/chat")
add_routes(app, rag_chain, path="/rag")

# 自动生成OpenAPI文档
# 访问 /docs 查看API文档

from langserve import add_routes

# 流式输出
add_routes(app, chain, path="/chat/stream", stream=True)

from langchain_core.messages import HumanMessage

# 图像输入
message = HumanMessage(
    content=[
        {"type": "text", "text": "描述这张图"},
        {"type": "image_url", "image_url": "data:image/jpeg;base64,..."}
    ]
)
response = model.invoke([message])

# 自动评估链的性能
from langsmith.evaluation import LangChainStringEvaluator

evaluator = LangChainStringEvaluator("qa_correctness")
results = evaluator.evaluate_chain(
    chain=chain,
    dataset="qa_test_set"
)

# 异步调用
response = await model.ainvoke("你好")

# 批量异步
responses = await model.abatch(["问题1", "问题2", "问题3"])

from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache

# 启用缓存
set_llm_cache(InMemoryCache())

# 相同输入只调用一次
response1 = model.invoke("你好")  # 调用LLM
response2 = model.invoke("你好")  # 从缓存读取

# 流式输出
async for chunk in model.astream("写一首诗"):
    print(chunk.content, end="", flush=True)