自定义节点实现

本页面提供了在 Koog 框架中实现自定义节点的详细说明。自定义节点允许您通过创建执行特定操作的可复用组件来扩展智能体工作流的功能。

要了解更多关于图节点是什么、其用法以及现有的默认节点，请参阅图节点。

节点架构概述

在深入实现细节之前，了解 Koog 框架中节点的架构非常重要。节点是智能体工作流的基本构建块，每个节点代表工作流中的一个特定操作或转换。您使用边连接节点，边定义了节点之间的执行流程。

每个节点都有一个 execute 方法，该方法接收输入并产生输出，然后传递给工作流中的下一个节点。

实现自定义节点

自定义节点的实现范围广泛，从对输入数据执行基本逻辑并返回输出的简单实现，到接受参数并在多次运行之间维护状态的更复杂节点实现。

基本节点实现

在图中实现自定义节点并定义自定义逻辑的最简单方法是使用以下模式：

KotlinJava

val myNode by node<Input, Output>("node_name") { input ->
    // 处理逻辑
    returnValue
}

var myNode = AIAgentNode.builder("node_name")
    .withInput(Input.class)
    .withOutput(Output.class)
    .withAction((input, ctx) -> {
        // 处理逻辑
        return returnValue;
    })
    .build();

上面的代码表示一个自定义节点 myNode，具有预定义的 Input 和 Output 类型，以及可选的名称字符串参数（node_name）。在实际示例中，这里是一个简单的节点，它接收字符串输入并返回输入的长度：

KotlinJava

val myNode by node<String, Int>("node_name") { input ->
    // 处理逻辑
    input.length
}

var myNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(Integer.class)
    .withAction((input, ctx) -> {
        // 处理逻辑
        return input.length();
    })
    .build();

在 Kotlin 中创建自定义节点的另一种方法是，在 AIAgentSubgraphBuilderBase 上定义一个扩展函数，该函数调用 node 函数。在 Java 中，您可以通过将节点构建器调用提取到辅助方法中来实现相同的可复用性：

KotlinJava

