From 1c0296fa80d861e560016257fdf3fa16eeb34bc4 Mon Sep 17 00:00:00 2001 From: unclesam <2036253946@qq.com> Date: Thu, 11 Jun 2026 01:49:21 +0800 Subject: [PATCH] Initial commit --- .gitignore | 8 ++ README.md | 215 +++++++++++++++++++++++++++++++++++ engine.go | 147 ++++++++++++++++++++++++ examples/translation/main.go | 168 +++++++++++++++++++++++++++ go.mod | 3 + graph.go | 50 ++++++++ graph_test.go | 192 +++++++++++++++++++++++++++++++ 7 files changed, 783 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 engine.go create mode 100644 examples/translation/main.go create mode 100644 go.mod create mode 100644 graph.go create mode 100644 graph_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d5bc47f --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.idea +.vscode +settings_dev.yaml +logs/ +testdata/ +tmp +.DS_Store +.Gopher-Graph_开发设计文档.md \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..4a7ab11 --- /dev/null +++ b/README.md @@ -0,0 +1,215 @@ +# GopherGraph + +A lightweight, high-performance, and type-safe Multi-Agent workflow orchestration engine written in Go. +基于 Go 语言（泛型）实现的轻量级、高性能、类型安全的多智能体（Multi-Agent）协同与任务编排引擎。 +Go 言語（ジェネリクス）で書かれた、軽量・高性能・型安全なマルチエージェント・ワークフロー協調エンジン。 + +--- + +* [简体中文](#-简体中文) +* [English](#-english) +* [日本語](#-日本語) + +--- + +## 简体中文 + +### 简介 +`GopherGraph` 是一个用 Go 语言编写的代码即图（Code-as-Graph）智能体编排引擎，设计灵感来源于 Python 生态的 `LangGraph`。它利用 Go 1.18+ 的泛型机制，让工作流的上下文状态在编译期就具备强类型约束，并依靠 Go 原生的 Goroutines 和 Channels 实现极高并发的本地数据流转。它特别适合构建包含复杂循环（Looping）和人机协同（Human-in-the-Loop）的智能体工作流。 + +### 核心特性 +- **强类型状态管理：** 利用 Go 泛型定义全局状态 `Graph[S]`，彻底告别松散的 `map[string]any` 和繁琐的类型断言。 +- **支持循环结构：** 突破了传统 DAG（有向无环图）工作流的限制，原生支持节点间的双向循环和条件路由。 +- **人机协同中断 (HITL)：** 支持在特定节点前自动挂起，返回执行线程快照，等待人工修改状态或审批通过后一键恢复 (`Resume`)。 +- **纯粹的高性能：** 基于内存流转，零外部依赖，极低的上下文切换开销。 + +### 快速开始 + +#### 1. 定义您的状态 (State) +```go +type MyState struct { + Query string + Response string +} +``` + +#### 2. 构建工作流图 +```go +package main + +import ( + "context" + "fmt" + "GopherGraph" +) + +func main() { + // 初始化一个强类型的图构建器 + g := GopherGraph.NewGraph[MyState]() + + // 注册节点 (Node) + g.AddNode("agentA", func(ctx context.Context, s MyState) (MyState, error) { + s.Response = "Hello from Agent A" + return s, nil + }) + + // 建立连线 (Edge) + g.AddEdge("agentA", "anotherNode") + + // 设置中断拦截点 (Interrupt) + g.AddInterrupt("anotherNode") + + // 编译图 + cg, _ := g.Compile() + + // 启动运行 + thread, _ := cg.Start(context.Background(), "agentA", MyState{Query: "Test"}) +} +``` + +#### 运行内置示例 +项目包含了一个完整的“AI翻译 -> 质量检测 -> 人工审核 -> 发布”的交互式 Demo： +```bash +go run examples/translation/main.go +``` + +#### 运行单元测试 +```bash +go test -v ./... +``` + +--- + +## English + +### Introduction +`GopherGraph` is a Code-as-Graph agent orchestration engine built in Go, inspired by Python's `LangGraph`. By leveraging Go 1.18+ Generics, GopherGraph ensures that workflow states are strictly typed at compile-time. Powered by native Goroutines and Channels, it executes agent communication with microsecond latency, making it the perfect engine for building complex looping agent workflows with Human-in-the-Loop (HITL) requirements. + +### Key Features +- **Strictly Typed State:** Bind your custom struct to `Graph[S]` via Go generics. Say goodbye to unsafe `map[string]any` and runtime type assertions. +- **Support for Cycles/Loops:** Unlike traditional DAG (Directed Acyclic Graph) engines, GopherGraph natively supports loops and dynamic routing based on agent evaluation. +- **Human-in-the-Loop (HITL):** Pause workflow execution *before* a designated node, capture a snapshot (`Thread`), modify the state, and `Resume` seamlessly. +- **Zero-Dependency & High-Performance:** Written in pure Go with zero external dependencies, leveraging in-memory queues for lightning-fast orchestration. + +### Quick Start + +#### 1. Define Your State +```go +type MyState struct { + Query string + Response string +} +``` + +#### 2. Define and Run the Graph +```go +package main + +import ( + "context" + "fmt" + "GopherGraph" +) + +func main() { + // Initialize the strongly-typed graph builder + g := GopherGraph.NewGraph[MyState]() + + // Add nodes (agents) + g.AddNode("agentA", func(ctx context.Context, s MyState) (MyState, error) { + s.Response = "Hello from Agent A" + return s, nil + }) + + // Add edges + g.AddEdge("agentA", "anotherNode") + + // Add interrupts + g.AddInterrupt("anotherNode") + + // Compile + cg, _ := g.Compile() + + // Start execution + thread, _ := cg.Start(context.Background(), "agentA", MyState{Query: "Test"}) +} +``` + +#### Run the Interactive Demo +An interactive "Translation -> Review -> Human Approval -> Publish" demo is included: +```bash +go run examples/translation/main.go +``` + +#### Run Unit Tests +```bash +go test -v ./... +``` + +--- + +## 日本語 + +### 概要 +`GopherGraph` は、Python エコシステムの `LangGraph` に着想を得て開発された、Go 言語向けの Code-as-Graph 型エージェントオーケストレーションエンジンです。Go 1.18+ のジェネリクスを活用することで、ワークフローの状態（State）をコンパイル時に厳密に型定義できます。Go 純正の Goroutines と Channels を利用したメモリ内メッセージングにより、高い並行性と極めて低いレイテンシを実現しています。自律的なループ（Looping）や人間参加型（Human-in-the-Loop）の意思決定を伴うエージェントワークフローの開発に最適です。 + +### 主な特徴 +- **型安全な状態管理:** ジェネリクスを用いて `Graph[S]` に構造体をバインドします。冗長な `map[string]any` やランタイム時の型アサーションから解放されます。 +- **ループと条件付き分岐のサポート:** 従来の DAG（有向非巡回グラフ）の制限を超え、エージェントの判定に基づく条件付きルートや双方向のループをサポート。 +- **Human-in-the-Loop (HITL) の一時停止と再開:** 特定のノードの実行前に処理を自動停止し、スナップショット（`Thread`）を返します。承認や状態修正の後にシームレスに処理を再開（`Resume`）できます。 +- **ピュア Go & 高性能:** 外部依存関係ゼロ。メモリ内チャネルを用いた高速なコンテキスト切り替え。 + +### クイックスタート + +#### 1. 状態（State）の定義 +```go +type MyState struct { + Query string + Response string +} +``` + +#### 2. ワークフロー図の構築と実行 +```go +package main + +import ( + "context" + "fmt" + "GopherGraph" +) + +func main() { + // 型安全なグラフビルダーの初期化 + g := GopherGraph.NewGraph[MyState]() + + // ノード（エージェント）の追加 + g.AddNode("agentA", func(ctx context.Context, s MyState) (MyState, error) { + s.Response = "Hello from Agent A" + return s, nil + }) + + // エッジ（遷移線）の接続 + g.AddEdge("agentA", "anotherNode") + + // 割り込み（一時停止）ポイントの設定 + g.AddInterrupt("anotherNode") + + // コンパイル + cg, _ := g.Compile() + + // 実行開始 + thread, _ := cg.Start(context.Background(), "agentA", MyState{Query: "Test"}) +} +``` + +#### インタラクティブデモの実行 +「AI翻訳 -> 監査 -> 人間による確認 -> 公開」の一連のループを体験できるデモを用意しています： +```bash +go run examples/translation/main.go +``` + +#### ユニットテストの実行 +```bash +go test -v ./... +``` diff --git a/engine.go b/engine.go new file mode 100644 index 0000000..dd2c2a6 --- /dev/null +++ b/engine.go @@ -0,0 +1,147 @@ +package GopherGraph + +import ( + "context" + "errors" + "fmt" +) + +// Thread 代表图执行的具体实例快照（线程上下文） +// 它可以被序列化并存储，用来实现中断后的状态恢复 +type Thread[S any] struct { + State S // 当前共享的强类型状态数据 + NextNode string // 下一步即将执行的节点名称 + IsPaused bool // 是否因为触发中断而处于暂停状态 + IsFinished bool // 工作流是否全部结束（即到达了终点） +} + +// CompiledGraph 是编译后的只读图，准备好投入运行。 +type CompiledGraph[S any] struct { + nodes map[string]NodeFn[S] + edges map[string]string + conditional map[string]RouterFn[S] + interrupts map[string]bool +} + +// Compile 校验图结构的合法性，并生成可运行的 CompiledGraph。 +func (g *Graph[S]) Compile() (*CompiledGraph[S], error) { + if len(g.nodes) == 0 { + return nil, errors.New("cannot compile graph: graph contains no nodes") + } + + // 校验静态边 (From -> To) 的起始和目标节点是否存在 + for from, to := range g.edges { + if _, exists := g.nodes[from]; !exists { + return nil, fmt.Errorf("compile error: edge origin %q does not exist", from) + } + if _, exists := g.nodes[to]; !exists { + return nil, fmt.Errorf("compile error: edge origin %q does not exist", to) + } + } + + // 校验条件路由边的源节点是否存在 + for from := range g.conditional { + if _, exists := g.nodes[from]; !exists { + return nil, fmt.Errorf("compile error: conditional edge origin %q does not exist", from) + } + } + + // 校验被标记为中断的节点是否存在 + for node := range g.interrupts { + if _, exists := g.nodes[node]; !exists { + return nil, fmt.Errorf("compile error: interrupt node %q does not exist", node) + } + } + + // 返回编译好的只读图 + return &CompiledGraph[S]{ + nodes: g.nodes, + edges: g.edges, + conditional: g.conditional, + interrupts: g.interrupts, + }, nil +} + +// Start 从指定的起始节点开始运行图，并持续流转，直到遇到中断或运行结束 +func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initialState S) (*Thread[S], error) { + thread := &Thread[S]{ + State: initialState, + NextNode: startNode, + } + + return cg.run(ctx, thread) +} + +// Resume 恢复执行一个被暂停的线程，并允许注入外部修改后的状态数据（例如人工审批修改后的结果） +func (cg *CompiledGraph[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) { + if !thread.IsPaused { + return nil, errors.New("cannot resume: thread is not paused") + } + if thread.IsFinished { + return nil, errors.New("cannot resume: thread is already finished") + } + + // 注入人工修改后的状态，并解除暂停标记 + thread.State = modifiedState + thread.IsPaused = false + + return cg.run(ctx, thread) +} + +// run 是引擎内部循环调度器，负责驱动节点向前流转 +func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S]) (*Thread[S], error) { + for { + // 检查 Context，支持外部超时控制或取消 + if err := ctx.Err(); err != nil { + return thread, err + } + + currentNodeName := thread.NextNode + // 如果下一个执行节点为空，说明整条流水线已运行结束 + if currentNodeName == "" { + thread.IsFinished = true + return thread, nil + } + + // 查找并执行当前节点 + nodeFn, exists := cg.nodes[currentNodeName] + if !exists { + return thread, fmt.Errorf("runtime error: node %q not found", currentNodeName) + } + + newState, err := nodeFn(ctx, thread.State) + if err != nil { + return thread, fmt.Errorf("node %q execution error: %w", currentNodeName, err) + } + thread.State = newState + + // 计算下一个该执行的节点名称 + var nextNode string + if routerFn, ok := cg.conditional[currentNodeName]; ok { + // 如果有条件路由函数，则通过路由函数动态计算去向 + next, err := routerFn(ctx, thread.State) + if err != nil { + return thread, fmt.Errorf("router for node %q execution error: %w", currentNodeName, err) + } + nextNode = next + + } else { + // 否则使用静态连线边 + nextNode = cg.edges[currentNodeName] + } + + // 更新快照中的“下一个节点” + thread.NextNode = nextNode + + // 如果下一站是终点，直接进入下一次循环触发结束逻辑 + if nextNode == "" { + continue + } + + // 核心中断机制：如果即将进入的下一个节点被标记为了中断节点，则在此挂起 + if cg.interrupts[nextNode] { + thread.IsPaused = true + return thread, nil // 暂停执行，返回当前快照供外部人工介入 + } + } +} diff --git a/examples/translation/main.go b/examples/translation/main.go new file mode 100644 index 0000000..e2e6400 --- /dev/null +++ b/examples/translation/main.go @@ -0,0 +1,168 @@ +package main + +import ( + "GopherGraph" + "bufio" + "context" + "fmt" + "os" + "strings" +) + +// TranslationState 代表我们工作流的共享状态（State） +type TranslationState struct { + InputText string // 原文（中文） + TranslatedText string // 译文（日语） + ReviewNotes string // 质检员反馈意见 + Approved bool // 是否已获批准 +} + +// 翻译员 Agent (Node) +func translatorNode(ctx context.Context, state TranslationState) (TranslationState, error) { + fmt.Printf("[翻译员] 正在翻译原文: %q...\n", state.InputText) + + // 模拟翻译：如果是被拒绝退回的（含有 ReviewNotes），我们进行修正 + if state.ReviewNotes != "" { + fmt.Printf("[翻译员] 收到质检意见：%q，正在修正...\n", state.ReviewNotes) + if strings.Contains(state.InputText, "你好") { + state.TranslatedText = "こんにちは（修正版）" + } else { + state.TranslatedText = state.InputText + " (修正後の日本語訳)" + } + } else { + // 初次翻译 + if strings.Contains(state.InputText, "你好") { + state.TranslatedText = "こんにちは" + } else { + state.TranslatedText = state.InputText + " (日本語訳)" + } + } + state.ReviewNotes = "" // 翻译完成，清空历史意见 + return state, nil +} + +// 质检员 Agent (Node) +func reviewerNode(ctx context.Context, state TranslationState) (TranslationState, error) { + fmt.Printf("[质检员] 正在审核译文: %q...\n", state.TranslatedText) + + // 模拟审核：如果译文是完美的 "こんにちは" 或者包含 "修正" 字样，则通过；否则退回 + if state.TranslatedText == "こんにちは" { + state.Approved = true + fmt.Println("[质检员] 审核通过！直接发布。") + } else if strings.Contains(state.TranslatedText, "修正") { + state.Approved = true + fmt.Println("[质检员] 修正版审核通过！") + } else { + state.Approved = false + state.ReviewNotes = "翻译过于生硬，请重新翻译并润色" + fmt.Println("[质检员] 审核不通过，建议退回重写。") + } + return state, nil +} + +// 人工审核节点（这个节点被标记为中断点，运行前会暂停） +func humanReviewNode(ctx context.Context, state TranslationState) (TranslationState, error) { + // 这个节点只有在 Resume 恢复执行后才会运行 + fmt.Printf("[系统] 人工审核节点已被触发。当前审批状态 Approved = %t\n", state.Approved) + return state, nil +} + +// 发布员 Agent (Node) +func publisherNode(ctx context.Context, state TranslationState) (TranslationState, error) { + fmt.Printf("🎉 [发布员] 恭喜！内容发布成功！\n最终译文: %s\n", state.TranslatedText) + return state, nil +} + +// 路由函数：决定质检后去哪里 +func reviewRouter(ctx context.Context, state TranslationState) (string, error) { + if state.Approved { + return "publisher", nil // 审核通过，去发布 + } + + // 审核不通过时：如果原文包含敏感词“秘密”，必须人工审核 + if strings.Contains(state.InputText, "秘密") { + fmt.Println("[路由逻辑] 检测到敏感内容，必须路由到 [human_review] 进行人工审批") + return "human_review", nil + } + + // 否则，普通翻译错误，直接退回给翻译员重新处理 + fmt.Println("[路由逻辑] 普通翻译错误，退回给 [translator] 重新处理") + return "translator", nil +} + +// 路由函数：决定人工审核完去哪里 +func humanRouter(ctx context.Context, state TranslationState) (string, error) { + if state.Approved { + return "publisher", nil // 人工同意，去发布 + } + return "translator", nil // 人工拒绝，回滚到翻译员重写 +} + +func main() { + // 1. 创建图并注册节点 + g := GopherGraph.NewGraph[TranslationState]() + g.AddNode("translator", translatorNode) + g.AddNode("reviewer", reviewerNode) + g.AddNode("human_review", humanReviewNode) + g.AddNode("publisher", publisherNode) + // 2. 建立连线 + g.AddEdge("translator", "reviewer") + g.AddConditionalEdges("reviewer", reviewRouter) + g.AddConditionalEdges("human_review", humanRouter) + // 3. 标记 human_review 节点为中断点（进入该节点前暂停） + g.AddInterrupt("human_review") + // 4. 编译图 + cg, err := g.Compile() + if err != nil { + fmt.Printf("编译失败: %v\n", err) + return + } + // 5. 启动图 + // 包含敏感词“秘密”，会触发人工审核中断： + inputText := "这是一个秘密你好" + fmt.Printf("🚀 启动工作流，输入原文: %q\n", inputText) + ctx := context.Background() + thread, err := cg.Start(ctx, "translator", TranslationState{InputText: inputText}) + if err != nil { + fmt.Printf("运行失败: %v\n", err) + return + } + // 6. 循环检查线程状态，处理中断 + reader := bufio.NewReader(os.Stdin) + for { + if thread.IsFinished { + break + } + if thread.IsPaused { + fmt.Println("\n==================================================") + fmt.Printf("⚠️ [工作流挂起] 遇到中断节点: %s\n", thread.NextNode) + fmt.Printf("👉 当前状态：\n - 原文: %s\n - 译文: %s\n", thread.State.InputText, thread.State.TranslatedText) + fmt.Print("🤔 请审批：[y] 批准发布 / [n] 拒绝退回 / [或者直接输入你修改后的译文]: ") + input, _ := reader.ReadString('\n') + input = strings.TrimSpace(input) + // 根据人工输入更新状态 + currentState := thread.State + if strings.ToLower(input) == "y" { + currentState.Approved = true + fmt.Println("👍 人工审核：批准发布") + } else if strings.ToLower(input) == "n" { + currentState.Approved = false + currentState.ReviewNotes = "人工审核拒绝：请重新润色" + fmt.Println("👎 人工审核：拒绝，退回重写") + } else { + // 用户直接输入了修改后的译文 + currentState.TranslatedText = input + currentState.Approved = true + fmt.Printf("✍️ 人工审核：修改译文为 %q 并批准发布\n", input) + } + fmt.Println("==================================================") + // 7. 恢复执行，并将修改后的状态注入 + thread, err = cg.Resume(ctx, thread, currentState) + if err != nil { + fmt.Printf("恢复失败: %v\n", err) + return + } + } + } + fmt.Println("\n🏁 工作流顺利结束！") +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..31e85e7 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module GopherGraph + +go 1.26.2 diff --git a/graph.go b/graph.go new file mode 100644 index 0000000..abe9b36 --- /dev/null +++ b/graph.go @@ -0,0 +1,50 @@ +package GopherGraph + +import "context" + +// NodeFn 代表图中的一个执行节点（Agent）。 +// 它接收当前的上下文和状态 S，处理后返回更新后的状态 S，或者返回错误。 +type NodeFn[S any] func(ctx context.Context, state S) (S, error) + +// RouterFn 代表条件路由函数。 +// 它根据当前状态 S，动态决定下一个要执行的节点名称（例如返回 "translator" 或 "reviewer" +type RouterFn[S any] func(ctx context.Context, state S) (string, error) + +// Graph 是图的构建器，S 代表用户自定义的状态结构体。 +type Graph[S any] struct { + nodes map[string]NodeFn[S] // 节点名称 -> 节点执行函数 + edges map[string]string // 普通单向边：源节点 -> 目标节点 + conditional map[string]RouterFn[S] // 条件路由边：源节点 -> 路由函数 + interrupts map[string]bool // 标记需要中断的节点：执行完该节点后挂起 +} + +// NewGraph 创建并初始化一个全新强类型的图构建器 +func NewGraph[S any]() *Graph[S] { + return &Graph[S]{ + nodes: make(map[string]NodeFn[S]), + edges: make(map[string]string), + conditional: make(map[string]RouterFn[S]), + interrupts: make(map[string]bool), + } +} + +// AddNode 向图中注册一个节点（Agent） +func (g *Graph[S]) AddNode(name string, fn NodeFn[S]) { + g.nodes[name] = fn +} + +// AddEdge 建立一条从 from 节点到 to 节点的静态连接线 +func (g *Graph[S]) AddEdge(from, to string) { + g.edges[from] = to +} + +// AddConditionalEdges 建立一条从 from 节点出发的条件路由 +// 到底去哪，由 router 函数在运行时根据当前 State 动态决定 +func (g *Graph[S]) AddConditionalEdges(from string, router RouterFn[S]) { + g.conditional[from] = router +} + +// AddInterrupt 标记在执行 nodeName 节点“之前”进行中断挂起，等待人工确认 +func (g *Graph[S]) AddInterrupt(nodeName string) { + g.interrupts[nodeName] = true +} diff --git a/graph_test.go b/graph_test.go new file mode 100644 index 0000000..3d6b8da --- /dev/null +++ b/graph_test.go @@ -0,0 +1,192 @@ +package GopherGraph + +import ( + "context" + "errors" + "testing" + "time" +) + +// TestState 是测试用的强类型状态 +type TestState struct { + Value int + Log []string +} + +// TestSequentialExecution 测试最基础的线性工作流 (A -> B) +func TestSequentialExecution(t *testing.T) { + g := NewGraph[TestState]() + + g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { + s.Value += 1 + s.Log = append(s.Log, "A") + return s, nil + }) + g.AddNode("B", func(ctx context.Context, s TestState) (TestState, error) { + s.Value *= 2 + s.Log = append(s.Log, "B") + return s, nil + }) + + g.AddEdge("A", "B") + + cg, err := g.Compile() + if err != nil { + t.Fatalf("编译图失败: %v", err) + } + + thread, err := cg.Start(context.Background(), "A", TestState{Value: 5}) + if err != nil { + t.Fatalf("启动图失败: %v", err) + } + + // 验证最终状态 + if !thread.IsFinished { + t.Errorf("期望工作流运行结束，实际未结束") + } + // 计算公式: (5 + 1) * 2 = 12 + if thread.State.Value != 12 { + t.Errorf("期望 Value 为 12，实际为 %d", thread.State.Value) + } + if len(thread.State.Log) != 2 || thread.State.Log[0] != "A" || thread.State.Log[1] != "B" { + t.Errorf("执行路径日志不匹配，实际为: %v", thread.State.Log) + } +} + +// TestConditionalRouting 测试路由条件分支跳转 (start -> even/odd) +func TestConditionalRouting(t *testing.T) { + g := NewGraph[TestState]() + + g.AddNode("start", func(ctx context.Context, s TestState) (TestState, error) { + s.Log = append(s.Log, "start") + return s, nil + }) + g.AddNode("even", func(ctx context.Context, s TestState) (TestState, error) { + s.Log = append(s.Log, "even") + return s, nil + }) + g.AddNode("odd", func(ctx context.Context, s TestState) (TestState, error) { + s.Log = append(s.Log, "odd") + return s, nil + }) + + g.AddConditionalEdges("start", func(ctx context.Context, s TestState) (string, error) { + if s.Value%2 == 0 { + return "even", nil + } + return "odd", nil + }) + + cg, err := g.Compile() + if err != nil { + t.Fatalf("编译图失败: %v", err) + } + + // 测试偶数分支 + thread1, err := cg.Start(context.Background(), "start", TestState{Value: 4}) + if err != nil { + t.Fatalf("启动图失败: %v", err) + } + if thread1.State.Log[1] != "even" { + t.Errorf("偶数测试路由错误，实际路径: %v", thread1.State.Log) + } + + // 测试奇数分支 + thread2, err := cg.Start(context.Background(), "start", TestState{Value: 7}) + if err != nil { + t.Fatalf("启动图失败: %v", err) + } + if thread2.State.Log[1] != "odd" { + t.Errorf("奇数测试路由错误，实际路径: %v", thread2.State.Log) + } +} + +// TestInterruptAndResume 测试中断挂起与恢复 (A -> [Interrupt B] -> C) +func TestInterruptAndResume(t *testing.T) { + g := NewGraph[TestState]() + + g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { + s.Value += 10 + return s, nil + }) + g.AddNode("B", func(ctx context.Context, s TestState) (TestState, error) { + s.Value += 20 + return s, nil + }) + g.AddNode("C", func(ctx context.Context, s TestState) (TestState, error) { + s.Value += 30 + return s, nil + }) + + g.AddEdge("A", "B") + g.AddEdge("B", "C") + + // 标记在执行 B 之前进行中断 + g.AddInterrupt("B") + + cg, err := g.Compile() + if err != nil { + t.Fatalf("编译图失败: %v", err) + } + + // 1. 启动图，应该在执行 B 之前停下来 + thread, err := cg.Start(context.Background(), "A", TestState{Value: 0}) + if err != nil { + t.Fatalf("启动图失败: %v", err) + } + + // 验证是否成功停在 B 之前 + if !thread.IsPaused { + t.Errorf("期望工作流处于 Paused 状态") + } + if thread.NextNode != "B" { + t.Errorf("期望下一个节点是 B，实际是 %q", thread.NextNode) + } + if thread.State.Value != 10 { // 只执行了 A: 0 + 10 = 10 + t.Errorf("期望 Value 为 10，实际为 %d", thread.State.Value) + } + + // 2. 模拟人工介入：修改状态值为 100，并调用 Resume 恢复执行 + thread, err = cg.Resume(context.Background(), thread, TestState{Value: 100}) + if err != nil { + t.Fatalf("恢复图失败: %v", err) + } + + // 验证是否顺利走完剩余的 B 和 C + if !thread.IsFinished { + t.Errorf("期望工作流已结束") + } + // 计算公式: B 执行 (100 + 20 = 120) -> C 执行 (120 + 30 = 150) + if thread.State.Value != 150 { + t.Errorf("期望最终 Value 为 150，实际为 %d", thread.State.Value) + } +} + +// TestTimeoutCancellation 测试通过 Context 对长时间运行的节点进行超时退出控制 +func TestTimeoutCancellation(t *testing.T) { + g := NewGraph[TestState]() + + g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { + select { + case <-time.After(100 * time.Millisecond): + s.Value = 1 + return s, nil + case <-ctx.Done(): + return s, ctx.Err() + } + }) + + cg, err := g.Compile() + if err != nil { + t.Fatalf("编译图失败: %v", err) + } + + // 创建一个 20 毫秒的超短超时 Context，而节点 A 需要 100 毫秒才能跑完 + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + _, err = cg.Start(ctx, "A", TestState{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("期望返回 context.DeadlineExceeded 错误，实际返回: %v", err) + } +}