LangChain 知识速查表

本页面汇总了 LangChain 和 LangGraph 编程中最常用的语法和知识点，方便快速查阅。

安装

# 核心包
pip install langchain

# 模型支持
pip install langchain-openai      # OpenAI
pip install langchain-anthropic   # Anthropic
pip install langchain-google-genai # Google

# LangGraph（Agent 框架）
pip install langgraph

# 向量数据库
pip install langchain-chroma
pip install langchain-pinecone
pip install langchain-qdrant

# 工具和追踪
pip install langsmith

初始化模型

from langchain.chat_models import init_chat_model

# 推荐格式：provider:model
model = init_chat_model("openai:gpt-4o-mini")
model = init_chat_model("anthropic:claude-sonnet-4-5-20250929")
model = init_chat_model("google_genai:gemini-1.5-pro")

# 或分开指定
model = init_chat_model(
    model="gpt-4o-mini",
    model_provider="openai"
)

# 带参数
model = init_chat_model(
    model="openai:gpt-4o-mini",
    temperature=0.7,
    max_tokens=1000,
    timeout=60,
    max_retries=6
)

调用模型

# invoke：单次调用
response = model.invoke("你好")
print(response.content)

# 带消息列表
messages = [
    {"role": "system", "content": "你是一个助手"},
    {"role": "user", "content": "你好"}
]
response = model.invoke(messages)

# stream：流式输出
for chunk in model.stream("讲一个故事"):
    print(chunk.content, end="", flush=True)

# batch：批量调用
results = model.batch(["你好", "再见"])

# 异步调用
result = await model.ainvoke("你好")
async for chunk in model.astream("故事"):
    print(chunk.content)

消息类型

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 人类消息
msg = HumanMessage(content="你好")

# AI 消息
msg = AIMessage(content="你好，有什么可以帮你？")

# 系统消息
msg = SystemMessage(content="你是一个专业助手")

# 消息列表
messages = [
    SystemMessage(content="你是Python专家"),
    HumanMessage(content="什么是装饰器？")
]

Prompt 模板

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 聊天模板
prompt = ChatPromptTemplate([
    ("system", "你是一个{role}专家"),
    ("human", "{question}")
])

# 格式化
messages = prompt.format_messages(role="Python", question="什么是装饰器？")

# 带历史消息占位符
prompt = ChatPromptTemplate([
    ("system", "你是AI助手"),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{question}")
])

# 部分填充
partial_prompt = prompt.partial(role="Python专家")

工具调用

from langchain.tools import tool

# 定义工具
@tool
def get_weather(location: str) -> str:
    """获取指定位置的天气"""
    return f"{location}今天晴朗，25°C"

# 绑定工具到模型
model_with_tools = model.bind_tools([get_weather])

# 调用
response = model_with_tools.invoke("北京天气怎么样？")

# 检查工具调用
if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"工具: {tool_call['name']}")
        print(f"参数: {tool_call['args']}")

结构化输出

from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

# 强制结构化输出
structured_model = model.with_structured_output(Person)
result = structured_model.invoke("张三今年28岁，是软件工程师")
# result = Person(name='张三', age=28, occupation='软件工程师')

LCEL 链式组合

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

model = init_chat_model("openai:gpt-4o-mini")

# 简单链
chain = prompt | model | StrOutputParser()

# 并行处理
chain = RunnableParallel(
    answer=prompt | model | StrOutputParser(),
    question=RunnablePassthrough()
)

# 调用
result = chain.invoke({"question": "什么是Python？"})

# 流式
for chunk in chain.stream({"question": "什么是Python？"}):
    print(chunk, end="")

# 批量
results = chain.batch([
    {"question": "什么是Python？"},
    {"question": "什么是Java？"}
])

Agent 创建（LangChain 1.0）

from langchain.agents import create_agent
from langchain.tools import tool

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取城市天气信息"""
    return f"{city}天气：晴，25°C"

# 创建 Agent
agent = create_agent(
    model=init_chat_model("openai:gpt-4o-mini"),
    tools=[get_weather],
    system_prompt="你是一个有帮助的助手"
)

# 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？"}]
})

# 带记忆的 Agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver(),
    system_prompt="你是一个助手"
)

# 使用 thread_id 区分对话
config = {"configurable": {"thread_id": "user_123"}}
result = agent.invoke({"messages": [...]}, config=config)

记忆功能

from langgraph.checkpoint.memory import InMemorySaver

# 内存记忆
checkpointer = InMemorySaver()

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer
)

# 使用 thread_id 区分对话
config = {"configurable": {"thread_id": "conversation_001"}}

# 对话会自动保持上下文
result1 = agent.invoke({"messages": [...]}, config=config)
result2 = agent.invoke({"messages": [...]}, config=config)

向量存储

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 相似度搜索
docs = vectorstore.similarity_search("查询文本", k=3)

# 带分数的搜索
results = vectorstore.similarity_search_with_score("查询文本", k=3)

# 获取检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# MMR 检索（多样性）
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 阈值检索
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 10, "score_threshold": 0.8}
)

RAG 检索增强

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain.output_parsers import StrOutputParser

model = init_chat_model("openai:gpt-4o-mini")

# RAG 提示模板
rag_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
{context}

问题：{question}
""")

