🔗 LangChain 框架

什么是 LangChain？

LangChain 是由 Harrison Chase 于 2022 年 10 月发起的开源框架，旨在为 LLM 应用开发提供标准化的抽象层和工具链。它解决了一个核心问题：如何将 LLM 的能力与外部数据源、API、记忆系统和工具无缝集成。

截至 2026 年初，LangChain 已成为 LLM 应用开发领域最流行的框架之一，GitHub Stars 超过 100K，npm/PyPI 月下载量数百万。

发展历程

• 2022.10 — Harrison Chase 发布 LangChain 首个版本（Python）
• 2023.01 — 获得 Benchmark 领投的 1000 万美元种子轮融资
• 2023.04 — LangChain.js 发布，正式支持 TypeScript/JavaScript 生态
• 2023.07 — LangSmith（调试与监控平台）发布
• 2024.01 — LangGraph 发布，引入基于图的 Agent 编排
• 2024.06 — v0.2 重大重构，模块化拆分为 langchain-core + 独立包
• 2025.02 — v0.3 发布，LCEL（LangChain Expression Language）成为一等公民

六大核心模块

LCEL — LangChain Expression Language

LCEL 是 LangChain v0.2+ 引入的声明式编排语法，使用管道符 | 将组件串联：

# LCEL 基本语法：用 | 管道符组合组件
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("用一句话解释 {topic}")
model = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

# 管道编排：Prompt → Model → Parser
chain = prompt | model | parser

# 调用
result = chain.invoke({"topic": "量子计算"})
print(result)  # "量子计算利用量子力学原理..."

三层模型抽象

Prompt Templates

LangChain 提供类型安全的提示词模板系统：

from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    HumanMessagePromptTemplate,
    SystemMessagePromptTemplate,
)

prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template(
        "你是 {role}，说话风格 {style}"
    ),
    MessagesPlaceholder(variable_name="history"),
    HumanMessagePromptTemplate.from_template("{input}"),
])

# 支持部分填充
partial_prompt = prompt.partial(role="AI 导师", style="简洁幽默")

Output Parsers

将 LLM 非结构化输出解析为结构化数据：

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class Movie(BaseModel):
    title: str = Field(description="电影名称")
    year: int = Field(description="上映年份")
    genre: str = Field(description="类型")

parser = JsonOutputParser(pydantic_object=Movie)

# 自动将 format_instructions 注入提示词
chain = prompt | model | parser
movie = chain.invoke({"query": "推荐一部科幻电影"})
# {"title": "银翼杀手2049", "year": 2017, "genre": "科幻"}

什么是 Chain？

Chain 是 LangChain 的核心编排单元——将多个操作步骤串联成一个可复用的工作流。在 LCEL 之前，LangChain 使用 LLMChain、SequentialChain 等类；LCEL 之后，所有编排统一为 Runnable 管道。

常用 Chain 模式

1. 顺序 Chain（Sequential）

# 先翻译，再总结
translate = ChatPromptTemplate.from_template(
    "将以下内容翻译成英文：\n{text}"
) | model | StrOutputParser()

summarize = ChatPromptTemplate.from_template(
    "Summarize in 2 sentences：\n{english}"
) | model | StrOutputParser()

# 用 RunnablePassthrough 传递中间结果
from langchain_core.runnables import RunnablePassthrough
chain = (
    {"english": translate}
    | summarize
)

2. 并行 Chain（Parallel）

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    summary=summarize_chain,
    sentiment=sentiment_chain,
    keywords=keyword_chain,
)

# 三个子链并行执行，返回 dict
result = chain.invoke({"text": article})

3. 路由 Chain（Router）

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "代码" in x["topic"], code_chain),
    (lambda x: "数学" in x["topic"], math_chain),
    default_chain,  # 兜底
)

为什么需要 Memory？

LLM 本身是无状态的——每次调用都是独立的。Memory 组件让对话系统具备"记忆"能力，能在多轮对话中保持上下文连贯。

四种主要 Memory 策略

代码示例：Summary Memory

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True
)

# 添加对话记录
memory.save_context(
    {"input": "我想学 AI，从哪里开始？"},
    {"output": "建议从 Python 和线性代数开始..."}
)

# 获取压缩摘要
messages = memory.load_memory_variables({})
# → "用户想学AI，助手建议从Python和数学基础开始"

Agent 的核心思想

Agent 是 LangChain 最强大的特性之一。与 Chain 的预设路径不同，Agent 让 LLM 自主决定执行哪些步骤、调用哪些工具。LLM 充当"大脑"，通过推理循环动态规划行动。

Agent 执行流程

创建 Agent 的两种方式

方式一：使用 create_tool_calling_agent（推荐）

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

tools = [search_tool, calculator_tool]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_tool_calling_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=tools,
    prompt=prompt
)

executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "今天北京天气怎么样？"})

方式二：使用 LangGraph（更灵活）

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=tools,
)

# LangGraph Agent 支持流式输出
for chunk in agent.stream(
    {"messages": [("human", "帮我搜索 LangChain 最新版本")]}
):
    print(chunk)

内置工具示例

自定义 Tool

from langchain_core.tools import tool

@tool
def get_stock_price(symbol: str) -> str:
    """查询股票实时价格。
    
    Args:
        symbol: 股票代码，例如 AAPL、TSLA
    """
    # 调用股票 API...
    return f"{symbol} 当前价格: $150.00"

# 方式二：使用 StructuredTool
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位")

weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="get_weather",
    description="获取指定城市天气",
    args_schema=WeatherInput,
)

Toolkits — 预打包工具集

项目目标

构建一个具有以下能力的智能客服系统：

• 基于产品文档的 RAG 知识库问答
• 多轮对话记忆
• 查询订单工具调用
• 流式输出与打字机效果

完整代码

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.tools import tool

# 1. 初始化模型和 Embedding
llm = ChatOpenAI(model="gpt-4o", streaming=True)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 2. 构建知识库
docs = load_product_docs()  # 加载产品文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50
)
chunks = splitter.split_documents(docs)
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. 定义工具
@tool
def query_order(order_id: str) -> str:
    """查询订单状态。Args: order_id: 订单编号"""
    return db.get_order(order_id)

# 4. RAG + Agent 融合
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是产品客服。优先用以下知识库回答：
{context}
如需查询订单，使用 query_order 工具。"""),
    MessagesPlaceholder("history", optional=True),
    ("human", "{input}"),
])

# 5. 组装 Chain
chain = (
    {
        "context": retriever | format_docs,
        "input": RunnablePassthrough(),
        "history": lambda x: get_history(x["session_id"])
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 6. 流式输出
for chunk in chain.stream({"input": "我的订单 #12345 到哪了？"}):
    print(chunk, end="")