LangChain主要组件 • lin

一个完整的 LangChain 应用通常需要多个组件协同工作。LangChain 主要提供以下 6 种核心组件：

Models（模型）：集成各类大语言模型及嵌入模型，并提供统一调用接口。
Prompts（提示词）：管理、优化和序列化提示词，支持零样本（zero-shot）和少样本（few-shot）等模板。
Memory（记忆）：保存与模型交互的上下文状态，实现多轮对话的记忆功能。
Indexes（索引）：结构化外部文档，便于与模型交互，支持文档加载、分割、向量存储和检索。
Chains（链）：将多个组件串联成端到端的调用流程，简化复杂应用的构建。
Agents（代理）：根据用户输入动态决定调用哪些工具（如搜索、计算等），并循环执行直至完成任务。

1.1 Models（模型）#

当前市场上的大语言模型种类繁多，LangChain 的模型组件提供了与各种模型的集成，并为所有模型提供精简的统一接口，使得开发者可以方便地切换和调用不同厂商的模型。

LangChain 支持三类模型：

LLMs（大语言模型）
接收文本字符串作为输入，返回文本字符串。适用于通用文本生成、问答等任务。
Chat Models（聊天模型）
基于 LLM，但输入输出均为特定格式的“聊天消息”（如 HumanMessage、AIMessage 等），更适配对话场景。
Embeddings Models（嵌入模型）
接收文本，输出浮点数向量（即文本的向量表示），常用于语义搜索、相似度计算等任务。

开发者应根据实际应用场景选择合适的模型类型。

1.1.1 LLMs（大语言模型）#

LLM 是应用最广泛的模型类型。常用的开源模型可从 Hugging Face ↗ 获取。以下以阿里云通义千问（qwen-max）为例，展示如何使用 LangChain 调用 LLM。

第一步：安装必要依赖包

pip install langchain langchain-openai

bash

注意：使用 OpenAI 风格的接口需先开通百炼平台服务，并获取 API Key。

第二步：编写调用代码

import os
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL")
)

# 流式输出
for chunk in llm.stream("你是什么模型"):
    print(chunk.content, end="")

python

1.1.2 Chat Models（聊天模型）#

聊天模型处理的消息类型包括：

HumanMessage：用户发送的消息。
AIMessage：模型返回的回复。
SystemMessage：用于设定模型角色或行为指令（如“你是一位编程专家”）。
ChatMessage：通用消息类型，但建议优先使用上述三种。

示例：多轮对话

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import os

llm = ChatOpenAI(
    base_url=os.getenv("BASE_URL"),
    api_key=os.getenv("API_KEY"),
    model="qwen-max"
)

messages = [
    HumanMessage("告诉我有哪些一夜暴富的方法？"),
    AIMessage("年轻人要脚踏实地。"),
    HumanMessage("我现在等不及了，直接告诉我方法？"),
    AIMessage("你太急了，先去工作吧。"),
    HumanMessage("我刚刚问了几个问题了？")
]

response = llm.invoke(messages)
print(response.content)

python

1.1.3 Embeddings Models（嵌入模型）#

嵌入模型将文本转化为浮点数向量，便于在向量空间中进行语义计算。常用场景包括语义搜索、文本聚类等。

示例：使用百炼平台的嵌入模型

from langchain_community.embeddings import DashScopeEmbeddings
import os

embedding_model = DashScopeEmbeddings(
    dashscope_api_key=os.getenv('API_KEY'),
    model="text-embedding-v3",
)

# 单条文本向量化
print(embedding_model.embed_query("AI好啊，得学啊"))

# 批量文本向量化
print(embedding_model.embed_documents(["AI好啊，得学啊", "hello world"]))

python

LangChain 还集成了多种嵌入模型服务，如 AzureOpenAI、百度千帆、Hugging Face Hub、OpenAI、Llama-cpp、SentenceTransformers 等。

1.2 Prompts（提示词）#

提示词（Prompt）是用户输入给模型的信息，其设计质量直接影响模型输出效果。LangChain 提供 PromptTemplate 组件，便于构建可复用的提示模板，支持零样本（zero-shot）和少样本（few-shot）两种常见方式。

1.2.1 零样本（Zero-shot）提示模板#

