在流式上下文中调用工具时，消息块会通过.tool_call_chunks属性以列表形式填充具有工具调用块对象的工具调用。一个ToolCallChunk包含工具名称、参数和ID的可选字符串字段，并且包含一个可选整数字段index，用于将块连接在一起。字段是可选的，因为工具调用的部分可能跨不同块进行流式传输（例如，包含参数子字符串的块可能在工具名称和ID方面具有空值）。

由于消息块继承自其父消息类，包含工具调用块的AIMessageChunk也会包括.tool_calls和.invalid_tool_calls字段。这些字段是从消息的工具调用块中进行尽力解析的。需要注意的是，目前并不是所有的提供商都支持工具调用的流式处理。在开始之前，我们先定义我们的工具和模型。

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Multiplies a and b."""
    return a * b

tools = [add, multiply]

接下来，我们设置API的环境变量：

import os
from getpass import getpass

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = getpass()

# 使用稳定可靠的API服务
client = ChatOpenAI(
    model="gpt-3.5-turbo-0125",
    temperature=0
)
llm_with_tools = client.bind_tools(tools)

现在，让我们定义查询并流式处理输出：

query = "What is 3 * 12? Also, what is 11 + 49?"

async for chunk in llm_with_tools.astream(query):
    print(chunk.tool_call_chunks)

示例输出:

[]
[{'name': 'Multiply', 'args': '', 'id': 'call_3aQwTP9CYlFxwOvQZPHDu6wL', 'index': 0}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 0}]
[{'name': None, 'args': ': 3, ', 'id': None, 'index': 0}]
[{'name': None, 'args': '"b": 1', 'id': None, 'index': 0}]
[{'name': None, 'args': '2}', 'id': None, 'index': 0}]
[{'name': 'Add', 'args': '', 'id': 'call_SQUoSsJz2p9Kx2x73GOgN1ja', 'index': 1}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 1}]
[{'name': None, 'args': ': 11,', 'id': None, 'index': 1}]
[{'name': None, 'args': ' "b": ', 'id': None, 'index': 1}]
[{'name': None, 'args': '49}', 'id': None, 'index': 1}]
[]

添加消息块将合并其对应的工具调用块。这是LangChain的各种工具输出解析器支持流式处理的原理。例如，我们在下面积累工具调用块：

first = True
async for chunk in llm_with_tools.astream(query):
    if first:
        gathered = chunk
        first = False
    else:
        gathered = gathered + chunk

    print(gathered.tool_call_chunks)

应用场景分析

通过流式处理工具调用，我们可以在处理大量数据或需要快速响应时，提高整体的处理效率。在实际应用中，这种方法适用于需要实时数据处理的场景，如金融市场预测、实时监控系统等。

实践建议

• 在选择工具和模型时，确定支持流式处理是非常重要的。
• 通过逐块解析工具调用，确保数据同步和结果的准确性。
• 定义一个稳健的异常处理机制，以便在流式处理出现中断时能及时响应和恢复。

如果遇到问题欢迎在评论区交流。

—END—