Koog 1.3.0 Strategy 详解：工作流编排、并行执行与状态传递

前言

前面的文章 Koog(2) —— Koog Agent 的编程模型与类型选择 介绍了不同 Agent 实现形式及其构建、配置和运行方式。本文继续讨论 Agent 的行为逻辑如何组织，重点是图式策略的组成、控制流、并行执行与数据传递。

本文以 Koog 1.3.0 的 Kotlin API 为基准。这里的“策略图”特指通过 strategy { } 构建的 AIAgentGraphStrategy；functionalStrategy 则用函数组织逻辑，不能把所有 Strategy 都等同于有向图。官方指南与该版本实现不一致的地方，下文会明确说明。除标明完整运行入口的示例外，其余代码是用于解释 API 的局部片段，需要放入相应 DSL 并补齐导入、工具与执行器配置。

一、Strategy graph

1.1 回顾：那个"拿来就用"的 strategy()

在 Koog(2) 讲 Graph-based agents 时，有这样一段示例代码（略有精简）：

val simpleGraphStrategy = strategy<String, String>("Simple graph-based agent") {
    val requestFavoritePetTool by nodeLLMRequestOnlyCallingTools("requestFavoritePetTool")
    val executeTools by nodeExecuteTools("executeTools")
    val answerWithToolResult by nodeLLMSendToolResultsWithoutTools("answerWithToolResult")

    edge(nodeStart forwardTo requestFavoritePetTool)
    edge(requestFavoritePetTool forwardTo executeTools onToolCalls { true })
    edge(executeTools forwardTo answerWithToolResult)
    edge(answerWithToolResult forwardTo nodeFinish onTextMessage { true })
}

val agent = AIAgent(
    promptExecutor = MultiLLMPromptExecutor(deepSeekClient),
    llmModel = DeepSeekModels.DeepSeekV4Flash,
    strategy = simpleGraphStrategy,  // 把图交给 Agent
    // ... 其他配置省略
)

当时为了把例子跑通，重点放在了"用 graph 强制规定 Agent 的执行路径"这个效果上。但代码里的这一堆概念，并没有展开讲：

• strategy<String, String> 的两个泛型是什么意思？
• by nodeXXX(...) 这种写法为什么能行？
• onToolCalls { true } 到底在检查什么？
• 为什么图要从 nodeStart 出发、到 nodeFinish 结束？

这些问题就是本文要回答的。先把核心结论放在这里：

图式策略用有向图组织执行逻辑：节点（Node）回答“做什么”，边（Edge）回答“接下来去哪”。一次正常完成的图执行从 nodeStart 出发，到 nodeFinish 返回结果。

1.2 Strategy graph 的架构

Strategy graph 是 Koog 框架中智能体工作流的骨干。它们定义了智能体如何处理输入、与工具交互以及生成输出。策略图由通过边连接的节点组成，并通过条件决定执行流程。

创建策略图能够根据具体需求定制智能体的行为，无论是在构建简单的聊天机器人、复杂的数据处理流水线，还是介于两者之间的任何东西。

Strategy graph 由以下的组件组成：

策略图从一个名为 nodeStart 的特殊节点开始，并在 nodeFinish 结束。 这些节点之间的路径由图中指定的边和条件决定。

把这几个概念标到 1.1 那段代码上：

flowchart LR subgraph S["strategy&lt;String, String&gt; —— 顶层容器"] nodeStart(["nodeStart"]) --> requestTool["requestFavoritePetTool<br/>（Node）"] requestTool -->|"onToolCalls（Edge 条件）"| executeTools["executeTools<br/>（Node）"] executeTools --> answer["answerWithToolResult<br/>（Node）"] answer -->|"onTextMessage（Edge 条件）"| nodeFinish(["nodeFinish"]) end

（这个例子只有一层，没有用到 Subgraph；Subgraph 在第四章单独讲。）

有几个点值得单独说：

1. strategy 的泛型 = Agent 的输入输出类型

strategy<String, String> 声明了这个工作流"输入是一个 String，返回也是一个 String"，它直接决定了使用它的 Agent 是 AIAgent<String, String>。

2. 图必须连通：nodeStart → nodeFinish

每张策略图都有一条从 nodeStart（内置起始节点，代表用户输入进来）到 nodeFinish（内置结束节点，代表产出最终结果）的路径。如果某个节点没有任何路径能走到 nodeFinish，执行流到那里就断了。（这个坑在第九章排错部分展开。）

3. by nodeXXX(...) 是 Kotlin 属性委托

val executeTools by nodeExecuteTools("executeTools") 里的 by 是 Kotlin 的属性委托语法：先"登记"一个节点定义，等图真正执行到它时才运行。声明节点用 by，连接节点用 edge。

4. 条件挂在边上，不挂在节点上

onToolCalls { true } 是边的流转条件："上一步 LLM 的响应里出现了工具调用，就走这条边"。把条件放在边上而不是节点里，是 Koog 图模型的关键设计——同一个节点的输出，可以按不同条件流向不同的下游，这就是第三章"条件分支"的基础。

1.3 Strategy graph 的核心组件

1.3.1 Node

节点是策略图的构建块。每个节点代表一个特定的操作。Koog 框架提供了预定义节点，同时也允许使用 node 函数创建自定义节点。

1.3.2 Edge

边连接节点并定义策略图中的操作流程。边通过 edge 函数和 forwardTo 中缀函数创建：

edge(sourceNode forwardTo targetNode)

1.3.3 Condition

条件决定了在策略图中何时沿某条特定的边继续执行。条件有多种类型，下面列出一些常见类型：

这里列的是 1.3.0 可用的常见写法。官方指南还出现过 onToolNotCalled，但该版本没有提供同名函数，不能直接照用。onToolCalls 不仅筛选调用，还会把消息转换为 ToolCalls；onTextMessage 会提取文本并转换为 String。它们不是单纯的布尔判断。

在将输出传递给目标节点之前，还可以使用 transformed 函数先对其进行转换：

edge(sourceNode forwardTo targetNode
        onCondition { input -> input.length > 10 }
        transformed { input -> input.uppercase() }
)

1.3.4 Subgraph

子图是策略图中拥有各自工具集和上下文的独立部分。 一个策略图可以包含多个子图。每个子图都通过 subgraph 函数定义：

val strategy = strategy<Input, Output>("strategy-name") {
    val firstSubgraph by subgraph<FirstInput, FirstOutput>("first") {
        // Define nodes and edges for this subgraph
    }
    val secondSubgraph by subgraph<SecondInput, SecondOutput>("second") {
        // Define nodes and edges for this subgraph
    }
}

子图可以使用工具注册表中的任意工具。 也可以从该注册表中指定一组仅供子图使用的工具子集，并将其作为参数传给 subgraph 函数：

val strategy = strategy<Input, Output>("strategy-name") {
    val firstSubgraph by subgraph<FirstInput, FirstOutput>(
        name = "first",
        tools = listOf(someTool)
    ) {
        // Define nodes and edges for this subgraph
    }
   // Define other subgraphs
}

1.4 学习路线：三层能力 + 两个专题

我把 Koog 中 Strategies 的相关知识点组织成"三层能力 + 两个专题"，也就是本文后续章节的内容，包含：预定义节点、预定义策略、自定义图、并行执行、节点间数据传递等。

flowchart TD L1["第一层：直接用预定义策略<br/>chatAgentStrategy() / reActStrategy()"] --> L2["第二层：用预定义节点组装自己的图<br/>nodeLLMRequest、nodeExecuteTools、edge ..."] L2 --> L3["第三层：完全自定义<br/>自定义 node、自定义 subgraph"] L2 -.-> T1["专题一：控制进阶<br/>条件分支、并行节点执行"] L2 -.-> T2["专题二：数据传递<br/>transformed、AIAgentStorage"]

• 第一层：Koog 把常见模式做成了现成策略，strategy = chatAgentStrategy() 一行搞定。先判断"我的场景够不够典型"，够典型就不用自己画图。
• 第二层：预定义节点是积木，Strategy graph 是图纸。掌握十几个常用节点和边的条件写法，能拼出绝大多数工作流。
• 第三层：积木不够用就自己造——自定义 node、自定义 subgraph。
• 专题一：条件分支和并行执行，决定图的"控制力"上限。
• 专题二：节点之间除了沿着边传数据，还能通过 AIAgentStorage 共享数据。

总结一下本章的核心：

