LangChain模块详解

赤狐尼克

383人浏览 · 2026-05-01 14:59:35

赤狐尼克 · 2026-05-01 14:59:35 发布

langchain_openai ------------------ ChatOpenAI

langchain_core.prompts------------------ChatPromptTemplate 和 MessagesPlaceholder

langchain_core.runnables.history--------------------RunnableWithMessageHistory

langchain_core.chat_history--BaseChatMessageHistory和InMemoryChatMessageHistory

langchain_core.tools------------------------------tool

langchain.agents--------------------------------------AgentExecutor和create_tool_calling_agent

langchain_community.utilities.google_serper----------GoogleSerperAPIWrapper

把这些模块的作用、核心参数、代码示例、以及它们之间的配合关系讲清楚，帮你建立完整的认知体系。

整体架构概览

这些模块共同构成了一个带记忆、能调用工具的智能 Agent，流程如下：

ChatOpenAI：负责思考和生成文本
ChatPromptTemplate + MessagesPlaceholder：负责组织提示词，管理对话历史和工具调用过程
BaseChatMessageHistory + InMemoryChatMessageHistory：负责存储对话记忆
RunnableWithMessageHistory：负责把记忆自动注入到链里
tool：定义 Agent 能调用的工具
create_tool_calling_agent + AgentExecutor：把 LLM、工具、提示词组装成 Agent
GoogleSerperAPIWrapper：封装搜索引擎，作为工具供 Agent 调用

1. `langchain_openai.ChatOpenAI`：大语言模型封装

作用

这是对 OpenAI 兼容接口的 LLM 的统一封装，不仅可以调用 OpenAI 的 GPT，还可以调用通义千问、智谱 AI 等兼容 OpenAI 接口的模型。它负责接收提示词，生成文本回复。

核心参数

参数	说明	示例
`model`	模型名称	`"gpt-4o"`、`"qwen-max"`
`api_key`	API 密钥	`os.getenv("DASHSCOPE_API_KEY")`
`base_url`	API 地址（兼容第三方模型时必填）	`"https://dashscope.aliyuncs.com/compatible-mode/v1"`
`temperature`	采样温度，0~1，越小越稳定	`0.1`
`max_tokens`	最大生成 token 数	`2048`

代码示例

from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.1
)

# 直接调用
response = llm.invoke("你好")
print(response.content)

2. `langchain_core.prompts.ChatPromptTemplate` & `MessagesPlaceholder`：提示词模板

作用

ChatPromptTemplate：构建对话式提示词，支持系统提示、用户输入、历史消息等多种消息类型。
MessagesPlaceholder：占位符，用于在提示词中预留位置，动态插入对话历史或Agent 的工具调用过程（agent_scratchpad）。

核心参数

`ChatPromptTemplate.from_messages()`

接收一个消息列表，支持以下类型：

("system", "系统提示词")：设定 AI 的角色和规则
("user", "{input}")：用户输入，用 {} 占位
MessagesPlaceholder(variable_name="chat_history")：插入对话历史
MessagesPlaceholder(variable_name="agent_scratchpad")：插入 Agent 的工具调用日志（Agent 专用）

代码示例（Agent 专用提示词）

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个强大的智能助手，能通过调用工具解决问题。"),
    # 插入对话历史，变量名必须和RunnableWithMessageHistory里的history_messages_key一致
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{input}"),
    # 插入Agent的工具调用过程，create_tool_calling_agent强制要求这个变量名
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

3. `langchain_core.chat_history.BaseChatMessageHistory` & `InMemoryChatMessageHistory`：对话记忆存储

作用

BaseChatMessageHistory：抽象基类，定义了对话记忆的接口（添加消息、清空消息等）。你可以继承它，实现自定义的记忆存储（比如存到 Redis、数据库）。
InMemoryChatMessageHistory：BaseChatMessageHistory 的默认实现，把记忆存在内存里，程序重启后记忆会丢失，适合开发测试。

核心方法

表格

方法	说明
`add_message(message)`	添加一条消息
`add_user_message(content)`	添加一条用户消息
`add_ai_message(content)`	添加一条 AI 消息
`clear()`	清空记忆

代码示例

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

# 初始化记忆
history = InMemoryChatMessageHistory()

# 添加消息
history.add_user_message("你好")
history.add_ai_message("你好，有什么可以帮你的？")

# 查看消息
print(history.messages)
# 输出: [HumanMessage(content='你好'), AIMessage(content='你好，有什么可以帮你的？')]

4. `langchain_core.runnables.history.RunnableWithMessageHistory`：给链注入记忆

作用

它是一个包装器，把你的链（LLM 链或 Agent 链）和记忆存储结合起来，自动在每次调用时加载历史消息，调用完成后保存新消息。

核心参数

