03、LangChain4j的stream流式tool前后机制调用

概述

本文档主要介绍 LangChain4j 中流式（Stream）调用场景下，Tool 工具的调用机制及其前后钩子（Hook）的处理流程。

Tool 调用触发时机

调用链路追踪

Tool 调用在流式场景中的触发时机如下：

QwenStreamingChatModel#generateByNonMultimodalModel
  └─ QwenStreamingChatModel#chat [StreamingChatResponseHandler#onCompleteResponse]
       └─ 透传到 LangChain4j 中的自定义 Handler
            └─ AiServiceStreamingResponseHandler#onCompleteResponse

调用流程示意图

ToolExecutor 执行逻辑

onCompleteResponse 方法

Tool 的实际执行逻辑位于 AiServiceStreamingResponseHandler#onCompleteResponse 方法中：

public void onCompleteResponse(ChatResponse chatResponse) {
    // ... 省略部分代码

    if (toolExecutor != null) {
        // 异步执行模式下获取结果
        for (CompletableFuture<ToolExecutionResultMessage> toolResultFuture : toolResultFutures) {
            try {
                ToolExecutionResultMessage toolExecutionResultMessage = toolResultFuture.get();
                addToMemory(toolExecutionResultMessage);
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
    } else {
        // 同步执行模式
        for (ToolExecutionRequest toolExecutionRequest : aiMessage.toolExecutionRequests()) {
            // 本质就是找到指定的ToolExecutor来进行任务执行
            String toolResult = execute(toolExecutionRequest);
            addToMemory(ToolExecutionResultMessage.from(toolExecutionRequest, toolResult));
        }
    }

    // 构建新的请求，继续对话
    ChatRequest chatRequest = ChatRequest.builder()
            .messages(messagesToSend(memoryId))
            .toolSpecifications(toolSpecifications)
            .build();

    // 创建新的响应处理器
    var handler = new AiServiceStreamingResponseHandler(
            chatExecutor,
            context,
            memoryId,
            partialResponseHandler,
            partialThinkingHandler,
            beforeToolExecutionHandler,
            toolExecutionHandler,
            intermediateResponseHandler,
            completeResponseHandler,
            errorHandler,
            temporaryMemory,
            TokenUsage.sum(tokenUsage, chatResponse.metadata().tokenUsage()),
            toolSpecifications,
            toolExecutors,
            toolExecutor,
            commonGuardrailParams,
            methodKey);

    // 最终本质依旧使用streamchatmodel来进行chat问答 & handler后续处理器
    context.streamingChatModel.chat(chatRequest, handler);
}

Tool 前后钩子机制

execute 方法

private String execute(ToolExecutionRequest toolExecutionRequest) {
    ToolExecutor toolExecutor = toolExecutors.get(toolExecutionRequest.name());
    // TODO applyToolHallucinationStrategy
    // 前置tools执行
    handleBeforeTool(toolExecutionRequest);
    String toolExecutionResult = toolExecutor.execute(toolExecutionRequest, memoryId);
    // 后置tools执行
    handleAfterTool(toolExecutionRequest, toolExecutionResult);
    return toolExecutionResult;
}

前置钩子 handleBeforeTool

private void handleBeforeTool(ToolExecutionRequest toolExecutionRequest) {
    if (beforeToolExecutionHandler != null) {
        BeforeToolExecution beforeToolExecution = BeforeToolExecution.builder()
                .request(toolExecutionRequest)
                .build();
        beforeToolExecutionHandler.accept(beforeToolExecution);
    }
}

后置钩子 handleAfterTool

private void handleAfterTool(ToolExecutionRequest toolExecutionRequest, String toolExecutionResult) {
    if (toolExecutionHandler != null) {
        ToolExecution toolExecution = ToolExecution.builder()
                .request(toolExecutionRequest)
                .result(toolExecutionResult)
                .build();
        toolExecutionHandler.accept(toolExecution);
    }
}

Handler 初始化机制

初始化位置

toolExecutionHandler 和 beforeToolExecutionHandler 这两个处理器是在 AiServiceTokenStream 中进行初始化的。

AiService 接口定义

interface Assistant {
    TokenStream chat(String userMessage);
}

总结

LangChain4j 在流式调用场景下，通过 AiServiceStreamingResponseHandler 实现了 Tool 的调用及其前后钩子机制：

1. 触发时机：在流式响应完成的 onCompleteResponse 回调中触发
2. 执行模式：支持同步和异步两种执行模式
3. 钩子机制：提供 beforeToolExecutionHandler 和 toolExecutionHandler 两个钩子用于前后置处理
4. 循环调用：Tool 执行完成后会将结果加入内存，并继续发起下一轮对话

---

整理者：长路 时间：2025.9.27