Agent 类型决定"外壳"（Koog(2) 已讲），Strategy 决定"内核"。本文的图式策略是一张从 nodeStart 到 nodeFinish 的有向图，由策略、子图、节点和边共同组成；学习它可以按"用现成策略 → 攒预定义节点 → 完全自定义"三层递进。

接下来，我们从最小的构建块——Node 开始。

二、Node：工作流的基本构建块

2.1 Node 的本质：一个类型化的函数

节点是 Koog 智能体工作流的基本构建块，每个节点表示工作流中的一个特定操作或转换，可以用边连接起来形成执行流。

节点可以把复杂逻辑封装为可复用的执行单元，再通过边组合进工作流。

更进一步说，每个 Node 本质上就是一个函数：接收一个指定类型的输入，返回一个指定类型的输出。

自定义一个最简单的节点：

val nodeLength by node<String, Int> { input ->
    input.length
}

接收一个 String，返回它的长度 Int。不需要继承任何类、实现任何接口，一个 lambda 就是一个节点。

Koog(2) 的示例代码中用过的那些 nodeXXX 预定义节点，全部符合这个定义：它们只是框架已经写好的"函数"。

2.2 常用预定义节点速查

下面按 Koog 1.3.0 的实际 Kotlin 签名列出常用节点，不是全部 API 清单。String 输入的 nodeLLMRequest* 与 Message.User 输入的 nodeLLMSendMessage* 是两组不同接口，不能混用。

2.2.1 辅助节点（Utility nodes）

val passthrough by nodeDoNothing<String>("passthrough")
edge(nodeStart forwardTo passthrough)
edge(passthrough forwardTo nodeFinish)

2.2.2 LLM 节点（LLM nodes）

“要求调用工具”仍依赖 Provider 与模型对工具选择参数的支持；图中的条件边也应处理异常响应。LLMChoice、StructuredResponse、StreamFrame 等类型需要各自的导入与配置，这里只列出接口关系。

2.2.3 Tool 节点（Tool nodes）

ToolCalls、ReceivedToolResults 是框架的封装类型，不能直接当作 List 使用。SafeTool.Result 区分成功与失败，指定工具重载的调用方应显式处理这两种情况。第八章会给出实际用法。

这些节点不用背：先按“请求模型”“执行工具”“回传结果”查找，再检查输入输出是否能衔接。

2.3 重点 LLM 节点详解

nodeLLMRequest：最常用的请求节点

追加用户文本到 prompt，再发起请求；nodeLLMRequest() 使用当前上下文中的工具集。若明确禁止工具调用，应选择 nodeLLMRequestWithoutTools()：

val requestLLM by nodeLLMRequest("requestLLM")
edge(getUserQuestion forwardTo requestLLM)

Koog(2) 的示例代码中用过它的变体——"只允许调工具"的版本。

nodeAppendPrompt：先铺垫，再请求

它不请求 LLM，只往 prompt 里追加消息，用于在真正的 LLM 请求之前准备上下文，比如插入 system 指令或补充 user 消息：

val setupContext by nodeAppendPrompt<String>("setupContext") {
    system("You are a helpful assistant specialized in Kotlin programming.")
    user("I need help with Kotlin coroutines.")
}

nodeLLMCompressHistory：历史压缩

把当前 LLM prompt（消息历史）压缩为一条保留关键信息的摘要，适合长对话场景控制 token：

val compressHistory by nodeLLMCompressHistory<String>(
    "compressHistory",
    strategy = HistoryCompressionStrategy.FromLastNMessages(10),
    preserveMemory = true
)

历史压缩的完整机制（压缩策略、和记忆的关系）会在后面讲记忆与持久化的文章中展开，这里只需要知道：图里有一个现成的压缩节点，挂上去就能用。

nodeLLMRequestStructured 与 nodeLLMRequestStreaming

分别对应结构化输出和流式输出。这两个能力值得单独一篇文章，后续会专门讲，此处先记住节点名字和用途即可。

2.4 transform()：转换节点的输出

AIAgentNodeDelegate 提供了 transform 成员方法，可以包装一个已有节点，并对它的输出应用转换。比如把 nodeLLMRequest 的响应转换成文本长度：

val lengthNode by nodeLLMRequest("llmRequest").transform<Int> { assistantMessage ->
    assistantMessage.parts.filterIsInstance<MessagePart.Text>().joinToString("\n") { it.text }.length
}

edge(nodeStart forwardTo lengthNode)
edge(lengthNode forwardTo nodeFinish)

理解这段代码：lengthNode 仍然是 String → Int 的节点（输入用户消息，输出文本长度），但内部复用了 nodeLLMRequest 的能力。transform 的价值就是让预定义节点"长成"你需要的形状。

2.5 自定义节点：从直接定义到封装复用

预定义节点不够用时，可以通过 node 定义自己的执行逻辑。下面从最简单的节点开始，再看如何封装为可复用的节点工厂，以及如何为节点增加配置参数和局部状态。

方式一：直接定义节点

在 strategy { } 或 subgraph { } 中直接定义节点，适合只在当前流程中使用的逻辑：

val textLength by node<String, Int>("textLength") { input ->
    input.length
}

这个节点接收 String，返回 Int。节点的输入输出类型由 node<String, Int> 指定，不要求与整个策略图的输入输出类型一致，只要连接它的上下游类型兼容即可。

方式二：提取为扩展函数，跨图复用

如果多个工作流都需要同一段节点逻辑，可以将它提取为返回 AIAgentNodeDelegate 的节点工厂。官方 Custom nodes 页面采用了在 AIAgentSubgraphBuilderBase 上定义扩展函数的写法。

AIAgentSubgraphBuilderBase<Input, Output> 是策略图构建器与子图构建器的共同抽象基类，提供起止节点、边连接等建图能力：

• strategy { } 的接收者是 AIAgentGraphStrategyBuilder，继承自这个基类。
• subgraph { } 的接收者是 AIAgentSubgraphBuilder，同样继承自这个基类。

因此，定义在这个基类上的扩展函数，可以在两种 DSL 块中调用。比如，封装一个记录输入、随后原样返回的日志节点：

import ai.koog.agents.core.dsl.builder.AIAgentNodeDelegate
import ai.koog.agents.core.dsl.builder.AIAgentSubgraphBuilderBase
import ai.koog.agents.core.dsl.builder.node
import ai.koog.agents.core.dsl.builder.strategy

fun AIAgentSubgraphBuilderBase<*, *>.loggingNode(
    name: String? = null
): AIAgentNodeDelegate<String, String> = node<String, String>(name) { input ->
    println("Input: $input")
    input
}

val loggingStrategy = strategy<String, String>("logging-example") {
    val logInput by loggingNode("logInput")

    edge(nodeStart forwardTo logInput)
    edge(logInput forwardTo nodeFinish)
}

这里有三个容易混淆的点：

• AIAgentSubgraphBuilderBase<*, *> 中的 * 是星投影，表示这个扩展函数不依赖外围图的具体输入输出类型；它不会声明名为 Input、Output 的类型参数。
• 返回类型 AIAgentNodeDelegate<String, String> 描述的是节点自身的输入输出，而不是外围图的输入输出。每次调用 loggingNode(...) 都会创建一个新的节点委托，供 by 声明节点。
• 在 Koog 1.3.0 中，node() 是顶层函数，不是这个基类的成员。普通函数也可以调用它并返回节点委托；使用扩展函数是为了把节点工厂组织到建图 DSL 中，并不是为了获得调用 node() 的权限。

如果希望日志节点支持任意输入类型，可以将上面的工厂替换为泛型版本：

inline fun <reified T> AIAgentSubgraphBuilderBase<*, *>.loggingNode(
    name: String? = null
): AIAgentNodeDelegate<T, T> = node<T, T>(name) { input ->
    println("Input: $input")
    input
}

// 在 strategy { } 或 subgraph { } 中使用
val logInput by loggingNode<String>("logInput")

这里使用 inline 和 reified，是因为 node<T, T>() 需要获取输入输出的类型信息。这个 T 是节点工厂自己的类型参数，与外围图的类型参数无关。

方式三：为节点工厂增加配置参数

配置参数在创建节点时传入，用于定制节点的行为；节点真正执行时，仍然通过 input 接收上游数据。两者的时机不同：

fun AIAgentSubgraphBuilderBase<*, *>.truncateNode(
    maxLength: Int,
    name: String? = null
): AIAgentNodeDelegate<String, String> {
    require(maxLength >= 0) { "maxLength must be non-negative" }

    return node<String, String>(name) { input ->
        input.take(maxLength)
    }
}

// 在 strategy { } 或 subgraph { } 中使用
val truncateText by truncateNode(maxLength = 100, name = "truncateText")