直接根据模板生成提示，无需示例。

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    base_url=os.getenv("BASE_URL"),
    api_key=os.getenv("API_KEY"),
    model="qwen-max"
)

prompt = PromptTemplate.from_template("我的邻居姓{lastname}，他生了个儿子，给他儿子起一个名字。")
prompt_text = prompt.format_prompt(lastname="张")
print(llm.invoke(prompt_text))

python

1.2.2 少样本（Few-shot）提示模板#

通过提供少量示例，引导模型学习任务模式。例如，根据若干组“单词-反义词”示例，让模型为新词生成反义词：

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_openai import ChatOpenAI
import os

examples = [
    {"word": "开心", "antonym": "难过"},
    {"word": "高", "antonym": "矮"},
    {"word": "胖", "antonym": "瘦"},
]

example_prompt = PromptTemplate(
    input_variables=["word", "antonym"],
    template="单词: {word}\n反义词: {antonym}\n"
)

few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="给出每个单词的反义词，直接输出答案。",
    suffix="单词: {input}\n反义词:",
    input_variables=["input"],
    example_separator="\n"
)

prompt_text = few_shot_prompt.format(input="夯")
llm = ChatOpenAI(
    model="qwen3-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL"),
    extra_body={"enable_thinking": False}
)
print(llm.invoke(prompt_text))

python

1.3 Chains（链）#

链（Chain）将多个组件（如提示模板、模型、输出解析器等）串联成一个处理管道，实现复杂的业务流程。LangChain 支持使用管道操作符 | 快速构建链。

示例 1：单链（提示 + 模型）

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    model="qwen-max",
    base_url=os.getenv("BASE_URL")
)

prompt = PromptTemplate(
    template="我的邻居姓{lastname}，他生了个儿子，给他起3个最好听的名字。",
    input_variables=["lastname"]
)

chain = prompt | llm
print(chain.invoke({"lastname": "张"}).content)

python

示例 2：多链串联（提示 → 模型 → 提示 → 模型 → 解析器）

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os

llm = init_chat_model(
    model="qwen3-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL"),
    model_provider="openai"
)

first_prompt = PromptTemplate.from_template("我的邻居姓{lastname}，他生了个儿子，给他起个名字。")
second_prompt = PromptTemplate.from_template(
    "邻居的儿子名字叫{child_name}，给他起个小名，输出大名和推荐的小名。"
)

chain = first_prompt | llm | second_prompt | llm | StrOutputParser()
output = chain.invoke({"lastname": "孙"})
print(output)

python

1.4 Agents（代理）#

代理（Agent）的核心思想是让 LLM 自主决定调用哪些外部工具（如搜索引擎、计算器、文件读写等）来完成任务。由于 LLM 本身存在无法获取实时信息、数学计算能力弱等局限性，借助工具可以大幅拓展其能力。

示例 1：使用 DuckDuckGo 搜索引擎查询实时信息

import os
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_community.tools import DuckDuckGoSearchRun

ddg_search = DuckDuckGoSearchRun()
llm = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    model="qwen3-max",
    base_url=os.getenv("BASE_URL"),
    extra_body={"enable_thinking": False}
)

agent = create_agent(
    model=llm,
    tools=[ddg_search],
    system_prompt="你是一个有用的助手，根据用户输入选择合适工具来回答问题。"
)

response = agent.invoke({
    "messages": [{"role": "user", "content": "中国目前有多少人口？"}]
})
for msg in response["messages"]:
    print(msg)

python

示例 2：自定义工具（使用 @tool 装饰器）

from langchain.tools import tool
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
import os
import requests

@tool
def write_file(file_path: str, content: str):
    """将内容写入指定文件路径。"""
    with open(file_path, "w") as f:
        f.write(content)
    return f"文件 {file_path} 写入成功。"