参数	说明
`runnable`	要包装的链（比如 `agent_executor` 或 `prompt
`get_session_history`	一个函数，输入 `session_id`，返回对应的 `BaseChatMessageHistory` 对象
`input_messages_key`	输入中用户消息的 key（比如 `"input"`）
`history_messages_key`	提示词中历史消息占位符的变量名（比如 `"chat_history"`）
`output_messages_key`	输出中 AI 消息的 key（如果输出是字典，需要指定）

代码示例

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory

# 记忆存储字典
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """根据session_id获取对应的记忆对象"""
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 假设 agent_executor 是你已经创建好的Agent
agent_with_memory = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# 调用时传入 session_id，区分不同用户的对话
result = agent_with_memory.invoke(
    {"input": "你好"},
    config={"configurable": {"session_id": "user_001"}}
)

5. `langchain_core.tools.tool`：定义工具

作用

一个装饰器，把普通的 Python 函数包装成 LangChain 的 Tool 对象，供 Agent 调用。

核心要点

函数必须有清晰的文档字符串：说明工具的用途、参数格式，这会作为提示词的一部分发给 LLM，LLM 会根据文档字符串决定是否调用工具。
参数必须是简单类型（比如 str、int），不能是复杂对象。

代码示例

from langchain_core.tools import tool
import numexpr

@tool
def calculator(expression: str) -> str:
    """
    【数学计算专用】安全计算器，用于解决数学计算问题
    支持：加 减 乘 除 括号 小数 指数运算
    输入格式：纯数学表达式，不要有中文，比如 “35*23+18”、“100/(2+3)”
    """
    try:
        result = numexpr.evaluate(expression).item()
        return f'计算结果： {result}'
    except Exception as e:
        return f'计算错误：{str(e)}'

# 查看工具信息
print(calculator.name)
print(calculator.description)

6. `langchain.agents.AgentExecutor` & `create_tool_calling_agent`：构建 Agent

作用

create_tool_calling_agent：创建一个工具调用型 Agent，它会把 LLM、工具、提示词组装起来，负责决定什么时候调用工具、调用什么工具、怎么解析工具返回结果。
AgentExecutor：Agent 的执行器，负责循环执行 Agent 的思考 - 行动 - 观察流程，直到任务完成或达到最大迭代次数。

核心参数

`create_tool_calling_agent(llm, tools, prompt)`

表格

参数	说明
`llm`	`ChatOpenAI` 对象
`tools`	工具列表（比如 `[calculator, web_search]`）
`prompt`	`ChatPromptTemplate` 对象，必须包含 `agent_scratchpad` 占位符

`AgentExecutor(agent, tools, ...)`

表格

参数	说明
`agent`	`create_tool_calling_agent` 返回的 Agent
`tools`	工具列表（和上面一致）
`verbose`	是否打印详细日志（建议设为 `True`，方便调试）
`handle_parsing_errors`	是否自动处理解析错误（建议设为 `True`）
`max_iterations`	最大迭代次数，防止 Agent 无限循环

代码示例

from langchain.agents import AgentExecutor, create_tool_calling_agent

# 假设 llm、tools、prompt 都已经定义好
agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10,
)

# 调用Agent
result = agent_executor.invoke({"input": "2026年世界杯举办地的人口乘以3.14是多少？"})
print(result["output"])

7. `langchain_community.utilities.google_serper.GoogleSerperAPIWrapper`：搜索引擎封装

作用

封装了 Serper API（Google 搜索的第三方 API），负责执行搜索，返回结果。通常和 tool 装饰器配合，包装成 Agent 能调用的工具。

核心参数

表格

参数	说明
`k`	返回结果的数量
`serper_api_key`	Serper API Key（也可以通过环境变量 `SERPER_API_KEY` 设置）

代码示例

python

运行

from langchain_community.utilities.google_serper import GoogleSerperAPIWrapper
from langchain_core.tools import tool
import os

os.environ["SERPER_API_KEY"] = "你的Serper API Key"

# 初始化包装器
search_wrapper = GoogleSerperAPIWrapper(k=5)

# 包装成工具
@tool
def web_search(query: str) -> str:
    """
    【实时信息/知识查询专用】联网搜索引擎
    只要满足以下任意一个条件，必须调用此工具
    1. 用户查询的实时信息（如：今天天气，最新新闻）
    2. 你不确定的答案，或者你的知识已经过时
    """
    return search_wrapper.run(query)

# 测试工具
print(web_search.invoke("今天北京天气"))

完整的 Agent 组装流程（把所有模块串起来）

python

运行

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_community.utilities.google_serper import GoogleSerperAPIWrapper
import numexpr

# ================== 1. 环境配置 ==================
os.environ["DASHSCOPE_API_KEY"] = "你的阿里云API Key"
os.environ["SERPER_API_KEY"] = "你的Serper API Key"

# ================== 2. 初始化LLM ==================
llm = ChatOpenAI(
    model="qwen-max",
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.1
)

# ================== 3. 定义工具 ==================
@tool
def calculator(expression: str) -> str:
    """
    【数学计算专用】安全计算器，用于解决数学计算问题
    支持：加 减 乘 除 括号 小数 指数运算
    输入格式：纯数学表达式，不要有中文，比如 “35*23+18”
    """
    try:
        result = numexpr.evaluate(expression).item()
        return f'计算结果： {result}'
    except Exception as e:
        return f'计算错误：{str(e)}'

search_wrapper = GoogleSerperAPIWrapper(k=5)
@tool
def web_search(query: str) -> str:
    """
    【实时信息/知识查询专用】联网搜索引擎
    只要满足以下任意一个条件，必须调用此工具
    1. 用户查询的实时信息（如：今天天气，最新新闻）
    2. 你不确定的答案，或者你的知识已经过时
    """
    return search_wrapper.run(query)

tools = [calculator, web_search]

# ================== 4. 构建提示词 ==================
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个强大的智能助手，能通过调用工具解决问题。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# ================== 5. 构建Agent ==================
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10,
)

# ================== 6. 配置记忆 ==================
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

agent_with_memory = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# ================== 7. 运行Agent ==================
if __name__ == "__main__":
    result = agent_with_memory.invoke(
        {"input": "2026年世界杯举办地的人口是多少？乘以3.14是多少？"},
        config={"configurable": {"session_id": "user_001"}}
    )
    print(f"\n🤖 最终回答：{result['output']}")