maxLength 在建图时确定，并由执行 lambda 捕获；input 则是在执行流到达节点时才传入的文本。这样可以复用同一个节点工厂，在不同位置创建具有不同长度限制的节点。

方式四：通过闭包保存局部状态

节点工厂还可以创建局部变量，让返回的执行逻辑捕获它。例如，统计这段节点逻辑被执行了多少次：

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

    return node<String, String>(name) { input ->
        counter++
        println("Node executed $counter times")
        input
    }
}

counter 在调用 countingNode(...) 时创建，不是在每次执行节点时重新创建。同一闭包被重复执行时会保留计数；再次调用节点工厂，则会创建另一个独立的计数器。

但这种写法有明确的边界：

• 不会自动按运行隔离：如果同一节点逻辑被多个 Agent 或多次运行复用，它们可能共用这个计数器，不能把它理解为“本次运行的计数”。
• 不保证并发安全：counter++ 不是原子操作。上面的示例只用于说明串行执行下的闭包行为，不应直接用于并发计数。
• 不会自动持久化：闭包变量不会因为启用了 Agent Persistence 就自动进入 checkpoint。

对于需要在工作流中管理、跨节点访问的运行状态，应优先考虑第七章介绍的 AIAgentStorage，并结合其运行生命周期和并行上下文规则设计读写方式；不要依赖闭包变量隐式共享。

还要区分构建期接收者和运行期接收者：扩展函数的接收者 AIAgentSubgraphBuilderBase 用于建图；传给 node 的执行 lambda，其接收者是 AIAgentGraphContextBase，执行时访问的 llm、storage 等能力来自后者。下面的 LLM 交互示例用的就是运行期上下文。

常见模式小结

官方文档总结了自定义节点的四种常见模式：

其中"LLM 交互"模式后面并行执行和综合实战里都会用到，值得单独看一眼：

val summarizeTextNode by node<String, String>("summarize") { input ->
    llm.writeSession {
        appendPrompt {
            user("Please summarize the following text: $input")
        }
        val response = requestLLMWithoutTools()
        response.parts.filterIsInstance<MessagePart.Text>().joinToString("\n") { it.text }
    }
}

在自定义节点里，llm.writeSession 就是第 3 篇讲过的 LLM 写会话：追加 prompt、切换模型、发请求，全都可以手动控制。

2.6 类型决定连接

Strategy graph 最重要的一条规则：

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

这就是为什么需要 transformed（边上的转换）和 transform()（节点上的转换）：它们是类型不匹配时的"转接头"。

一句话总结本章：

Node 是类型化的函数；Koog 提供了覆盖"LLM 请求 / 工具执行 / 历史压缩 / 透传"的预定义节点；不够用时，一个 lambda 就是自定义节点；而节点能不能连上，看类型。

节点解决了"做什么"，那"接下来去哪"呢？这就是 Edge 的事。下一章我们把第一章的基础组件拼成一张真正能跑的图，看看边如何构成完整的控制流。

三、组装：从基础图到条件分支

第一章中已经认识了 edge、forwardTo 和常见条件写法。这一章关心的问题是：这些边拼在一起，整体是怎么跑的？怎么表达分支和顺序？

3.1 最基本的工具调用图

下面是一个 Basic Agent 的策略图示例的执行流程：

1. 将输入发送给 LLM。
2. 如果 LLM 直接回复了消息，流程结束。
3. 如果 LLM 调用工具，执行该工具。
4. 把工具结果发回 LLM。
5. 如果 LLM 回复了消息，流程结束。
6. 如果 LLM 又调用工具，执行工具，回到第 4 步。

--- title: my-strategy --- stateDiagram state "nodeCallLLM" as nodeCallLLM state "executeToolCall" as executeToolCall state "sendToolResult" as sendToolResult [*] --> nodeCallLLM nodeCallLLM --> [*] : transformed nodeCallLLM --> executeToolCall : onCondition executeToolCall --> sendToolResult sendToolResult --> [*] : transformed sendToolResult --> executeToolCall : onCondition

对应代码：

val myStrategy = strategy<String, String>("my-strategy") {
    val nodeCallLLM by nodeLLMRequest()
    val executeToolCall by nodeExecuteTools()
    val sendToolResult by nodeLLMSendToolResults()

    edge(nodeStart forwardTo nodeCallLLM)
    edge(nodeCallLLM forwardTo nodeFinish onTextMessage { true })
    edge(nodeCallLLM forwardTo executeToolCall onToolCalls { true })
    edge(executeToolCall forwardTo sendToolResult)
    edge(sendToolResult forwardTo nodeFinish onTextMessage { true })
    edge(sendToolResult forwardTo executeToolCall onToolCalls { true })
}

理解这张图：

• 三个节点构成"请求 → 执行 → 回传"的循环三角。
• nodeCallLLM 和 sendToolResult 各有两条出边：LLM 回了文本 → 走向结束；LLM 要调工具 → 走向工具执行。这正是"条件挂在边上"的价值——同一个节点的输出，按响应内容分流。
• sendToolResult → executeToolCall 这条回边形成了循环：LLM 可以连续调用多次工具，直到它给出最终文本。

顺便说一句：这张图就是官方 singleRunStrategy 的真身。Basic Agent 之所以"拿来就能用"，就是因为它背后挂的就是这张图。

3.2 条件分支

当工作流需要根据中间结果走不同路径时，给同一个源节点定义多条带条件的出边：

val branchA by node<String, String> { input ->
    "Branch A: $input"
}

val branchB by node<String, String> { input ->
    "Branch B: $input"
}

edge(
    (someNode forwardTo branchA)
            onCondition { input -> input.contains("A") }
)
edge(
    (someNode forwardTo branchB)
            onCondition { input -> input.contains("B") }
)

输出包含 "A" 时可匹配分支 A，包含 "B" 时可匹配分支 B。若同时包含二者，按下文的边顺序规则选择；若二者都不包含，需要另设兜底边，否则会抛出 AIAgentStuckInTheNodeException。

一个关键细节：边按定义顺序评估。同一个源节点的多条出边，Koog 按代码中定义的顺序逐条检查条件，第一条满足的生效。所以宽泛的条件不要放在前面，否则会"截胡"后面的具体条件。

3.3 then：顺序流的语法糖

对于没有条件的顺序流，可以用 then 链式连接，比一条条写 edge 简洁：

nodeStart then firstNode then secondNode then nodeFinish

有条件流转时，仍然需要 edge + onXxx 的完整写法。

3.4 可视化：asMermaidDiagram()

"易于可视化"是 Graph-based Agent 的三大优势之一，来源就在这里——JVM 上可以为任何策略图生成 Mermaid 状态图：

val mermaidDiagram: String = myStrategy.asMermaidDiagram()
println(mermaidDiagram)

对 3.1 创建的图，输出如下：

---
title: my-strategy
---
stateDiagram
    state "nodeCallLLM" as nodeCallLLM
    state "executeToolCall" as executeToolCall
    state "sendToolResult" as sendToolResult

    [*] --> nodeCallLLM
    nodeCallLLM --> [*] : transformed
    nodeCallLLM --> executeToolCall : onCondition
    executeToolCall --> sendToolResult
    sendToolResult --> [*] : transformed
    sendToolResult --> executeToolCall : onCondition

把这段输出贴到任何支持 Mermaid 的地方（IDE 插件、GitHub、博客平台）就能直接渲染出图。图反映生成时的代码结构；代码修改后需要重新生成。transformed、onCondition 是生成器提取的内部实现标签，不一定与 DSL 中 onTextMessage、onToolCalls 的名称一致，也不会完整展示条件 lambda 的业务含义。

一句话总结：

顺序流用 then，分支用多条带条件的出边，循环用回边；边按定义顺序评估条件；一张 asMermaidDiagram() 就能让图"画"出来。

四、Subgraph：封装与复用局部流程

4.1 Subgraph 是什么

Subgraph 定义：子图是策略内自包含的处理单元，用于把复杂工作流拆解成可管理的顺序步骤。每个子图：

• 有唯一的名称；
• 内部是一张由节点/子图和边构成的图；
• 可以使用工具注册表的全部工具，也可以只用其中的一个子集；
• 接收上一个子图（或初始用户输入）的输出，产出传递给下一个子图（或最终输出）。

每个子图在自己的上下文中执行，这个上下文提供：环境、agent 输入、agent 配置、LLM 上下文（包括对话历史）、状态管理器、storage、会话与策略。

第一章 1.3.4 已经看过 subgraph 的创建语法。补一个理解方式：

