LangChain 框架全景

from langchain.chat_models import init_chat_model

# 字符串格式："provider:model"
llm = init_chat_model("openai:gpt-4o-mini")
llm = init_chat_model("anthropic:claude-sonnet-4-5")
llm = init_chat_model("ollama:qwen2.5:7b")          # 本地模型

# 也可以显式传参，适合统一配置温度等选项
llm = init_chat_model(
    model="gpt-4o-mini",
    model_provider="openai",
    temperature=0,
    max_tokens=2048,
)

# 运行时动态切换：configurable 字段
configurable_llm = init_chat_model(
    temperature=0,
    configurable_fields=("model", "model_provider", "temperature"),
)
fast = configurable_llm.with_config(
    config={"configurable": {"model": "gpt-4o-mini", "model_provider": "openai"}}
)
smart = configurable_llm.with_config(
    config={"configurable": {"model": "claude-sonnet-4-5", "model_provider": "anthropic"}}
)

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

from langchain.messages import HumanMessage, SystemMessage

resp = llm.invoke([
    SystemMessage("你是一个严谨的技术文档助手，回答必须附带出处。"),
    HumanMessage("什么是 Runnable？"),
])
print(resp.content)          # 文本内容
print(resp.tool_calls)       # 若模型决定调用工具，这里会有调用请求
print(resp.usage_metadata)   # {'input_tokens': 36, 'output_tokens': 128, 'total_tokens': 164}
print(resp.response_metadata)  # 原始厂商返回信息，排查问题时用

from langchain.messages import HumanMessage

# 多模态输入：文本 + 图片混合成块列表
msg = HumanMessage(content=[
    {"type": "text", "text": "这张架构图里有哪些组件？"},
    {"type": "image", "source_type": "url", "url": "https://example.com/arch.png"},
])
resp = llm.invoke([msg])

# 1.x 统一的 content_blocks 视图（推荐在应用层消费）
for block in resp.content_blocks:
    if block["type"] == "text":
        print(block["text"])
    elif block["type"] == "reasoning":       # 推理模型的思考过程
        print("[思考]", block["reasoning_text"][:50])

# 单次调用
resp = llm.invoke([HumanMessage("用一句话解释向量检索")])

# 流式输出：逐 token 返回
for chunk in llm.stream([HumanMessage("写一首关于编译器的俳句")]):
    print(chunk.text(), end="", flush=True)

# 批量调用：自动并发
results = llm.batch([
    [HumanMessage("总结这段代码")],
    [HumanMessage("总结这段文档")],
])

# 异步版本：高并发场景直接换前缀 a
# await llm.ainvoke(...) / async for chunk in llm.astream(...)

from langchain_core.prompts import ChatPromptTemplate
from langchain.messages import HumanMessage

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{name}领域的高级工程师，回答简洁、可执行。"),
    ("human", "{question}"),
])

# 模板本身是 Runnable：可以单独调用、校验变量
msgs = prompt.invoke({"name": "数据库", "question": "如何定位慢查询？"})
print(msgs)  # ChatPromptValue，含格式化好的消息列表

# 变量缺失会直接报错，而不是把 {question} 原样发给模型
# prompt.invoke({"name": "数据库"})  --> pydantic 校验错误

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是技术面试官，正在面试一位{level}工程师。"),
    MessagesPlaceholder("history"),        # 历史消息插槽
    ("human", "{question}"),
])

chain = prompt | llm

history = [
    ("ai", "请先自我介绍。"),
    ("human", "我有五年后端经验，主要做交易系统。"),
]
resp = chain.invoke({
    "level": "资深",
    "history": history,
    "question": "谈谈你处理过的最复杂的线上故障。",
})

from langchain_core.prompts import (
    ChatPromptTemplate, FewShotChatMessagePromptTemplate
)

examples = [
    {"input": "这个退款流程太绕了", "output": "负面 | 体验"},
    {"input": "客服响应很快，赞", "output": "正面 | 效率"},
    {"input": "价格还行", "output": "中性 | 价格"},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是客服评论分析器，按 情感 | 主题 格式输出。"),
    few_shot,                              # 展开为多轮 human/ai 消息
    ("human", "{input}"),
])