fun AIAgentSubgraphBuilderBase<*, *>.myCustomNode(
    name: String? = null
): AIAgentNodeDelegate<Input, Output> = node(name) { input ->
    // 自定义逻辑
    input // 将输入作为输出返回（直通）
}
```    val myCustomNode by myCustomNode("node_name")

var myCustomNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        // 自定义逻辑
        return input; // 将输入作为输出返回（直通）
    })
    .build();

这会创建一个直通节点，它执行一些自定义逻辑，但将输入作为输出返回而不做修改。

带额外参数的节点

您可以创建接受参数以自定义其行为的节点：

KotlinJava

    fun AIAgentSubgraphBuilderBase<*, *>.myNodeWithArguments(
    name: String? = null,
    arg1: String,
    arg2: Int
): AIAgentNodeDelegate<Input, Output> = node(name) { input ->
    // 在自定义逻辑中使用 arg1 和 arg2
    input // 将输入作为输出返回
}

val myCustomNode by myNodeWithArguments("node_name", arg1 = "value1", arg2 = 42)

String arg1 = "value1";
int arg2 = 42;

var myCustomNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        // 在自定义逻辑中使用 arg1 和 arg2
        return input; // 将输入作为输出返回
    })
    .build();

参数化节点

您可以定义具有输入和输出参数的节点：

KotlinJava

inline fun <reified T> AIAgentSubgraphBuilderBase<*, *>.myParameterizedNode(
    name: String? = null,
): AIAgentNodeDelegate<T, T> = node(name) { input ->
    // 执行一些额外操作
    // 将输入作为输出返回
    input
}

val strategy = strategy<String, String>("strategy_name") {
    val myCustomNode by myParameterizedNode<String>("node_name")
}

// 在 Java 中，构建节点时显式指定类型
var myCustomNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        // 执行一些额外操作
        // 将输入作为输出返回
        return input;
    })
    .build();

有状态节点

如果您的节点需要在多次运行之间保持状态，可以使用闭包变量：

Kotlin

fun AIAgentSubgraphBuilderBase<*, *>.myStatefulNode(
    name: String? = null
): AIAgentNodeDelegate<Input, Output> {
    var counter = 0

    return node(name) { input ->
        counter++
        println("节点已执行 $counter 次")
        input
    }
}

=== "Java"

// 在 Java 中，使用 AtomicInteger（或类似工具），因为 lambda 捕获的变量必须是有效的 final 变量
AtomicInteger counter = new AtomicInteger(0);

var myStatefulNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        int count = counter.incrementAndGet();
        System.out.println("Node executed " + count + " times");
        return input;
    })
    .build();

节点输入与输出类型

节点可以具有不同的输入和输出类型，这些类型被指定为泛型参数：

KotlinJava

val stringToIntNode by node<String, Int>("node_name") { input: String ->
    // 处理过程
    input.toInt() // 将字符串转换为整数
}

var stringToIntNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(Integer.class)
    .withAction((input, ctx) -> {
        // 处理过程
        return Integer.parseInt(input); // 将字符串转换为整数
    })
    .build();

Note

输入和输出类型决定了节点在工作流中如何与其他节点连接。只有当源节点的输出类型与目标节点的输入类型兼容时，节点才能被连接。

最佳实践

在实现自定义节点时，请遵循以下最佳实践：

保持节点功能专注：每个节点应执行单一、定义明确的操作。
使用描述性名称：节点名称应清晰表明其用途。
记录参数：为所有参数提供清晰的文档说明。
优雅地处理错误：实现适当的错误处理，以防止工作流失败。
使节点可重用：设计节点时考虑在不同工作流中的可重用性。
使用类型参数：在适当的情况下使用泛型类型参数，使节点更加灵活。
提供默认值：尽可能为参数提供合理的默认值。

常见模式

以下部分提供了一些实现自定义节点的常见模式。

直通节点

执行操作但将输入作为输出返回的节点。

KotlinJava

val loggingNode by node<String, String>("node_name") { input ->
    println("Processing input: $input")
    input // 将输入作为输出返回
}

var loggingNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        System.out.println("Processing input: " + input);
        return input; // 将输入作为输出返回
    })
    .build();

转换节点

将输入转换为不同输出的节点。

=== "Kotlin"

val upperCaseNode by node<String, String>("node_name") { input ->
    println("Processing input: $input")
    input.uppercase() // 将输入转换为大写
}

Java

var upperCaseNode = AIAgentNode.builder("node_name")
    .withInput(String.class)
    .withOutput(String.class)
    .withAction((input, ctx) -> {
        System.out.println("Processing input: " + input);
        return input.toUpperCase(); // 将输入转换为大写
    })
    .build();

LLM 交互节点

与 LLM 交互的节点。

KotlinJava

val summarizeTextNode by node<String, String>("node_name") { input ->
    llm.writeSession {
        appendPrompt {
            user("请总结以下文本: $input")
        }

        val response = requestLLMWithoutTools()
        response.content
    }
}

// 在 Java 中，LLM 交互使用预构建的工厂节点处理。
// AIAgentNode.llmRequest() 创建一个节点，将输入字符串作为用户消息发送到 LLM 并返回响应。提示文本在图中执行节点时作为节点的输入提供。
var summarizeTextNode = AIAgentNode.llmRequest(true, "node_name");

// 要从 LLM 响应中提取文本内容，可以链接一个单独的节点：
var extractContent = AIAgentNode.builder("extract-content")
    .withInput(Message.Response.class)
    .withOutput(String.class)
    .withAction((response, ctx) -> response.getContent())
    .build();

Note

上面的 Kotlin 示例展示了对 LLM 会话的细粒度控制（自定义提示构建、显式 requestLLMWithoutTools 调用）。Java API 提供了更高级的工厂方法，如 AIAgentNode.llmRequest()，可自动处理提示构建——输入字符串即成为用户消息。对于大多数用例来说这已足够；对于高级提示定制，可以组合多个节点或使用自定义子图。

工具运行节点

Kotlin

val nodeExecuteCustomTool by node<String, String>("node_name") { input ->
    val toolCall = Message.Tool.Call(
        id = UUID.randomUUID().toString(),
        tool = toolName,
        metaInfo = ResponseMetaInfo.create(Clock.System),
        content = Json.encodeToString(ToolArgs(arg1 = input, arg2 = 42)) // 使用输入作为工具参数
    )

    val result = environment.executeTool(toolCall)
    result.content
}

=== "Java"

// 在 Java 中，无法通过 Java 构建器 API 直接执行工具（如 Kotlin 示例所示）。
// 取而代之的是使用一个子图，将工具调用委托给 LLM，由其决定何时以及如何调用工具：
var toolSubgraph = AIAgentSubgraph.builder("tool-subgraph")
    .withInput(String.class)
    .withOutput(String.class)
    .withTask(input -> "Use my_tool with input: " + input)
    .build();

Note

Kotlin 示例通过手动构建 Message.Tool.Call 并调用 environment.executeTool() 来演示底层的工具执行方式。而 Java API 则提倡使用更高层级的子图方法，结合 withTask()，由 LLM 自动编排工具调用。若要限制可用工具的范围，可在 .withInput() 前链式调用 .limitedTools(List.of(myTool))。