subgraph 之于 strategy，就像函数之于程序——声明输入类型、封装内部逻辑、返回输出类型，可以被组合和复用。

4.2 子图中的工具配置

让子图"只能用某些工具"有三种方式。

方式一：定义时直接指定（第一章已见）

val mySubgraph by subgraph<String, String>(
    tools = listOf(AskUser)
) {
    // 子图定义
}

也可以用 toolSelectionStrategy 指定选择策略：

val subgraphIdentifier by subgraph<Input, Output>(
    name = "subgraph-name",
    toolSelectionStrategy = ToolSelectionStrategy.ALL
) {
    // Define nodes and edges for this subgraph
}

方式二：从工具注册表获取

val mySubgraph by subgraph<String, String>(
    tools = listOf(toolRegistry.getTool("AskUser"))
) {
    // 子图定义
}

方式三：运行时动态调整

在子图的节点里，通过 llm.writeSession 动态过滤当前可用的工具：

this.llm.writeSession {
    tools = tools.filter { it.name in listOf("first_tool_name", "second_tool_name") }
}

这三种方式的差别：前两种是静态声明，图跑起来之前就定了；第三种是运行时按逻辑调整，同一个子图在不同条件下可以暴露不同的工具集。

4.3 Multi-part strategies：多子图串联

复杂工作流可以拆成多个子图，各管一段。官方文档给了一个非常典型的 assistant 示例，整体结构是：

flowchart LR nodeStart([nodeStart]) --> research["researchSubgraph<br/>（联网搜索）"] research --> plan["planSubgraph<br/>（制定计划）"] plan --> execute["executeSubgraph<br/>（执行动作）"] execute --> nodeFinish([nodeFinish])

下面是根据官方示例整理并补齐结束边的策略片段。WebSearchTool、DoAction、DoAnotherAction 是示意的业务工具，需要自行实现并注册到 Agent 的 ToolRegistry；代码本身不是完整应用。每个工具循环都保留“继续调用”和“文本结束”两类出边：

val strategy = strategy<String, String>("assistant") {

    // 子图 1：调研 —— 只能用搜索工具
    val researchSubgraph by subgraph<String, String>(
        "research_subgraph",
        tools = listOf(WebSearchTool())
    ) {
        val nodeCallLLM by nodeLLMRequest("call_llm")
        val nodeExecuteTool by nodeExecuteTools()
        val nodeSendToolResult by nodeLLMSendToolResults()

        edge(nodeStart forwardTo nodeCallLLM)
        edge(nodeCallLLM forwardTo nodeExecuteTool onToolCalls { true })
        edge(nodeExecuteTool forwardTo nodeSendToolResult)
        edge(nodeSendToolResult forwardTo nodeExecuteTool onToolCalls { true })
        edge(nodeSendToolResult forwardTo nodeFinish onTextMessage { true })
        edge(nodeCallLLM forwardTo nodeFinish onTextMessage { true })
    }

    // 子图 2：规划 —— 不给任何工具，纯推理
    val planSubgraph by subgraph<String, String>(
        "plan_subgraph",
        tools = listOf()
    ) {
        val nodeUpdatePrompt by node<String, Unit> { research ->
            llm.writeSession {
                rewritePrompt {
                    prompt("research_prompt") {
                        system(
                            "You are given a problem and some research on how it can be solved." +
                                    "Make step by step a plan on how to solve given task."
                        )
                        user("Research: $research")
                    }
                }
            }
        }
        val nodeCallLLM by nodeLLMRequest("call_llm")

        edge(nodeStart forwardTo nodeUpdatePrompt)
        edge(nodeUpdatePrompt forwardTo nodeCallLLM transformed { "Task: $agentInput" })
        edge(nodeCallLLM forwardTo nodeFinish onTextMessage { true })
    }

    // 子图 3：执行 —— 只能用动作类工具
    val executeSubgraph by subgraph<String, String>(
        "execute_subgraph",
        tools = listOf(DoAction(), DoAnotherAction()),
    ) {
        val nodeUpdatePrompt by node<String, Unit> { plan ->
            llm.writeSession {
                rewritePrompt {
                    prompt("execute_prompt") {
                        system(
                            "You are given a task and detailed plan how to execute it." +
                                    "Perform execution by calling relevant tools."
                        )
                        user("Plan: $plan")
                    }
                }
            }
        }
        val nodeCallLLM by nodeLLMRequest("call_llm")
        val nodeExecuteTool by nodeExecuteTools()
        val nodeSendToolResult by nodeLLMSendToolResults()

        edge(nodeStart forwardTo nodeUpdatePrompt)
        edge(nodeUpdatePrompt forwardTo nodeCallLLM transformed { "Task: $agentInput" })
        edge(nodeCallLLM forwardTo nodeExecuteTool onToolCalls { true })
        edge(nodeExecuteTool forwardTo nodeSendToolResult)
        edge(nodeSendToolResult forwardTo nodeExecuteTool onToolCalls { true })
        edge(nodeSendToolResult forwardTo nodeFinish onTextMessage { true })
        edge(nodeCallLLM forwardTo nodeFinish onTextMessage { true })
    }

    // 三个子图串联
    nodeStart then researchSubgraph then planSubgraph then executeSubgraph then nodeFinish
}

这个例子有三个值得咀嚼的点：

1. 每个子图有自己的工具集：research 只能搜索，plan 什么工具都不给（纯推理），execute 只能执行动作。工具隔离让"这一段该干什么"变得非常明确，也防止 LLM 在规划阶段乱调工具。
2. rewritePrompt 让每个子图"带着自己的任务说明书"工作：进入子图先重写 prompt（换 system 指令、塞入上一阶段的结果），相当于每个工段开工前先换一张工单。
3. 子图之间靠类型衔接：三个子图都是 String → String，所以能直接串。如果中间需要换类型，用 transformed 在边上转。

4.4 预定义子图：subgraphWithTask / subgraphWithVerification

如果连子图内部的"请求-执行-回传"循环都懒得写，Koog 提供了两个现成子图。

subgraphWithTask：任务执行子图

给它一段任务文本和一组工具，它自动完成"请求 LLM → 执行工具 → 汇总结果"的完整流程：

val processQuery by subgraphWithTask<String, String>(
    tools = listOf(searchTool, calculatorTool, weatherTool),
    llmModel = OpenAIModels.Chat.GPT4o,
    parallelTools = false,
    assistantResponseRepeatMax = 3,
) { userQuery ->
    """
    You are a helpful assistant that can answer questions about various topics.
    Please help with the following query:
    $userQuery
    """.trimIndent()
}

两个参数值得注意：

• parallelTools：工具是否并行执行，默认顺序执行。
• assistantResponseRepeatMax：允许的最大助手响应次数。超过就认为任务无法完成，防止无限循环。

这类子图也可以通过 llmParams 设置任务专用的模型参数。以上工具名均为示意，需要替换为已注册的工具实例。

subgraphWithVerification：带验证的任务子图

subgraphWithTask 的加强版：保证 LLM 在流程结束时调用验证工具，检查任务是否真正完成，并返回 CriticResult（成功/失败状态 + 详细反馈）。适合需要质检环节的流程：

val verifyCode by subgraphWithVerification<String>(
    tools = listOf(runTestsTool, analyzeTool, readFileTool),
    llmModel = AnthropicModels.Opus_4_6,
    parallelTools = false,
    assistantResponseRepeatMax = 3,
) { codeToVerify ->
    """
    You are a code reviewer. Please verify that the following code meets all requirements:
    1. It compiles without errors
    2. All tests pass
    3. It follows the project's coding standards

    Code to verify:
    $codeToVerify
    """.trimIndent()
}

一句话总结本章：

Subgraph 是策略里的"函数"：唯一命名、类型化输入输出、自己的工具集和上下文；复杂工作流拆成多个子图串联，工具隔离 + rewritePrompt 让每段职责清晰；懒得写循环就用 subgraphWithTask，要质检就用 subgraphWithVerification。

至此，图的"局部"讲完了。但如果我们根本不想自己画图呢？Koog 还提供了一批现成的整体策略——预定义策略，下一章见。

五、预定义策略与常见模式

Koog 把两类东西做成了"现成的"：整体策略（这一章）和局部子图（上一章 4.4 节）。整体策略拿到就能当 Agent 的 strategy 用，覆盖两类最典型的 Agent 模式：聊天交互和推理-行动循环。

5.1 chatAgentStrategy：聊天交互策略

chatAgentStrategy() 在 1.3.0 中不是普通问答循环，而是带工具使用反馈的策略：

