LangChain框架之 invoke() 方法

在 LangChain 框架中，invoke 是核心的同步调用方法，用于触发可调用对象（如模型、链条、工具等）的执行并获取结果。它是 LangChain 中“可调用对象（Runnable）”体系的基础方法之一，理解 invoke 及其相关概念对使用 LangChain 至关重要。

一、核心概念：可调用对象（Runnable）

LangChain 中，所有支持 invoke、ainvoke、stream 等方法的对象统称为“可调用对象（Runnable）”。这些对象包括：

• 模型（如 ChatOpenAI、ChatDeepSeek）
• 链条（Chain，如 LLMChain、通过 | 组合的管道链条）
• 提示模板（PromptTemplate、ChatPromptTemplate）
• 输出解析器（StrOutputParser、JsonOutputParser）
• 工具（Tool，如 SerpAPI 搜索工具）

可调用对象的设计目的是统一调用接口，无论对象类型如何，都可以通过相同的方法（如 invoke）触发执行，简化复杂流程的组合（如用 | 拼接链条）。

二、invoke 函数的定义与功能

1. 函数原型

invoke 是 Runnable 类的实例方法，原型可简化为：

def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
    """
    同步调用对象，返回执行结果。
    
    参数：
        input: 输入数据（格式取决于对象类型，通常为字典、字符串或消息列表）
        config: 可选配置（如超时时间、追踪ID等）
    返回：
        执行结果（格式取决于对象类型）
    """

2. 核心功能

• 同步执行：阻塞当前线程，直到对象处理完成并返回结果。
• 输入适配：根据对象类型自动处理不同格式的输入（如字典、消息列表）。
• 流程触发：对于链条（Chain），会按顺序触发内部组件（提示模板→模型→解析器）的执行。

三、常用可调用对象的 invoke 示例

1. 模型（Model）的 invoke

模型（如 ChatOpenAI、ChatDeepSeek）是最基础的可调用对象，invoke 用于直接调用模型生成文本。

示例：调用 ChatOpenAI 模型

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 初始化模型
model = ChatOpenAI(model="gpt-3.5-turbo")

# 输入：消息列表（SystemMessage + HumanMessage）
input_messages = [
    SystemMessage(content="你是一个翻译助手，将中文翻译成英语"),
    HumanMessage(content="我爱编程")
]

# 调用 invoke
result = model.invoke(input_messages)

# 输出结果
print("结果类型：", type(result))
print("结果内容：", result.content)

运行结果：

结果类型： <class 'langchain_core.messages.ai.AIMessage'>
结果内容： I love programming.

结果分析：

• 模型的输入是消息列表（SystemMessage 定义行为，HumanMessage 是用户输入）。
• invoke 返回 AIMessage 对象，其中 content 属性为模型生成的文本。
• 模型会根据消息列表的上下文生成响应，这里完成了中文到英语的翻译。

2. 提示模板（PromptTemplate）的 invoke

提示模板用于将动态参数填充到静态文本中，invoke 用于生成最终的提示内容。

示例：调用 ChatPromptTemplate

from langchain_core.prompts import ChatPromptTemplate

# 定义提示模板（包含系统提示和用户输入变量）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{name}的助手，只说{language}"),
    ("human", "请介绍一下你自己")
])

# 输入：字典（键对应模板中的变量 {name} 和 {language}）
input_params = {
    "name": "程序员",
    "language": "中文"
}

# 调用 invoke
result = prompt.invoke(input_params)

# 输出结果
print("结果类型：", type(result))
print("结果内容：", result)

运行结果：

结果类型： <class 'langchain_core.messages.base.BaseMessageChunk'>
结果内容： [SystemMessage(content='你是程序员的助手，只说中文'), HumanMessage(content='请介绍一下你自己')]

结果分析：

• 提示模板的输入是字典，键必须与模板中的变量（{name}、{language}）对应。
• invoke 返回消息列表（SystemMessage + HumanMessage），已填充变量值。
• 这一步的作用是生成“可直接传给模型”的提示内容，后续可将结果传给模型的 invoke 方法。

3. 链条（Chain）的 invoke

链条是多个组件的组合（如 提示模板 | 模型 | 解析器），invoke 会触发整个链条的执行。

示例：调用组合链条

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

# 1. 定义组件
prompt = ChatPromptTemplate.from_template("计算：{math_expr} = ?")  # 提示模板
model = ChatOpenAI(model="gpt-3.5-turbo")  # 模型
parser = StrOutputParser()  # 输出解析器（提取文本）