@tool
def read_file(file_path: str):
    """读取本地文件内容。"""
    with open(file_path, "r") as f:
        return f.read()

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。"""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """计算两个整数的和。"""
    return a + b

@tool
def get_weather(city: str):
    """查询城市天气（模拟）。"""
    # 此处可替换为真实 API
    return f"{city} 天气晴朗，气温 25℃"

llm = ChatOpenAI(
    model="qwen3-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL")
)

tools = [get_weather, add, multiply, write_file, read_file]
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="根据用户需求决定是否调用工具。"
)

# 流式执行
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "帮我算 5*6，然后查一下深圳的天气。"}]
}):
    print(chunk)

python

1.5 Memory（记忆）#

大语言模型本身是无状态的，每次调用独立处理输入。为了实现多轮对话的上下文记忆，LangChain 提供了 Memory 组件，支持短期记忆（会话内）和长期记忆（跨会话，需持久化存储）。

1.5.1 使用 ChatMessageHistory#

ChatMessageHistory 是一个简单的内存型历史记录容器，可手动添加消息并传递给模型。

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL")
)

history = ChatMessageHistory()
history.add_user_message("你能做什么？")
history.add_ai_message("我能回答各种问题。")
history.add_user_message("小明有3个苹果和4个李子，他一共有几个水果？")
history.add_ai_message("小明一共有7个水果。")
history.add_user_message("我一共问了几个问题？")

print(history.messages)  # 查看历史
print(llm.invoke(history.messages).content)

python

1.5.2 手动维护消息列表#

也可以直接使用 HumanMessage 和 AIMessage 列表实现多轮对话。

from langchain.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="qwen3-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL"),
    extra_body={"enable_thinking": False}
)

messages = [
    HumanMessage("你好"),
    AIMessage("你好，有什么可以帮你？"),
    HumanMessage("LangChain 是什么？"),
    AIMessage("LangChain 是一个开源 LLM 应用开发框架。"),
    HumanMessage("我问了几个问题了？")
]

response = llm.invoke(messages)
print(response.content)

python

1.5.3 使用 InMemorySaver（检查点）#

利用 InMemorySaver 可以在 Agent 执行过程中自动保存对话状态，实现会话内记忆。

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL")
)

agent = create_agent(
    model=llm,
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [{"role": "user", "content": "你能做什么？"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "小明有3个苹果和4个李子，他一共有几个水果？"}]}, config=config)
result = agent.invoke({"messages": [{"role": "user", "content": "我问了几个问题了？"}]}, config=config)
print(result['messages'][-1].content)

python

1.5.4 使用 MySQL 实现长期记忆#

对于跨会话的长期记忆，可将对话历史持久化到数据库中。以下示例使用 PyMySQLSaver（需提前安装 pymysql 和 langgraph-checkpoint-mysql）。

from langchain.agents import create_agent
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
from langchain_openai import ChatOpenAI
import os
import uuid

llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.getenv('API_KEY'),
    base_url=os.getenv("BASE_URL")
)

DB_URI = f"mysql+pymysql://root:{os.getenv('PASSWORD')}@localhost:3306/langchain_db"

with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 自动创建所需数据表
    agent = create_agent(llm, tools=[], checkpointer=checkpointer)
    config = {"configurable": {"thread_id": str(uuid.uuid4())}}
    agent.invoke({"messages": [{"role": "user", "content": "你能做什么？"}]}, config=config)
    agent.invoke({"messages": [{"role": "user", "content": "小明有3个苹果和4个李子，他一共有几个水果？"}]}, config=config)
    result = agent.invoke({"messages": [{"role": "user", "content": "我问了几个问题了？"}]}, config=config)
    print(result['messages'][-1].content)

python

1.6 Indexes（索引）#

Indexes 组件使 LangChain 能够处理外部文档数据，支持文档加载、分割、向量化存储和检索。这一系列能力为构建 RAG（检索增强生成）应用奠定了基础。

1.6.1 文档加载器（Document Loaders）#

文档加载器负责将各种格式的文件（如 TXT、PDF、Markdown、HTML 等）转换为统一的 Document 对象。LangChain 基于 Unstructured 包实现多格式支持。

示例：加载文本文件

from langchain_unstructured import UnstructuredLoader
from langchain_community.document_loaders import TextLoader

# 使用 UnstructuredLoader（支持多种格式）
loader = UnstructuredLoader('../data/衣服属性.txt', encoding='utf8')
docs = loader.load()
print(docs[0].page_content[:50])

# 使用专用 TextLoader
loader = TextLoader('../data/衣服属性.txt', encoding='utf8')
docs = loader.load()

python