1. 初始请求若直接返回文本，进入反馈节点，要求模型改为调用可用工具。
2. 反馈阶段继续返回文本时，会再次反馈，直到出现工具调用或达到运行限制。
3. 执行工具并回传结果后，文本响应可以正常结束；新的工具调用则继续循环。
4. 工具结果回传后的响应若包含 __exit__ 调用，可走专门的结束路径。

因此，它适用于希望先使用工具的交互流程，不表示所有阶段都禁止文本输出，也不会自动提供跨 run 的持久会话记忆。多轮历史如何保存，需要另行配置。

使用方式（需要 ai.koog.agents.ext.agent.chatAgentStrategy 导入）：

val chatAgent = AIAgent(
    promptExecutor = promptExecutor,
    llmModel = model,
    // 使用 chatAgentStrategy 作为 agent 的策略
    strategy = chatAgentStrategy(),
    // 注册 agent 可用的工具
    toolRegistry = ToolRegistry {
        tool(searchTool)
        tool(weatherTool)
    }
)

suspend fun main() {
    val result = chatAgent.run("What's the weather like today and should I bring an umbrella?")
}

适用场景：

• 需要使用工具的会话型 agent
• 根据用户请求执行动作的助手
• 需要访问外部系统或数据的聊天机器人
• 希望强制使用工具的场景。

5.2 reActStrategy：推理与行动交替

ReAct（Reasoning and Acting）策略在推理和执行两个阶段之间交替，动态处理任务。它的循环是：

1. 推理当前状态，规划下一步。
2. 基于推理采取行动。
3. 观察行动结果。
4. 重复，直到得出最终答案。

reActStrategy 在 1.3.0 中提供以下参数：

以下沿用官方银行场景的推理与行动示意，不是本文实测输出。原例的日期跨两个自然月，因此这里将问题表述为“这段时间花了多少钱”，避免把滚动区间当成自然月：

第 1 步：用户输入——“2025-05-19 至 2025-06-18 期间我花了多少钱？”

第 2 步：推理——agent 先想清楚步骤：

I need to follow these steps:
1. Get all transactions from 2025-05-19 to 2025-06-18
2. Filter out deposits (positive amounts)
3. Calculate total spending

第 3 步：行动（一）——根据推理调用 get_transactions 工具：

{tool: "get_transactions", args: {startDate: "2025-05-19", endDate: "2025-06-18"}}

工具返回交易记录：

[
  {date: "2025-05-25", amount: -100.00, description: "Grocery Store"},
  {date: "2025-05-31", amount: +1000.00, description: "Salary Deposit"},
  {date: "2025-06-10", amount: -500.00, description: "Rent Payment"},
  {date: "2025-06-13", amount: -200.00, description: "Utilities"}
]

第 4 步：再推理——拿到结果，继续想：

I have the transactions. Now I need to:
1. Remove the salary deposit of +1000.00
2. Sum up the remaining transactions

第 5 步：行动（二）——剔除正数金额，调用 calculate_sum：

{tool: "calculate_sum", args: {amounts: [-100.00, -500.00, -200.00]}}

工具返回 -800.00。

第 6 步：最终回复——

You spent $800.00 during the specified period on groceries, rent, and utilities.

对比 chatAgentStrategy 就能看出差别：Chat 策略侧重工具使用反馈；ReAct 显式插入"先想后做"的推理步，把多步推理任务拆成 推理 → 行动 → 观察 的循环。

使用方式：

val bankingAgent = AIAgent(
    promptExecutor = promptExecutor,
    llmModel = model,
    // 使用 reActStrategy
    strategy = reActStrategy(
        reasoningInterval = 1,
        name = "banking_agent"
    ),
    toolRegistry = ToolRegistry {
        tool(getTransactions)
        tool(calculateSum)
    }
)

suspend fun main() {
    val result = bankingAgent.run("How much did I spend last month?")
}

以上为策略使用片段，依赖已配置的执行器和工具；本文没有将它作为真实模型实测输出。

5.3 三种常见策略模式

下面整理官方指南中的三类策略模式。其中 singleRunStrategy() 是已有的预定义函数；tool-based 与 streaming data 是组织工作流的模式名称，不代表另有同名 API，也不是三种互斥分类。

模式一：singleRunStrategy（单次运行）

非交互场景，处理一次输入，返回结果。3.1 节已经讲过它的代码——就是那张"请求 → 执行 → 回传"的三角循环图。Basic Agent 预定义策略背后就是它。

模式二：tool-based strategy（工具型）

以工具为核心的工作流：LLM 决定调用什么工具，执行后把结果回传，循环直到 LLM 给出文本响应。结构上与 singleRun 一致，官方示例的侧重点在于展示如何组织这个循环，可作为业务工具流的模板。

模式三：streaming data strategy（流式数据）

下面是流程片段：Book、markdownBookDefinition() 和 parseMarkdownStreamToBooks() 需要业务代码定义，不能只复制本段就运行。它在节点内部发起流式请求，边接收边解析：

val agentStrategy = strategy<String, List<Book>>("library-assistant") {
    val getMdOutput by node<String, List<Book>> { booksDescription ->
        val books = mutableListOf<Book>()
        val mdDefinition = markdownBookDefinition()

        llm.writeSession {
            appendPrompt { user(booksDescription) }
            // 以 mdDefinition 定义的形式发起流式响应
            val markdownStream = requestLLMStreaming(mdDefinition)
            // 流式解析并处理每一本书
            parseMarkdownStreamToBooks(markdownStream).collect { book ->
                books.add(book)
                println("Parsed Book: ${book.title} by ${book.author}")
            }
        }

        books
    }
    edge(nodeStart forwardTo getMdOutput)
    edge(getMdOutput forwardTo nodeFinish)
}

这个例子也演示了 strategy 的输出类型可以是任意类型：strategy<String, List<Book>>，输入文本、输出书籍列表。

5.4 选型：什么时候用现成的，什么时候自己画

把本系列讲过的选项放到一起：

判断顺序始终是：先看预定义策略够不够 → 再看预定义节点能不能拼出来 → 最后才自定义节点和子图。

一句话总结：

chatAgentStrategy 提供工具使用反馈，reActStrategy 管"先想后做"；三种常见模式（单次运行、工具型、流式数据）是自建图的标准起手式；选型永远从预定义往自定义走。

六、并行节点执行

前面的图虽然包含分支和循环，但一次只沿匹配的路径继续执行。但很多任务天然是并发的，比如以下三类典型情况：

• 同一个输入，同时用不同的模型或方法处理；
• 多个相互独立的操作并行执行；
• 竞争评选模式：生成多个方案，再比较选出最好的。

6.1 parallel()

并行节点执行通过 parallel 方法发起：

val nodeName by parallel<String, String>(
   firstNode, secondNode, thirdNode /* 更多节点 */
) {
   // 合并策略写在这里
   selectByMax { it.length }
}

结构上把它读成一句话：这几个节点并发跑，结果出来后按某种策略合并成一个。

一个实际例子——三个节点分别统计 token 数、字符数、单词数，并发执行后取最大值：

val calc by parallel<String, Int>(
   nodeCalcTokens, nodeCalcSymbols, nodeCalcWords,
) {
   selectByMax { it }
}

执行完成后，calc 这个"合成节点"的输出就是三个结果中的最大值，下游节点拿到的就是这一个值——对图的其余部分来说，parallel 就是一个普通节点。

6.2 四种合并策略

并行执行的核心问题是：多个结果出来了，留哪个、怎么合？Koog 提供四种合并策略：

selectBy：按条件挑

val nodeSelectJoke by parallel<String, String>(
   nodeOpenAI, nodeAnthropicSonnet, nodeAnthropicOpus,
) {
   selectBy { it.contains("programmer") }
}

1.3.0 会先等待全部分支完成，再按传给 parallel 的节点顺序查找第一个满足条件的结果，不是“谁先完成就选谁”。如果没有任何匹配，selectBy 会抛出 NoSuchElementException，实际业务需要设计兜底。

selectByMax：选最大

val nodeLongestJoke by parallel<String, String>(
   nodeOpenAI, nodeAnthropicSonnet, nodeAnthropicOpus,
) {
   selectByMax { it.length }
}

选最长的那个。

selectByIndex：让"评委"来选

这个最有意思——选择函数本身可以是一次 LLM 请求：