# 格式化文档
def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

# 构建 RAG 链
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | model
    | StrOutputParser()
)

# 调用
answer = rag_chain.invoke("什么是 Python？")

# 带来源的 RAG
rag_with_sources = RunnableParallel(
    answer=rag_chain,
    sources=retriever
)

文档加载与分割

from langchain_community.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载文档
loader = TextLoader("document.txt")
docs = loader.load()

loader = PyPDFLoader("report.pdf")
docs = loader.load()

loader = DirectoryLoader("docs/", glob="**/*.md")
docs = loader.load()

# 分割文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
splits = splitter.split_documents(docs)

可配置模型

from langchain.chat_models import init_chat_model

# 创建可配置模型（运行时切换模型）
configurable_model = init_chat_model(temperature=0)

# 运行时指定模型
response = configurable_model.invoke(
    "你好",
    config={"configurable": {"model": "openai:gpt-4o-mini"}}
)

# 使用不同模型
response = configurable_model.invoke(
    "你好",
    config={"configurable": {"model": "anthropic:claude-sonnet-4-5-20250929"}}
)

部署速查

LangServe 快速部署

from fastapi import FastAPI
from langserve import add_routes

app = FastAPI()

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

# 启动
# uvicorn server:app --host 0.0.0.0 --port 8000

LangGraph Platform 部署

# 安装 CLI
pip install langgraph-cli

# 本地开发
langgraph dev

# 构建 Docker 镜像
langgraph build -t my-agent

# 运行
docker run -p 8123:8000 \
    -e REDIS_URI=redis://... \
    -e DATABASE_URI=postgres://... \
    my-agent

客户端调用

from langserve import RemoteRunnable

# 连接远程服务
remote = RemoteRunnable("http://localhost:8000/chain")

# 调用
result = remote.invoke({"question": "..."})

# 流式
for chunk in remote.stream({"question": "..."}):
    print(chunk)

环境变量

# API Keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AIza...

# LangSmith 追踪
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=ls-...
LANGSMITH_PROJECT=my-project

# 后台回调（默认）
LANGCHAIN_CALLBACKS_BACKGROUND=true

调试

# 使用控制台回调
from langchain_core.tracers import ConsoleCallbackHandler

result = chain.invoke(
    {"question": "test"},
    config={"callbacks": [ConsoleCallbackHandler()]}
)

# 全局调试模式
from langchain.globals import set_debug
set_debug(True)

模块速查

模块	内容
`langchain.chat_models`	`init_chat_model`
`langchain_core.messages`	`HumanMessage`, `AIMessage`, `SystemMessage`
`langchain.agents`	`create_agent`, `AgentState`
`langchain.tools`	`@tool`, `BaseTool`
`langchain_core.prompts`	`ChatPromptTemplate`, `PromptTemplate`, `MessagesPlaceholder`
`langchain_core.runnables`	`RunnablePassthrough`, `RunnableLambda`, `RunnableParallel`
`langgraph.checkpoint.memory`	`InMemorySaver`
`langgraph.checkpoint.sqlite`	`SqliteSaver`
`langchain.text_splitter`	`RecursiveCharacterTextSplitter`
`langchain.output_parsers`	`StrOutputParser`, `PydanticOutputParser`

常用模式

管道链

chain = prompt | model | StrOutputParser()

并行处理

chain = RunnableParallel(
    a=chain_a,
    b=chain_b
)

条件分支

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: x["type"] == "a", chain_a),
    (lambda x: x["type"] == "b", chain_b),
    default_chain
)

重试机制

chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

回退机制

chain_with_fallback = chain.with_fallback([
    backup_chain_a,
    backup_chain_b
])

安装​

初始化模型​

调用模型​

消息类型​

Prompt 模板​

工具调用​

结构化输出​

LCEL 链式组合​

Agent 创建（LangChain 1.0）​

记忆功能​

向量存储​

RAG 检索增强​

文档加载与分割​

可配置模型​

部署速查​

LangServe 快速部署​

LangGraph Platform 部署​

客户端调用​

环境变量​

调试​

模块速查​

常用模式​

管道链​

并行处理​

条件分支​

重试机制​

回退机制​

参考资源​

安装

初始化模型

调用模型

消息类型

Prompt 模板

工具调用

结构化输出

LCEL 链式组合

Agent 创建（LangChain 1.0）

记忆功能

向量存储

RAG 检索增强

文档加载与分割

可配置模型

部署速查

LangServe 快速部署

LangGraph Platform 部署

客户端调用

环境变量

调试

模块速查

常用模式

管道链

并行处理

条件分支

重试机制

回退机制

参考资源