chain = final_prompt | llm
print(chain.invoke({"input": "物流等了半个月"}).content)
# 输出类似：负面 | 物流

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

llm = init_chat_model("openai:gpt-4o-mini", temperature=0)

class CodeReview(BaseModel):
    """代码评审结果的结构化契约"""
    severity: str = Field(description="严重程度：critical / warning / info")
    issues: list[str] = Field(description="发现的问题列表，每条一句话")
    suggestion: str = Field(description="最重要的修复建议")
    confidence: float = Field(ge=0, le=1, description="置信度")

reviewer = llm.with_structured_output(CodeReview)

result = reviewer.invoke("""
def div(a, b):
    return a / b
""")

# result 就是 CodeReview 实例
print(result.severity)     # 'critical'
print(result.issues)       # ['未处理除零异常', ...]
print(result.confidence)   # 0.9

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深 DevOps 工程师。"),
    ("human", "{question}"),
])

# | 运算符 = RunnableSequence
chain = prompt | llm | StrOutputParser()

chain.invoke({"question": "如何估算 Kubernetes 资源配额？"})
chain.stream({"question": "..."})     # 逐 token 流式
chain.batch([{"question": "..."}])    # 批量并发
# await chain.ainvoke({...})          # 异步版本同样存在

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

# 字典字面量会被自动转换为 RunnableParallel
rag_chain = (
    {
        "context": retriever | format_docs,      # 分支一：检索 + 格式化
        "question": RunnablePassthrough(),        # 分支二：原样透传
    }
    | prompt
    | llm
    | StrOutputParser()
)

rag_chain.invoke("什么是 LCEL？")

# .assign() 在不吞掉上游输出的前提下追加字段
chain_with_source = rag_chain.assign(
    sources=lambda x: [d.metadata["source"] for d in retriever.invoke(x)]
)

from langchain_core.runnables import RunnableBranch

# 条件路由：RunnableBranch 是 (condition, chain) 列表 + 默认分支
router = RunnableBranch(
    (lambda x: "退款" in x["question"], refund_chain),
    (lambda x: "技术" in x["question"], tech_chain),
    general_chain,                       # 兜底
)

# 重试与降级：让链具备生产韧性
robust_chain = (prompt | llm | parser).with_retry(
    stop_after_attempt=3, retry_if_exception_type=(TimeoutError,)
).with_fallbacks([fallback_chain])

# 查看链的输入输出 schema（调试利器）
print(rag_chain.input_schema.model_json_schema())

from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载：统一产出 Document(page_content, metadata)
docs = PyPDFLoader("handbook.pdf").load()
pages = WebBaseLoader("https://docs.langchain.com").load()

# 切分：递归分隔符，优先按段落、句子断开
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,          # 每块目标字符数
    chunk_overlap=120,       # 块间重叠，保住被切断的语义
    separators=["\n\n", "\n", "。", "，", " ", ""],   # 中文语料建议加中文标点
)
chunks = splitter.split_documents(docs)
print(len(chunks), chunks[0].metadata)   # metadata 保留页码、来源等信息

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 一行完成 向量化 + 入库
vectorstore = FAISS.from_documents(chunks, embeddings)

# 持久化与加载
vectorstore.save_local("faiss_index")
reloaded = FAISS.load_local(
    "faiss_index", embeddings, allow_dangerous_deserialization=True
)

# 相似度检索
hits = vectorstore.similarity_search("如何配置 checkpointer？", k=4)

# MMR 检索：兼顾相关性与多样性，减少返回内容重复
hits = vectorstore.max_marginal_relevance_search(
    "如何配置 checkpointer？", k=4, fetch_k=20
)

retriever = vectorstore.as_retriever(
    search_type="mmr",           # "similarity" | "mmr" | "similarity_score_threshold"
    search_kwargs={"k": 6, "fetch_k": 24},
)

# retriever 本身是 Runnable，可直接进入 LCEL 管道
docs = retriever.invoke("如何配置 checkpointer？")

# 接回上一节的 RAG 链即可运行
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | llm | StrOutputParser()
)