val nodeBestJoke by parallel<String, String>(
   nodeOpenAI, nodeAnthropicSonnet, nodeAnthropicOpus,
) {
   selectByIndex { jokes ->
      // 用另一个 LLM 来判断哪个笑话最好
      llm.writeSession {
         model = OpenAIModels.Chat.GPT4o
         appendPrompt {
            system("You are a comedy critic. Select the best joke.")
            user("Here are three jokes: ${jokes.joinToString("\n\n")}")
         }
         val response = requestLLMStructured<JokeRating>()
         response.getOrThrow().data.bestJokeIndex.also { index ->
             require(index in jokes.indices) { "模型返回的笑话索引越界：$index" }
         }
      }
   }
}

三个候选笑话全部生成后，再发一次结构化请求让 GPT-4o 当评委，返回最佳笑话的索引。"多方案并行生成 + LLM 评选"，这是竞争评选模式的完整形态。

fold：聚合全部

val nodeAllJokes by parallel<String, String>(
   nodeOpenAI, nodeAnthropicSonnet, nodeAnthropicOpus,
) {
   fold("Jokes:\n") { result, joke -> "$result\n$joke" }
}

这里聚合的是输出字符串，不是把所有分支的消息历史或 storage 写入自动合并。四种策略的上下文保留规则见第七章。

6.3 官方示例改编：Best joke 策略

以下根据官方 Best joke 示例整理策略主体，补充输出类型、评分数据类和索引校验。三个不同模型并行生成笑话，再由 GPT-4o 评选最佳。

这不是独立运行的完整应用：还需要支持 OpenAI 与 Anthropic 的执行器、两家的凭据，以及 Agent 的组装与调用。不要只配置一个 Provider 却在节点里切换到另一个 Provider。模型可用性也取决于账号权限。

评分数据类（6.2 的 JokeRating 也使用这个定义）：

@kotlinx.serialization.Serializable
data class JokeRating(val bestJokeIndex: Int)

val strategy = strategy<String, String>("best-joke") {
    // 三个节点分别用三个模型生成笑话
    val nodeOpenAI by node<String, String> { topic ->
        llm.writeSession {
            model = OpenAIModels.Chat.GPT4o
            appendPrompt {
                system("You are a comedian. Generate a funny joke about the given topic.")
                user("Tell me a joke about $topic.")
            }
            val response = requestLLMWithoutTools()
            response.parts.filterIsInstance<MessagePart.Text>().joinToString("\n") { it.text }
        }
    }

    val nodeAnthropicSonnet by node<String, String> { topic ->
        llm.writeSession {
            model = AnthropicModels.Sonnet_4_5
            appendPrompt {
                system("You are a comedian. Generate a funny joke about the given topic.")
                user("Tell me a joke about $topic.")
            }
            val response = requestLLMWithoutTools()
            response.parts.filterIsInstance<MessagePart.Text>().joinToString("\n") { it.text }
        }
    }

    val nodeAnthropicOpus by node<String, String> { topic ->
        llm.writeSession {
            model = AnthropicModels.Opus_4_6
            appendPrompt {
                system("You are a comedian. Generate a funny joke about the given topic.")
                user("Tell me a joke about $topic.")
            }
            val response = requestLLMWithoutTools()
            response.parts.filterIsInstance<MessagePart.Text>().joinToString("\n") { it.text }
        }
    }

    // 并行生成，然后让 GPT-4o 评选最佳
    val nodeGenerateBestJoke by parallel(
        nodeOpenAI, nodeAnthropicSonnet, nodeAnthropicOpus,
    ) {
        selectByIndex { jokes ->
            llm.writeSession {
                model = OpenAIModels.Chat.GPT4o
                rewritePrompt {
                    prompt("best-joke-selector") {
                        system("You are a comedy critic. Give a critique for the given joke.")
                        user(
                            """
                            Here are three jokes about the same topic:

                            ${jokes.mapIndexed { index, joke -> "Joke $index:\n$joke" }.joinToString("\n\n")}

                            Select the best joke and explain why it's the best.
                            """.trimIndent()
                        )
                    }
                }

                val response = requestLLMStructured<JokeRating>()
                val index = response.getOrThrow().data.bestJokeIndex
                require(index in jokes.indices) { "模型返回的笑话索引越界：$index" }
                index
            }
        }
    }

    // 连接节点
    nodeStart then nodeGenerateBestJoke then nodeFinish
}

flowchart LR nodeStart([nodeStart]) --> p["parallel：三个模型并发生成"] p --> judge["selectByIndex：GPT-4o 评选最佳"] judge --> nodeFinish([nodeFinish])

这个例子有两个值得注意的点：

1. 并行不只是"同一个调用发三次"——三个生成节点各自在 llm.writeSession 里指定了不同的 model，并行的是三个不同的模型。这就是官方说的"同一输入同时用不同模型处理"。
2. 评委逻辑写在合并策略里——selectByIndex 的 lambda 在所有并行结果就绪后执行，天然适合"汇总后再决策"的场景。

以上为策略使用片段，依赖已配置的执行器和工具；本文没有将它作为真实模型实测输出。

6.4 注意事项

并行不是免费的，官方文档明确列了开销和约束：

• 资源约束：每个并行节点都会创建一个新的协程；多个 LLM API 同时调用要考虑速率限制和成本。
• 上下文是 fork 出来的：每次并行执行都会 fork 一个上下文，合并结果时需要决定保留哪个上下文、或如何组合不同执行的上下文。
• 性能：上下文的 fork 与合并有计算成本，并行数太多会发生资源竞争。

因此官方建议只并行这样的操作：相互独立、执行时间长、不共享可变状态。反过来说，如果一个节点只是做个字符串转换，并行化反而得不偿失。

一句话总结：

parallel() 让多个节点并发执行，四种合并策略决定结果怎么合：selectBy 按条件挑、selectByMax 选最大、selectByIndex 让 LLM 当评委、fold 全要；并行的代价是协程、fork 上下文和资源竞争，只并行独立且耗时的操作。

七、节点间数据传递：AIAgentStorage

7.1 边传递与 storage

节点的返回值沿边传给下游，是最直接的数据流。若较早的结果需要在后面的非相邻节点中读取，可以使用 AIAgentStorage，避免中间节点一直透传与自身逻辑无关的数据。

storage 不是外部数据库，也不是所有并行分支共享的实时白板。并行上下文的行为见 7.4。

7.2 AIAgentStorageKey：名称决定键相等性

通过 createStorageKey<T>(name) 创建类型化的键：

val userDataKey = createStorageKey<UserData>("profile.user-data")

在 Koog 1.3.0 中，AIAgentStorageKey 的 equals 和 hashCode 按 name 判断；不是按对象实例判断。对于相同值类型：

val key1 = createStorageKey<UserData>("profile.user-data")
val key2 = createStorageKey<UserData>("profile.user-data")
check(key1 == key2)
// 在节点执行逻辑中：
// storage.set(key1, UserData("John", 26))
// storage.get(key2) 可以取到上面存入的值。

官方指南曾描述“同名 key 的实例互不相等”，但这与 1.3.0 的发布实现不一致。本文按该版本源码及离线运行结果说明。

类型参数 T 约束单个 key 的存取接口，不负责阻止名称冲突。不要创建同名但不同类型的 key；它们仍然可能命中同一个存储项，造成类型错误。复用组件时建议使用业务前缀，如 profile.user-data、travel.weather。

7.3 示例：明确写入与读取顺序

下面根据官方存取示例补齐图连接，并将数据类声明为可序列化，方便后续讨论持久化。只有内存 set/get 时不要求 @Serializable；这里采用 kotlinx.serialization，需要启用对应编译插件。

import ai.koog.agents.core.agent.entity.createStorageKey
import ai.koog.agents.core.dsl.builder.node
import ai.koog.agents.core.dsl.builder.strategy
import kotlinx.serialization.Serializable

@Serializable
data class UserData(val name: String, val age: Int)

val userDataKey = createStorageKey<UserData>("profile.user-data")

val storageStrategy = strategy<String, String>("storage-example") {
    val saveData by node<String, String> { message ->
        storage.set(userDataKey, UserData("John", 26))
        message
    }
    val retrieveData by node<String, String> { message ->
        val user = storage.getValue(userDataKey)
        "Hello ${user.name}, here's a message for you: $message"
    }

    nodeStart then saveData then retrieveData then nodeFinish
}

存用 set(key, value)；get(key) 在键不存在时返回 null，getValue(key) 则抛出 NoSuchElementException。示例由执行顺序保证先写后读，所以使用 getValue。

这两个节点不必直接相连：可以在中间插入其他节点。但它们仍需接入执行路径，且必须保证读取前已写入。storage 不会调度节点执行。