# 2. 组合成链条（管道符 | 表示组件串联）
chain = prompt | model | parser

# 3. 输入：字典（包含 {math_expr} 变量）
input_data = {"math_expr": "100 + 200 * 3"}

# 4. 调用 invoke
result = chain.invoke(input_data)

# 输出结果
print("结果类型：", type(result))
print("结果内容：", result)

运行结果：

结果类型： <class 'str'>
结果内容： 700

结果分析：

• 链条的输入是字典，仅需提供最底层的变量（{math_expr}），中间组件的输入由链条自动传递。
• 链条执行流程：
  提示模板填充变量 → 生成消息列表 → 模型处理生成 AIMessage → 解析器提取文本为字符串。
• invoke 直接返回最终结果（字符串），简化了多步骤调用的复杂度。

4. 工具（Tool）的 invoke

工具（如搜索、计算器）是可被 LangChain 调用的外部功能，invoke 用于触发工具执行。

示例：调用 Calculator 工具

from langchain.tools import Calculator

# 初始化工具
calculator = Calculator()

# 输入：工具调用参数（字符串形式的表达式）
input_query = "100 + 200 * 3"

# 调用 invoke
result = calculator.invoke(input_query)

# 输出结果
print("结果类型：", type(result))
print("结果内容：", result)

运行结果：

结果类型： <class 'str'>
结果内容： 700

结果分析：

• 工具的输入通常是字符串（工具可理解的指令）。
• invoke 直接返回工具的执行结果（这里计算器计算了表达式的值）。
• 工具常用于 Agent 中，帮助模型解决数学计算、实时信息查询等问题。

四、invoke 与其他调用方法的对比

除 invoke 外，LangChain 还提供其他调用方法，适用于不同场景：

方法	功能	适用场景	示例代码
invoke	同步调用，返回完整结果	简单场景，需要立即获取结果	result = chain.invoke(input)
ainvoke	异步调用（协程），返回完整结果	异步程序（如 FastAPI）	result = await chain.ainvoke(input)
stream	流式返回结果（逐段生成）	实时展示（如聊天界面）	for chunk in chain.stream(input): print(chunk)
batch	批量处理多个输入	批量任务，提高效率	results = chain.batch([input1, input2])

示例：stream 与 invoke 的对比

# 流式调用（stream）
for chunk in chain.stream({"math_expr": "10 + 20"}):
    print(chunk, end="", flush=True)  # 逐段输出：30

print("\n---")

# 同步调用（invoke）
print(chain.invoke({"math_expr": "10 + 20"}))  # 直接输出：30

运行结果：

30
---
30

差异分析：

• stream 适合需要“打字机效果”的场景（如聊天机器人实时显示回复）。
• invoke 适合需要一次性获取完整结果的场景（如后台数据处理）。

五、invoke 中的配置参数（config）

invoke 的第二个参数 config 用于传递额外配置，如超时时间、追踪ID等。常用配置项：

• timeout：超时时间（秒），超过时间则终止调用。
• tags：标签，用于追踪和日志分类。

示例：设置超时和标签

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-3.5-turbo")

# 配置：超时5秒，添加标签
config = {
    "timeout": 5,
    "tags": ["translation", "test"]
}

result = model.invoke(
    [HumanMessage(content="翻译：你好")],
    config=config
)

print(result.content)  # 输出：Hello

六、总结

1. invoke 是 LangChain 中可调用对象的核心同步方法，用于触发执行并返回结果。
2. 输入格式：根据对象类型不同，输入可以是字典（提示模板、链条）、消息列表（模型）或字符串（工具）。
3. 核心作用：简化复杂流程调用，统一接口，支持组件组合（如链条）。
4. 相关方法：ainvoke（异步）、stream（流式）、batch（批量），覆盖不同场景需求。

ntent) # 输出：Hello

### 六、总结
1. **`invoke` 是 LangChain 中可调用对象的核心同步方法**，用于触发执行并返回结果。
2. **输入格式**：根据对象类型不同，输入可以是字典（提示模板、链条）、消息列表（模型）或字符串（工具）。
3. **核心作用**：简化复杂流程调用，统一接口，支持组件组合（如链条）。
4. **相关方法**：`ainvoke`（异步）、`stream`（流式）、`batch`（批量），覆盖不同场景需求。

掌握 `invoke` 的使用，能帮助你灵活组合 LangChain 的各种组件，构建从简单查询到复杂Agent的各类LLM应用。