from langchain.tools import tool

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """查询指定城市的实时天气。

    Args:
        city: 城市名称，支持中文与拼音，如 "北京"、"shanghai"。
        unit: 温度单位，celsius 或 fahrenheit，默认 celsius。
    """
    # 真实实现：调用天气 API
    return f"{city}今天晴，26°C"

# 框架自动生成了 schema，检查一下
print(get_weather.name)          # 'get_weather'
print(get_weather.description)   # docstring 的第一段
print(get_weather.args_schema.model_json_schema())

from pydantic import BaseModel, Field
from langchain.tools import tool

class SearchArgs(BaseModel):
    query: str = Field(description="搜索关键词")
    date_range: str = Field(
        default="all",
        description="时间范围：day / week / month / year / all",
    )
    max_results: int = Field(default=5, ge=1, le=20, description="返回条数上限")

@tool(args_schema=SearchArgs)
def web_search(**kwargs) -> str:
    """按关键词搜索互联网并返回结果摘要。"""
    ...

from langchain.messages import HumanMessage

# 1. 绑定：把工具 schema 附加到模型
llm_with_tools = llm.bind_tools([get_weather, web_search])

# 2. 模型决定调用哪个工具、传什么参数
resp = llm_with_tools.invoke([HumanMessage("北京和上海今天多少度？")])
for tc in resp.tool_calls:
    print(tc["name"], tc["args"])
    # get_weather {'city': '北京', 'unit': 'celsius'}
    # get_weather {'city': '上海', 'unit': 'celsius'}

# 3. 执行：直接调用工具对象
result = get_weather.invoke(resp.tool_calls[0]["args"])

# 4. 在 LangGraph 中，ToolNode 负责批量执行并回填 ToolMessage
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([get_weather, web_search])

from typing import Annotated
from langchain_core.tools import InjectedToolArg
from langgraph.prebuilt import InjectedState

@tool
def query_orders(
    state: Annotated[dict, InjectedState()],        # 注入图状态，模型不可见
    user_id: Annotated[str, InjectedToolArg],       # 注入运行时参数，模型不可见
    keyword: str,                                    # 模型可见、模型填写
) -> str:
    """按关键词查询当前用户的订单。"""
    return f"用户 {user_id} 的订单：[{keyword}] 相关共 3 条"

# 模型只看到 keyword 参数；user_id 与 state 由运行时注入

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4o-mini",                # 直接传模型字符串或模型实例
    tools=[get_weather, web_search],
    system_prompt="你是天气与资讯助手，回答时引用数据来源。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "北京今天适合跑步吗？"}]
})
print(result["messages"][-1].content)

from langgraph.checkpoint.memory import MemorySaver

# 持久化：传入 checkpointer 即获得多轮记忆与断点恢复
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_weather],
    checkpointer=MemorySaver(),
)

config = {"configurable": {"thread_id": "user-42"}}
agent.invoke({"messages": [{"role": "user", "content": "我叫林可"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "我叫什么名字？"}]}, config)
# 第二次调用能答出“林可”，记忆由 checkpointer 维护

from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,               # 敏感信息脱敏或拦截
    SummarizationMiddleware,     # 历史过长时自动摘要压缩
    HumanInTheLoopMiddleware,    # 敏感工具调用前暂停等人审批
)

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[read_email, send_email],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            trigger={"tokens": 4000},      # 超过 4k token 触发摘要
        ),
        HumanInTheLoopMiddleware(
            interrupt_on={"send_email": {"allowed_decisions": ["approve", "edit", "reject"]}},
        ),
    ],
)

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2_pt_..."
# 此后所有 invoke / stream 自动上报到 LangSmith

# 给运行打标签，便于按实验 / 客户 / 环境分组分析
result = chain.with_config(
    tags=["rag-v3", "customer-a"],
    metadata={"prompt_version": "2026-08-20"},
).invoke("什么是 LCEL？")

// langgraph.json：声明图入口，dev/build 都读它
{
  "dependencies": ["."],
  "graphs": {
    "assistant": "./app/agent.py:agent"
  },
  "env": ".env"
}

pip install "langgraph-cli[inmem]"
langgraph dev     # http://127.0.0.1:2024，打开即见可视化调试界面
langgraph build   # 生成可部署的容器镜像