7.4 并行上下文与合并规则

1.3.0 的 parallel 会为每条分支 fork 上下文，包含 storage.copy()：

• 分支可以读取 fork 前已有的存储项。
• 分支里的 storage.set(...) 修改自己的副本，不会自动广播到其他分支。
• selectBy、selectByMax、selectByIndex 返回被选中结果及其上下文，下游因此沿用该分支的状态。
• fold 聚合输出值，保留基础上下文，不会自动合并各分支的 storage 写入或消息历史。

例如，基础值为 base，两个分支分别写入 A、B 并返回同名字符串：fold 可以得到输出 AB，但下游读取的基础值仍为 base；若选择第一个分支，下游值则为 A。本文已用离线执行验证这两种情况。

还要注意：storage 的复制是容器层面的浅复制，不保证任意可变值对象都会深拷贝。不要把共享可变对象塞入 storage 后，依赖分支原地修改它来通信。

需要汇总多分支数据时，优先让各分支显式返回结果，在合并步骤聚合，再由后续节点写入 storage。不要假定“有 Mutex”就等于“分支共享一个存储”或“读—改—写具有原子性”。

7.5 序列化与持久化边界

set/get 直接访问内存对象，不执行 JSON 序列化。序列化相关 API 则有不同要求：

该版本没有本文原稿所列的 toMap()。toSerializedMap() 会忽略无法被当前 JSONSerializer 序列化的值，不能把“能存进 storage”理解为“肯定能保存到 checkpoint”。

使用 kotlinx.serialization 时，通常为自定义模型加 @Serializable，并确保其字段也受支持；如果配置其他序列化器，则以该序列化器的要求为准。

Agent Persistence 需要另外安装与配置。只有成功序列化、纳入 checkpoint 且能够在恢复时反序列化的数据，才能被可靠还原。本文离线校验覆盖了 storage 的序列化导出、导入和读取，不等同于完整的进程重启、checkpoint 恢复测试。具体存储后端、恢复一致性和外部副作用将在后续文章展开。

边表达主数据流，storage 保存上下文内的辅助状态；键名需要避免冲突，并行分支状态需要显式合并，持久化则必须满足序列化与 checkpoint 配置要求。

八、综合示例：确定性查天气，再由模型生成建议

本章是自行设计的教学示例，不是官方天气服务示例。重点是把必须由程序保证的步骤与由模型生成的内容分开。

8.1 需求与流程

输入使用结构化的 TravelRequest(city, intent)，只讨论今天的出行。自然语言提取城市、日期不在本例范围内。

1. 在子图中直接执行天气工具，不让模型决定是否查询。
2. 工具返回结构化 Weather，依据 isRainy 做互补分支，不匹配“雨”字。
3. 程序组装室内或户外的活动限制，两个 LLM 节点并行生成简洁版与详细版。
4. fold 合并文本，收尾节点从基础 storage 读取先前保存的天气。

flowchart TD start([nodeStart]) --> research["research 子图：直接执行天气工具"] research --> save["saveWeather：保存结构化结果"] save -->|"isRainy"| indoor["室内活动限制"] save -->|"!isRainy"| outdoor["户外活动限制"] indoor --> polish["parallel：简洁版 + 详细版"] outdoor --> polish polish --> compose["fold 合并；读取基础 storage"] compose --> finish([nodeFinish])

天气工具只返回北京、上海的固定演示数据，未知城市明确失败。没有访问真实天气 API，不能将下列数据当成天气预报。

8.2 可编译的 Kotlin 示例

以下代码可放入一个 Kotlin 文件。校验环境使用 JDK 21、Kotlin 2.4.10、kotlinx.serialization 编译插件，依赖如下：

implementation("ai.koog:koog-agents:1.3.0")
implementation("ai.koog:koog-agents-additions:1.3.0-beta")

本例的 DeepSeek 客户端来自 additions 依赖中的 Beta 模块，版本与 API 稳定性需要单独留意。运行入口读取 DEEPSEEK_API_KEY，会调用真实模型并产生费用；不要把密钥写入源码。

import ai.koog.agents.core.agent.AIAgent
import ai.koog.agents.core.agent.entity.createStorageKey
import ai.koog.agents.core.dsl.builder.*
import ai.koog.agents.core.dsl.extension.*
import ai.koog.agents.core.environment.SafeTool
import ai.koog.agents.core.tools.Tool
import ai.koog.agents.core.tools.ToolRegistry
import ai.koog.prompt.executor.clients.deepseek.DeepSeekLLMClient
import ai.koog.prompt.executor.clients.deepseek.DeepSeekModels
import ai.koog.prompt.executor.llms.MultiLLMPromptExecutor
import ai.koog.prompt.executor.model.PromptExecutor
import ai.koog.prompt.message.MessagePart
import ai.koog.serialization.typeToken
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlinx.serialization.Serializable

@Serializable
data class TravelRequest(val city: String, val intent: String)

@Serializable
data class WeatherArgs(val city: String)

@Serializable
data class Weather(val city: String, val summary: String, val isRainy: Boolean)

// 固定测试数据，不是真实天气服务；未知城市明确失败。
class WeatherTool : Tool<WeatherArgs, Weather>(
    argsType = typeToken<WeatherArgs>(),
    resultType = typeToken<Weather>(),
    name = "get_weather",
    description = "查询指定城市今天的天气（演示数据）"
) {
    override suspend fun execute(args: WeatherArgs): Weather = when (args.city) {
        "北京" -> Weather("北京", "小雨，18 度", true)
        "上海" -> Weather("上海", "晴，26 度", false)
        else -> error("演示天气数据不支持城市：${args.city}")
    }
}

fun travelStrategy(weatherTool: WeatherTool) =
    strategy<TravelRequest, String>("travel-advisor") {
        val weatherKey = createStorageKey<Weather>("travel.weather")
        val requestKey = createStorageKey<TravelRequest>("travel.request")

        // 确定性执行工具，不让 LLM 决定是否查询、查询哪个城市。
        val research by subgraph<TravelRequest, Weather>(
            name = "research_weather", tools = listOf(weatherTool)
        ) {
            val toArgs by node<TravelRequest, WeatherArgs> { request ->
                require(request.city.isNotBlank()) { "城市不能为空" }
                storage.set(requestKey, request)
                WeatherArgs(request.city)
            }
            val queryWeather by nodeExecuteSingleTool<WeatherArgs, Weather>(
                tool = weatherTool, doAppendPrompt = false
            )
            val unwrap by node<SafeTool.Result<Weather>, Weather> { result ->
                when (result) {
                    is SafeTool.Result.Success -> result.result
                    is SafeTool.Result.Failure -> error("天气查询失败：${result.message}")
                }
            }
            nodeStart then toArgs then queryWeather then unwrap then nodeFinish
        }

        val saveWeather by node<Weather, Weather> { weather ->
            storage.set(weatherKey, weather)
            weather
        }
        val indoorAdvice by node<Weather, String> { weather ->
            "城市：${weather.city}；今天：${weather.summary}；" +
                "出行意图：${storage.getValue(requestKey).intent}。仅建议室内活动，提醒携带雨具。"
        }
        val outdoorAdvice by node<Weather, String> { weather ->
            "城市：${weather.city}；今天：${weather.summary}；" +
                "出行意图：${storage.getValue(requestKey).intent}。可以考虑户外活动，提醒防晒补水。"
        }

        val shortVersion by node<String, String> { context ->
            llm.writeSession {
                // 每个分支独立设置任务，避免累积互相冲突的 system 指令。
                rewritePrompt {
                    ai.koog.prompt.dsl.prompt("travel-short") {
                        system("根据给定的今天的天气和活动限制，用中文一句话概括建议，不改变活动类型。")
                        user(context)
                    }
                }
                val response = requestLLMWithoutTools()
                val text = response.parts.filterIsInstance<MessagePart.Text>()
                    .joinToString("\n") { it.text }
                check(text.isNotBlank()) { "模型没有返回简洁版文本" }
                "【简洁版】$text"
            }
        }
        val detailedVersion by node<String, String> { context ->
            llm.writeSession {
                rewritePrompt {
                    ai.koog.prompt.dsl.prompt("travel-detailed") {
                        system("根据给定的今天的天气和活动限制，生成中文详细建议与注意事项，不改变活动类型。")
                        user(context)
                    }
                }
                val response = requestLLMWithoutTools()
                val text = response.parts.filterIsInstance<MessagePart.Text>()
                    .joinToString("\n") { it.text }
                check(text.isNotBlank()) { "模型没有返回详细版文本" }
                "【详细版】$text"
            }
        }
        val polish by parallel<String, String>(shortVersion, detailedVersion) {
            fold("出行建议：") { acc, text -> "$acc\n\n$text" }
        }
        val composeFinal by node<String, String> { advice ->
            val weather = storage.getValue(weatherKey)
            "【今日天气·演示数据】${weather.city}：${weather.summary}\n\n$advice"
        }

        edge(nodeStart forwardTo research)
        edge(research forwardTo saveWeather)
        edge(saveWeather forwardTo indoorAdvice onCondition { it.isRainy })
        edge(saveWeather forwardTo outdoorAdvice onCondition { !it.isRainy })
        edge(indoorAdvice forwardTo polish)
        edge(outdoorAdvice forwardTo polish)
        edge(polish forwardTo composeFinal)
        edge(composeFinal forwardTo nodeFinish)
    }

suspend fun runTravel(
    executor: PromptExecutor,
    request: TravelRequest,
    maxIterations: Int = 30
): String {
    val weatherTool = WeatherTool()
    val agent = AIAgent(
        promptExecutor = executor,
        llmModel = DeepSeekModels.DeepSeekV4Flash,
        strategy = travelStrategy(weatherTool),
        toolRegistry = ToolRegistry { tool(weatherTool) },
        maxIterations = maxIterations
    )
    return try {
        withTimeout(60_000) { agent.run(request) }
    } finally {
        agent.close()
    }
}

fun main() = runBlocking {
    val apiKey = System.getenv("DEEPSEEK_API_KEY")
        ?: error("请设置 DEEPSEEK_API_KEY；本示例会调用真实模型并产生费用")
    val executor = MultiLLMPromptExecutor(DeepSeekLLMClient(apiKey = apiKey))
    try {
        println(runTravel(executor, TravelRequest("北京", "今天参观博物馆")))
    } finally {
        executor.close()
    }
}

8.3 确定性和预算分别保证什么

• 查询由图保证：nodeExecuteSingleTool(tool = ...) 直接调用已注册工具；结果失败就在 unwrap 中终止，不会悄悄伪造晴天继续生成。
• 分支由数据保证：isRainy 来自工具结果，不经过 LLM 总结，所以“无雨”之类的措辞不会影响分支判断。工具注册表仍负责实际执行，子图工具描述的可见性不等于外部系统的权限控制。
• 意图不会丢失：输入请求保存在 travel.request 中，活动限制同时包含城市、天气和意图。
• 模型只负责生成措辞：正常路径发起两次非流式请求。程序保证先查天气、选择对应分支，但提示词不能保证模型完全遵守活动限制。生产场景仍需输出校验；两次请求也不意味着固定 token 成本。
• 并行不写共享状态：天气在 fork 前保存。fold 保留基础上下文，所以最后能读回天气；代码没有依赖并行分支写入互相可见。
• 预算分层设置：maxIterations = 30 是经过离线成功路径验证的图执行上限，不是“最多调用模型 30 次”。另外用 withTimeout(60_000) 限制本次运行的协程时间；真实服务还应配置 HTTP 超时、重试及 token/费用限额。修改图或增加循环后应重新测试预算。
• 资源归属明确：runTravel 关闭自己创建的 Agent；调用方负责关闭传入的执行器。

8.4 已验证范围

本章代码已在 Koog 1.3.0 依赖下编译。离线执行使用明确标记的 PromptExecutor 测试替身，不连接厂商 API，不评估生成质量。校验程序实际打印的结果如下：

PASS travel 北京: branch=仅建议室内 modelCalls=2
PASS travel 上海: branch=可以考虑户外 modelCalls=2
PASS unknown city: explicit failure before model calls
PASS iteration budget: maxIterations=1 rejects execution

其中 modelCalls 是执行器测试替身的调用次数，不是线上模型调用记录。未知城市和过小预算用例预期抛异常；它们验证了失败路径。真实 DeepSeek 联调、网络超时和模型输出质量尚未验证。

九、最佳实践与排错

9.1 最佳实践

官方 Custom strategy graphs 页面给出的建图最佳实践，逐条整理如下：

1. 保持简单：从简单的图开始，按需增加复杂度。不要一上来就画"蜘蛛网"。
2. 描述性命名：给节点和边起有意义的名字，图会好懂得多。
3. 覆盖所有路径和边界情况：确保每条路径最终都能到达结束节点。
4. 用各种输入测试图：确保图的行为符合预期。
5. 给图写文档：记录图的目的和行为，方便将来维护。
6. 从预定义策略或常见模式起步：第五章的三种模式就是标准起手式。
7. 长对话用历史压缩：减少 token 消耗（挂 nodeLLMCompressHistory）。
8. 用子图组织图、管理工具访问：第四章的 multi-part 模式。

写自定义节点时（Custom nodes 页面），还有几条针对性的建议：每个节点只做一件事（单一职责）、为参数写清楚文档、做好错误处理避免整个工作流失败、尽量设计成可复用、合适时用泛型参数化。

9.2 常见问题排查

官方 Troubleshooting 部分列了五个常见问题，都是实战中真的会撞上的：

1. 图走不到结束节点

• 检查是否所有路径最终都通向 nodeFinish；
• 检查条件是不是太严，导致某一步没有任何边能满足；
• 检查循环是否有正常出口；缺少结束边时，文本响应会因没有匹配出边而抛异常，持续的工具调用则可能耗尽迭代预算。

第八章使用 isRainy / !isRainy 两条互补条件，保证布尔状态都有边可走。对更复杂的数据类型，也应覆盖空值、异常状态等边界。

2. 工具调用不执行

• 工具是否正确注册进了 ToolRegistry（第 4 篇讲过的注册流程）；
• LLM 节点到工具执行节点的边，条件是否正确（onToolCalls { true }）；
• 如果在子图里，工具是否在该子图的 tools 子集里。

3. 历史变得太大

• 加历史压缩节点 nodeLLMCompressHistory；
• 用条件检查历史长度，超阈值才压缩：

edge(
    (nodeExecuteTool forwardTo nodeCompressHistory)
        onCondition { _ -> llm.readSession { prompt.messages.size > 100 } }
)

• 用更激进的压缩策略，比如 FromLastNMessages 设更小的 N。

4. 图的行为不符合预期

• 条件定义是否正确；
• 边条件按定义顺序评估——这是本章最想强调的坑。同一个源节点的多条出边，按代码中定义的顺序逐条检查，第一条满足的生效。如果一条宽泛的条件排在具体条件前面，具体条件永远轮不到；
• 是否不小心用更宽泛的条件覆盖了预期条件。

5. 性能问题

• 精简图：删掉不必要的节点和边；
• 相互独立的操作用并行（第六章）；
• 压缩历史；
• 使用更高效的节点和操作。

子图相关的两个问题（来自 Custom subgraphs 页面）也顺带列一下：

• 子图里工具不可用：检查工具是否正确注册在工具注册表里。
• 子图不按预期顺序执行：检查子图之间的连接顺序、输出是否正确传给下一个子图、以及子图是否真的连通——特别注意条件边要覆盖所有情况，否则可能在节点上因无匹配出边抛出 AIAgentStuckInTheNodeException。

十、后记

本文从 Koog(2) 里那张"拿来就用"的图出发，把 Strategy 的内部结构拆开讲了一遍。最后用一张表回顾整个心智模型：

对于本文讨论的 Graph-based Agent，图式策略描述了输入处理、工具执行、分支、并行与数据流。Functional Agent 可以直接用函数组织行为，不应把本篇的图模型泛化到所有策略实现。而搭图永远从三层递进里选：先预定义策略，再预定义节点组装，最后才完全自定义。

图跑起来之后，一个新问题随之而来：多轮会话怎么办？历史越来越长怎么办？Agent 执行到一半挂了怎么办？下一篇我们讲 Koog 的记忆、历史压缩与持久化。

参考资料

• Pre-defined nodes and components: https://docs.koog.ai/nodes-and-components/
• Predefined strategies: https://docs.koog.ai/predefined-agent-strategies/
• Custom strategy graphs: https://docs.koog.ai/custom-strategy-graphs/
• Parallel node execution: https://docs.koog.ai/parallel-node-execution/
• Data transfer between nodes: https://docs.koog.ai/data-transfer-between-nodes/
• Custom nodes: https://docs.koog.ai/custom-nodes/
• Subgraphs overview: https://docs.koog.ai/subgraphs-overview/
• Custom subgraphs: https://docs.koog.ai/custom-subgraphs/