5 Commits
Author SHA1 Message Date
unclesam 7028571a4f docs: update README to include Checkpointer.Delete example 2026-07-04 20:47:59 +08:00
unclesam 5782504351 fix(engine): resolve parallel execution races and improve max steps semantics
- Fix data race in runParallelBranches where first error could be masked
- Improve MaxSteps to allow exactly N steps
- Consolidate Resume validation logic and fix silent overwrite in Graph AddNode/AddEdge
- Add sync.RWMutex to Engine for thread-safe configuration
- Fix P2 issues: Add immediate check for router output, add mutex to hooks example, add context timeout to translation example, and update docs for Checkpointer.Delete
2026-07-04 20:46:36 +08:00
unclesam d2d8eb5a3a docs: 更新README以反映新增特性与安全校验 2026-06-17 11:02:46 +08:00
unclesam de457f670a feat: production-grade hardening v1.1.0
- fix: Compile() defensive map copy, prevents post-compile Graph mutation
- fix: correct 'origin'→'destination' in edge validation error message
- feat: export ErrNotPaused/ErrAlreadyFinished for errors.Is() support
- feat: WithPreNodeHook/WithPostNodeHook now support chaining (multiple hooks)
- feat: Compile detects conflicting outgoing edge types on same node
- fix: FileCheckpointer atomic write (tmp+rename, crash-safe)
- fix: FileCheckpointer threadID path traversal validation
- refactor: remove redundant continue in run() loop
- docs: add doc.go package-level documentation
- test: +7 test cases covering error paths, edge conflicts, hook chaining
2026-06-17 10:44:59 +08:00
unclesam 235f66c9ba Fix(README.md) GopherGraph import path casing
修复 README.md 示例错误导入路径
2026-06-13 21:16:32 +08:00
11 changed files with 657 additions and 80 deletions

No files matched your search

+30 -12
View File
@@ -22,10 +22,11 @@ Go 言語(ジェネリクス)で書かれた、軽量・高性能・型安
- **支持循环结构:** 突破了传统 DAG(有向无环图)工作流的限制,原生支持节点间的双向循环和条件路由。 - **支持循环结构:** 突破了传统 DAG(有向无环图)工作流的限制,原生支持节点间的双向循环和条件路由。
- **并发分支与短路取消:** 原生支持多个 Agent 节点的并发执行,基于 `context.WithCancel` 实现短路机制——任意分支出错立即取消其余兄弟 goroutine,避免算力空转。 - **并发分支与短路取消:** 原生支持多个 Agent 节点的并发执行,基于 `context.WithCancel` 实现短路机制——任意分支出错立即取消其余兄弟 goroutine,避免算力空转。
- **并发安全深拷贝:** 通过 `Engine.WithStateCloner` 注入自定义深拷贝函数,彻底消除含引用类型的状态在并发分支中的数据竞争。 - **并发安全深拷贝:** 通过 `Engine.WithStateCloner` 注入自定义深拷贝函数,彻底消除含引用类型的状态在并发分支中的数据竞争。
- **人机协同中断 (HITL):** 支持在特定节点前自动挂起,返回执行线程快照,等待人工修改状态或审批通过后一键恢复 (`Resume`)。 - **人机协同中断 (HITL):** 支持在特定节点前自动挂起,返回执行线程快照,等待人工修改状态或审批通过后一键恢复 (`Resume`)。支持 `ErrNotPaused`、`ErrAlreadyFinished` 哨兵错误,便于精准判断异常。
- **死循环硬性熔断:** 通过 `Engine.WithMaxSteps` 设置步数上限,图中的任何意外环路都会被明确报错而非无限阻塞。 - **死循环硬性熔断:** 通过 `Engine.WithMaxSteps` 设置步数上限,图中的任何意外环路都会被明确报错而非无限阻塞。
- **轻量级生命周期 Hooks:** `Engine.WithPreNodeHook` / `WithPostNodeHook` 提供节点执行前后的钩子,无需修改任何 `NodeFn`,即可接入日志、链路追踪(OpenTelemetry)或流式输出。 - **轻量级生命周期 Hooks(支持多重链式累加):** `Engine.WithPreNodeHook` / `WithPostNodeHook` 支持多次调用链式累加(不覆盖已有钩子),无需修改 `NodeFn`,极易解耦日志、链路追踪(OpenTelemetry)或流式输出等逻辑。
- **开箱即用持久化:** 提供 `Checkpointer[S]` 接口与内置的 `FileCheckpointer[S]`,支持将进度保存为 JSON 文件,实现停机或重启后进度无损恢复。 - **编译期结构校验:** 图在编译 (`Compile`) 时,不仅会验证所有节点和边是否存在,还会严格检测同一节点是否冲突地定义了多种出边类型(如同时定义静态边、条件边与并发边),规避多义性逻辑隐患。
- **开箱即用持久化与安全校验:** 提供 `Checkpointer[S]` 接口与内置的 `FileCheckpointer[S]`,采用“先写临时文件再原子重命名”防损坏机制,并对 `threadID` 进行路径穿越防御性校验,实现进程意外退出后进度安全无损地恢复。
- **纯粹的高性能:** 基于内存流转,零外部依赖,极低的上下文切换开销。 - **纯粹的高性能:** 基于内存流转,零外部依赖,极低的上下文切换开销。
### 快速开始 ### 快速开始
@@ -46,7 +47,7 @@ package main
import ( import (
"context" "context"
GopherGraph "github.com/unclesam-LY/GopherGraph" GopherGraph "github.com/unclesam-ly/GopherGraph"
) )
func main() { func main() {
@@ -134,6 +135,11 @@ fc.Save(context.Background(), sessionID, thread)
// 重启程序后,可以从磁盘重新加载进度并恢复执行 // 重启程序后,可以从磁盘重新加载进度并恢复执行
loadedThread, _ := fc.Load(context.Background(), sessionID) loadedThread, _ := fc.Load(context.Background(), sessionID)
thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState) thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState)
// 任务彻底跑完后,清理磁盘上的存档文件
if thread.IsFinished {
fc.Delete(context.Background(), sessionID)
}
``` ```
#### 运行内置示例 #### 运行内置示例
@@ -163,10 +169,11 @@ go test -v -race ./...
- **Support for Cycles/Loops:** Unlike traditional DAG (Directed Acyclic Graph) engines, GopherGraph natively supports loops and dynamic routing based on agent evaluation. - **Support for Cycles/Loops:** Unlike traditional DAG (Directed Acyclic Graph) engines, GopherGraph natively supports loops and dynamic routing based on agent evaluation.
- **Parallel Branching with Short-Circuit Cancellation:** Run multiple agent nodes concurrently. Built on `context.WithCancel`, a failure in any branch immediately cancels all sibling goroutines, preventing wasted compute on LLM API calls. - **Parallel Branching with Short-Circuit Cancellation:** Run multiple agent nodes concurrently. Built on `context.WithCancel`, a failure in any branch immediately cancels all sibling goroutines, preventing wasted compute on LLM API calls.
- **Deep-Clone for Concurrency Safety:** Inject a custom `StateCloner` via `Engine.WithStateCloner` to safely deep-copy reference types (slices, maps, pointers) before fan-out, eliminating data races at the root. - **Deep-Clone for Concurrency Safety:** Inject a custom `StateCloner` via `Engine.WithStateCloner` to safely deep-copy reference types (slices, maps, pointers) before fan-out, eliminating data races at the root.
- **Human-in-the-Loop (HITL):** Pause workflow execution *before* a designated node, capture a snapshot (`Thread`), modify the state, and `Resume` seamlessly. - **Human-in-the-Loop (HITL):** Pause workflow execution *before* a designated node, capture a snapshot (`Thread`), modify the state, and `Resume` seamlessly. Returns exported sentinel errors `ErrNotPaused` or `ErrAlreadyFinished` when resume pre-conditions are not met.
- **Infinite Loop Circuit Breaker:** Set a hard step limit with `Engine.WithMaxSteps`. Any accidental cycle in the graph returns a clear error instead of blocking forever. - **Infinite Loop Circuit Breaker:** Set a hard step limit with `Engine.WithMaxSteps`. Any accidental cycle in the graph returns a clear error instead of blocking forever.
- **Lightweight Lifecycle Hooks:** `Engine.WithPreNodeHook` / `WithPostNodeHook` fire before and after each node without touching any `NodeFn` signature — perfect for logging, OpenTelemetry tracing, or streaming output. - **Lightweight Lifecycle Hooks (Additive Chaining):** `Engine.WithPreNodeHook` / `WithPostNodeHook` fire before and after each node without touching any `NodeFn` signature. Supports multiple chained registrations (additive) instead of overwriting, making it easy to decouple concerns like logging, tracing (OpenTelemetry), and streaming.
- **State Persistence (Checkpointer):** Generic `Checkpointer[S]` interface with built-in `FileCheckpointer[S]` for saving and loading execution snapshots to/from JSON files. - **Compile-Time Structural Validation:** During `Compile()`, the engine validates node/edge existence and strictly checks for outgoing edge conflicts on each node (e.g., preventing a node from simultaneously defining static, conditional, and parallel edges) to ensure unambiguous flow execution.
- **State Persistence & Path Safety:** Generic `Checkpointer[S]` interface with built-in `FileCheckpointer[S]`. Uses an atomic write (temp-file write followed by rename) to prevent corruption, and includes path traversal validation on `threadID`.
- **Zero-Dependency & High-Performance:** Written in pure Go with zero external dependencies, leveraging in-memory queues for lightning-fast orchestration. - **Zero-Dependency & High-Performance:** Written in pure Go with zero external dependencies, leveraging in-memory queues for lightning-fast orchestration.
### Quick Start ### Quick Start
@@ -187,7 +194,7 @@ package main
import ( import (
"context" "context"
GopherGraph "github.com/unclesam-LY/GopherGraph" GopherGraph "github.com/unclesam-ly/GopherGraph"
) )
func main() { func main() {
@@ -274,6 +281,11 @@ fc.Save(context.Background(), sessionID, thread)
// Reload the snapshot (e.g. after a process restart) and resume // Reload the snapshot (e.g. after a process restart) and resume
loadedThread, _ := fc.Load(context.Background(), sessionID) loadedThread, _ := fc.Load(context.Background(), sessionID)
thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState) thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState)
// Clean up the storage file once the workflow is completely finished
if thread.IsFinished {
fc.Delete(context.Background(), sessionID)
}
``` ```
#### Run the Interactive Demo #### Run the Interactive Demo
@@ -303,10 +315,11 @@ go test -v -race ./...
- **ループと条件付き分岐のサポート:** 従来の DAG(有向非巡回グラフ)の制限を超え、エージェントの判定に基づく条件付きルートや双方向のループをサポート。 - **ループと条件付き分岐のサポート:** 従来の DAG(有向非巡回グラフ)の制限を超え、エージェントの判定に基づく条件付きルートや双方向のループをサポート。
- **並行処理とショートサーキットキャンセル:** `context.WithCancel` に基づく並行ブランチ実行を実装。いずれかのブランチでエラーが発生した瞬間に他の全 goroutine へキャンセルシグナルを伝播し、LLM API の無駄な呼び出しを防ぎます。 - **並行処理とショートサーキットキャンセル:** `context.WithCancel` に基づく並行ブランチ実行を実装。いずれかのブランチでエラーが発生した瞬間に他の全 goroutine へキャンセルシグナルを伝播し、LLM API の無駄な呼び出しを防ぎます。
- **データ競合を防ぐ深いコピー(DeepClone):** `Engine.WithStateCloner` でカスタムの深コピー関数を注入。参照型(スライス・マップ・ポインタ)を持つ状態でも、並行ブランチでのデータ競合を根本から解消します。 - **データ競合を防ぐ深いコピー(DeepClone):** `Engine.WithStateCloner` でカスタムの深コピー関数を注入。参照型(スライス・マップ・ポインタ)を持つ状態でも、並行ブランチでのデータ競合を根本から解消します。
- **Human-in-the-Loop (HITL) の一時停止と再開:** 特定のノードの実行前に処理を自動停止し、スナップショット(`Thread`)を返します。承認や状態修正の後にシームレスに処理を再開(`Resume`)できます。 - **Human-in-the-Loop (HITL) の一時停止と再開:** 特定のノードの実行前に処理を自动停止し、スナップショット(`Thread`)を返します。承認や状態修正の後にシームレスに処理を再開(`Resume`)できます。再開判定用に `ErrNotPaused` や `ErrAlreadyFinished` といった明確なセンチネルエラーを提供しています。
- **無限ループのサーキットブレーカー:** `Engine.WithMaxSteps` でステップ数の上限を設定。グラフ内に意図しないサイクルが発生しても、永久ブロックせず明確なエラーを返します。 - **無限ループのサーキットブレーカー:** `Engine.WithMaxSteps` でステップ数の上限を設定。グラフ内に意図しないサイクルが発生しても、永久ブロックせず明確なエラーを返します。
- **軽量なライフサイクルフック:** `Engine.WithPreNodeHook` / `WithPostNodeHook` により、`NodeFn` のシグネチャを一切変更せずにロギング・トレーシング・ストリーミング出力を実装できます。 - **軽量なライフサイクルフック(複数チェーン登録対応):** `Engine.WithPreNodeHook` / `WithPostNodeHook` により、`NodeFn` のシグネチャを変更することなくロギング、トレーシング(OpenTelemetry)、ストリーミング出力を実装できます。**複数回呼び出すことで複数のフックを順番に実行(チェーン蓄積)**可能です。
- **状態の永続化 (Checkpointer):** 抽象的な `Checkpointer[S]` インタフェースと、JSONファイルへの書き出し・読み込みを行う組み込みの `FileCheckpointer[S]` をサポート。サーバー再起動後の進捗復旧が可能。 - **コンパイル時の構造検証:** `Compile()` 実行時に、すべてのノードや辺の存在をチェックするだけでなく、同一ノードに対する出辺タイプ(静的辺、条件付き辺、並行辺)の重複競合も厳密に検出し、意図しない曖昧な挙動を防ぎます。
- **状態の永続化とパス安全性:** 抽象的な `Checkpointer[S]` インタフェースと、書き込み時の破損を防ぐアトミック書き込み(一時ファイルからリネーム)に対応した `FileCheckpointer[S]` を内蔵。`threadID` のパス・トラバーサル検証によりセキュアに状態の復旧が可能です。
- **ピュア Go & 高性能:** 外部依存関係ゼロ。メモリ内チャネルを用いた高速なコンテキスト切り替え。 - **ピュア Go & 高性能:** 外部依存関係ゼロ。メモリ内チャネルを用いた高速なコンテキスト切り替え。
### クイックスタート ### クイックスタート
@@ -327,7 +340,7 @@ package main
import ( import (
"context" "context"
GopherGraph "github.com/unclesam-LY/GopherGraph" GopherGraph "github.com/unclesam-ly/GopherGraph"
) )
func main() { func main() {
@@ -414,6 +427,11 @@ fc.Save(context.Background(), sessionID, thread)
// スナップショットを読み込み、プロセス再起動後に処理を再開 // スナップショットを読み込み、プロセス再起動後に処理を再開
loadedThread, _ := fc.Load(context.Background(), sessionID) loadedThread, _ := fc.Load(context.Background(), sessionID)
thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState) thread, _ = cg.Resume(context.Background(), loadedThread, modifiedState)
// 処理が完全に終了した後、ストレージ上のファイルをクリーンアップする
if thread.IsFinished {
fc.Delete(context.Background(), sessionID)
}
``` ```
#### インタラクティブデモの実行 #### インタラクティブデモの実行
+95 -7
View File
@@ -6,6 +6,7 @@ import (
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"strings"
) )
// Checkpointer 定义了持久化状态的接口,S 是工作流的强类型状态。 // Checkpointer 定义了持久化状态的接口,S 是工作流的强类型状态。
@@ -14,38 +15,93 @@ type Checkpointer[S any] interface {
Save(ctx context.Context, threadID string, thread *Thread[S]) error Save(ctx context.Context, threadID string, thread *Thread[S]) error
// Load 读取之前保存的线程快照,用于恢复执行。 // Load 读取之前保存的线程快照,用于恢复执行。
Load(ctx context.Context, threadID string) (*Thread[S], error) Load(ctx context.Context, threadID string) (*Thread[S], error)
// Delete 删除指定线程的快照文件,用于工作流完结后的清理。
Delete(ctx context.Context, threadID string) error
} }
// FileCheckpointer 实现了基于本地 JSON 文件的状态持久化。 // FileCheckpointer 实现了基于本地 JSON 文件的状态持久化。
type FileCheckpointer[S any] struct { type FileCheckpointer[S any] struct {
dir string // 存放状态文件的文件夹路径 dir string // 存放状态文件的文件夹路径
pretty bool // 是否使用缩进格式(方便人工阅读,但体积更大)
}
// FileCheckpointerOption 是 FileCheckpointer 的可选配置项。
type FileCheckpointerOption func(*fileCheckpointerConfig)
type fileCheckpointerConfig struct {
pretty bool
}
// WithPrettyJSON 开启 JSON 缩进格式(方便人工阅读,适合开发调试阶段)。
// 默认关闭,使用紧凑格式以减少磁盘 I/O 和文件体积。
func WithPrettyJSON() FileCheckpointerOption {
return func(c *fileCheckpointerConfig) {
c.pretty = true
}
} }
// NewFileCheckpointer 创建一个文件检查点管理器。 // NewFileCheckpointer 创建一个文件检查点管理器。
func NewFileCheckpointer[S any](dir string) (*FileCheckpointer[S], error) { // 支持通过 opts 传入可选配置(如 WithPrettyJSON())。
func NewFileCheckpointer[S any](dir string, opts ...FileCheckpointerOption) (*FileCheckpointer[S], error) {
// 确保文件夹存在,不存在则创建 // 确保文件夹存在,不存在则创建
if err := os.MkdirAll(dir, 0755); err != nil { if err := os.MkdirAll(dir, 0755); err != nil {
return nil, fmt.Errorf("failed to create checkpoint dir: %w", err) return nil, fmt.Errorf("failed to create checkpoint dir: %w", err)
} }
return &FileCheckpointer[S]{dir: dir}, nil cfg := &fileCheckpointerConfig{}
for _, opt := range opts {
opt(cfg)
}
return &FileCheckpointer[S]{dir: dir, pretty: cfg.pretty}, nil
} }
// Save 将 Thread 序列化为 JSON 并写入本地文件 // Save 将 Thread 序列化为 JSON 并写入本地文件。
// 采用"先写临时文件再原子重命名"的方式,防止写入中途程序崩溃导致文件损坏。
// 若存在上次崩溃残留的临时文件,会在写入前自动清理。
func (fc *FileCheckpointer[S]) Save(ctx context.Context, threadID string, thread *Thread[S]) error { func (fc *FileCheckpointer[S]) Save(ctx context.Context, threadID string, thread *Thread[S]) error {
path := filepath.Join(fc.dir, threadID+".json") if err := validateThreadID(threadID); err != nil {
return err
}
// 使用 MarshalIndent 方便人工阅读生成的 JSON 文件 path := filepath.Join(fc.dir, threadID+".json")
data, err := json.MarshalIndent(thread, "", " ") tmpPath := path + ".tmp"
// 序列化:生产环境默认使用紧凑格式以减少磁盘 I/O;开发阶段可用 WithPrettyJSON() 开启缩进
var data []byte
var err error
if fc.pretty {
data, err = json.MarshalIndent(thread, "", " ")
} else {
data, err = json.Marshal(thread)
}
if err != nil { if err != nil {
return fmt.Errorf("failed to marshal thread: %w", err) return fmt.Errorf("failed to marshal thread: %w", err)
} }
return os.WriteFile(path, data, 0644) // 清理上次崩溃可能残留的临时文件(忽略不存在的错误)
_ = os.Remove(tmpPath)
// 原子写入:先写临时文件,再 Rename 替换,确保断电/崩溃安全
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
return fmt.Errorf("failed to write temp file: %w", err)
}
if err := os.Rename(tmpPath, path); err != nil {
// Rename 失败时,尝试清理临时文件,避免残留
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to rename temp file: %w", err)
}
return nil
} }
// Load 从本地文件中读取 JSON 并反序列化为 Thread。 // Load 从本地文件中读取 JSON 并反序列化为 Thread。
// 若加载到一个已完结(IsFinished == true)的线程,会包装 ErrAlreadyFinished 作为额外信息返回,
// 方便调用方做幂等处理(注意:此时 thread 依然有效,err 可通过 errors.Is 判断)。
func (fc *FileCheckpointer[S]) Load(ctx context.Context, threadID string) (*Thread[S], error) { func (fc *FileCheckpointer[S]) Load(ctx context.Context, threadID string) (*Thread[S], error) {
if err := validateThreadID(threadID); err != nil {
return nil, err
}
path := filepath.Join(fc.dir, threadID+".json") path := filepath.Join(fc.dir, threadID+".json")
data, err := os.ReadFile(path) data, err := os.ReadFile(path)
@@ -58,5 +114,37 @@ func (fc *FileCheckpointer[S]) Load(ctx context.Context, threadID string) (*Thre
return nil, fmt.Errorf("failed to unmarshal thread: %w", err) return nil, fmt.Errorf("failed to unmarshal thread: %w", err)
} }
// 如果加载到已完结的线程,附带哨兵错误提示,方便调用方感知
if thread.IsFinished {
return &thread, fmt.Errorf("thread %q has already finished: %w", threadID, ErrAlreadyFinished)
}
return &thread, nil return &thread, nil
} }
// Delete 删除指定 threadID 对应的快照文件。
// 通常在工作流完结后调用,防止快照文件无限积累占用磁盘空间。
// 若文件不存在,静默返回 nil(幂等操作)。
func (fc *FileCheckpointer[S]) Delete(ctx context.Context, threadID string) error {
if err := validateThreadID(threadID); err != nil {
return err
}
path := filepath.Join(fc.dir, threadID+".json")
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to delete checkpoint file: %w", err)
}
return nil
}
// validateThreadID 校验 threadID 的合法性,防止路径穿越(path traversal)攻击。
// threadID 中不允许出现目录分隔符或 ".." 序列。
func validateThreadID(id string) error {
if id == "" {
return fmt.Errorf("threadID must not be empty")
}
if strings.ContainsAny(id, "/\\") || strings.Contains(id, "..") {
return fmt.Errorf("threadID %q contains invalid characters", id)
}
return nil
}
+19
View File
@@ -0,0 +1,19 @@
// Package GopherGraph 是一个用 Go 语言编写的代码即图(Code-as-Graph)智能体编排引擎,
// 设计灵感来源于 Python 生态的 LangGraph。
//
// 它利用 Go 1.18+ 的泛型机制,让工作流的上下文状态在编译期就具备强类型约束,
// 并依靠 Go 原生的 Goroutines 和 Channels 实现极高并发的本地数据流转。
//
// # 核心概念
//
// - [Graph] 图构建器:通过 [NewGraph] 创建,使用 AddNode/AddEdge 系列方法声明节点和连线。
// - [CompiledGraph] 编译后的只读图:通过 [Graph.Compile] 生成,可直接调用 Start/Resume 运行。
// - [Engine] 增强包装器:通过 [NewEngine] 创建,在 CompiledGraph 基础上叠加
// StateCloner(深拷贝)、MaxSteps(死循环熔断)、PreNodeHook/PostNodeHook(生命周期钩子)。
// - [Thread] 执行快照:承载当前状态、下一节点、暂停/结束标记,可序列化用于断点恢复。
// - [Checkpointer] 持久化接口:提供 Save/Load/Delete 用于工作流进度存储和清理,内置 [FileCheckpointer] 实现基于 JSON 文件的状态存储。
//
// # 零外部依赖
//
// GopherGraph 仅使用 Go 标准库,无任何第三方依赖。
package GopherGraph
+99 -32
View File
@@ -27,6 +27,11 @@ type CompiledGraph[S any] struct {
// Compile 校验图结构的合法性,并生成可运行的 CompiledGraph。 // Compile 校验图结构的合法性,并生成可运行的 CompiledGraph。
func (g *Graph[S]) Compile() (*CompiledGraph[S], error) { func (g *Graph[S]) Compile() (*CompiledGraph[S], error) {
// 首先展示所有注册阶段积累的错误(如重复注册节点或边)
if err := errorsJoin(g.errs); err != nil {
return nil, fmt.Errorf("graph registration errors:\n%w", err)
}
if len(g.nodes) == 0 { if len(g.nodes) == 0 {
return nil, errors.New("cannot compile graph: graph contains no nodes") return nil, errors.New("cannot compile graph: graph contains no nodes")
} }
@@ -37,7 +42,7 @@ func (g *Graph[S]) Compile() (*CompiledGraph[S], error) {
return nil, fmt.Errorf("compile error: edge origin %q does not exist", from) return nil, fmt.Errorf("compile error: edge origin %q does not exist", from)
} }
if _, exists := g.nodes[to]; !exists { if _, exists := g.nodes[to]; !exists {
return nil, fmt.Errorf("compile error: edge origin %q does not exist", to) return nil, fmt.Errorf("compile error: edge destination %q does not exist", to)
} }
} }
@@ -79,16 +84,72 @@ func (g *Graph[S]) Compile() (*CompiledGraph[S], error) {
} }
} }
// 返回编译好的只读图 // 校验同一节点的出边冲突:不能同时存在于 edges、conditional、parallels 中
for node := range g.nodes {
conflictCount := 0
if _, ok := g.edges[node]; ok {
conflictCount++
}
if _, ok := g.conditional[node]; ok {
conflictCount++
}
if _, ok := g.parallels[node]; ok {
conflictCount++
}
if conflictCount > 1 {
return nil, fmt.Errorf("compile error: node %q has conflicting outgoing edges (defined in multiple edge types simultaneously)", node)
}
}
// 返回编译好的只读图(防御性拷贝,阻止编译后修改 Graph 对 CompiledGraph 造成污染)
nodesCopy := make(map[string]NodeFn[S], len(g.nodes))
for k, v := range g.nodes {
nodesCopy[k] = v
}
edgesCopy := make(map[string]string, len(g.edges))
for k, v := range g.edges {
edgesCopy[k] = v
}
conditionalCopy := make(map[string]RouterFn[S], len(g.conditional))
for k, v := range g.conditional {
conditionalCopy[k] = v
}
interruptsCopy := make(map[string]bool, len(g.interrupts))
for k, v := range g.interrupts {
interruptsCopy[k] = v
}
parallelsCopy := make(map[string]parallelStep[S], len(g.parallels))
for k, v := range g.parallels {
parallelsCopy[k] = v
}
return &CompiledGraph[S]{ return &CompiledGraph[S]{
nodes: g.nodes, nodes: nodesCopy,
edges: g.edges, edges: edgesCopy,
conditional: g.conditional, conditional: conditionalCopy,
interrupts: g.interrupts, interrupts: interruptsCopy,
parallels: g.parallels, parallels: parallelsCopy,
}, nil }, nil
} }
// ErrNotPaused 在调用 Resume 时线程未处于暂停状态时返回。可用于 errors.Is 判断。
var ErrNotPaused = errors.New("cannot resume: thread is not paused")
// ErrAlreadyFinished 在调用 Resume 时线程已结束时返回。可用于 errors.Is 判断。
var ErrAlreadyFinished = errors.New("cannot resume: thread is already finished")
// validateResume 验证线程是否满足恢复执行的前提条件。
// CompiledGraph.Resume 和 Engine.Resume 共用此逻辑,避免重复实现。
func validateResume[S any](thread *Thread[S]) error {
if !thread.IsPaused {
return ErrNotPaused
}
if thread.IsFinished {
return ErrAlreadyFinished
}
return nil
}
// Start 从指定的起始节点开始运行图,并持续流转,直到遇到中断或运行结束 // Start 从指定的起始节点开始运行图,并持续流转,直到遇到中断或运行结束
func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initialState S) (*Thread[S], error) { func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initialState S) (*Thread[S], error) {
thread := &Thread[S]{ thread := &Thread[S]{
@@ -101,11 +162,8 @@ func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initial
// Resume 恢复执行一个被暂停的线程,并允许注入外部修改后的状态数据(例如人工审批修改后的结果) // Resume 恢复执行一个被暂停的线程,并允许注入外部修改后的状态数据(例如人工审批修改后的结果)
func (cg *CompiledGraph[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) { func (cg *CompiledGraph[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) {
if !thread.IsPaused { if err := validateResume(thread); err != nil {
return nil, errors.New("cannot resume: thread is not paused") return nil, err
}
if thread.IsFinished {
return nil, errors.New("cannot resume: thread is already finished")
} }
// 注入人工修改后的状态,并解除暂停标记 // 注入人工修改后的状态,并解除暂停标记
@@ -128,17 +186,6 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run
stepCount := 0 stepCount := 0
for { for {
// 【熔断 1】检查步数上限,防御死循环(infinite loop protection)
if opts.maxSteps > 0 {
if stepCount >= opts.maxSteps {
return thread, fmt.Errorf(
"engine halt: max steps (%d) exceeded, possible infinite loop in graph",
opts.maxSteps,
)
}
stepCount++
}
// 【熔断 2】检查 Context,支持外部超时控制或取消 // 【熔断 2】检查 Context,支持外部超时控制或取消
select { select {
case <-ctx.Done(): case <-ctx.Done():
@@ -153,6 +200,17 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run
return thread, nil return thread, nil
} }
// 【熔断 1】检查步数上限,防御死循环(infinite loop protection)
if opts.maxSteps > 0 {
if stepCount >= opts.maxSteps {
return thread, fmt.Errorf(
"engine halt: max steps (%d) exceeded, possible infinite loop in graph",
opts.maxSteps,
)
}
stepCount++
}
// 查找当前节点 // 查找当前节点
nodeFn, exists := cg.nodes[currentNodeName] nodeFn, exists := cg.nodes[currentNodeName]
if !exists { if !exists {
@@ -198,6 +256,12 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run
if err != nil { if err != nil {
return thread, fmt.Errorf("router for node %q execution error: %w", currentNodeName, err) return thread, fmt.Errorf("router for node %q execution error: %w", currentNodeName, err)
} }
// 立即校验返回的节点是否存在,提供更清晰的报错
if next != "" {
if _, exists := cg.nodes[next]; !exists {
return thread, fmt.Errorf("router for node %q returned non-existent node %q (not found)", currentNodeName, next)
}
}
nextNode = next nextNode = next
} else { } else {
// 否则使用静态连线边 // 否则使用静态连线边
@@ -207,11 +271,6 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run
// 更新快照中的"下一个节点" // 更新快照中的"下一个节点"
thread.NextNode = nextNode thread.NextNode = nextNode
// 如果下一站是终点,直接进入下一次循环触发结束逻辑
if nextNode == "" {
continue
}
// 核心中断机制:如果即将进入的下一个节点被标记为了中断节点,则在此挂起 // 核心中断机制:如果即将进入的下一个节点被标记为了中断节点,则在此挂起
if cg.interrupts[nextNode] { if cg.interrupts[nextNode] {
thread.IsPaused = true thread.IsPaused = true
@@ -283,18 +342,26 @@ func (cg *CompiledGraph[S]) runParallelBranches(
wg.Wait() wg.Wait()
close(resultCh) close(resultCh)
// 按索引顺序收集结果,保证合并函数收到的 branches 顺序与 targets 一致 // 【P0 修复】先收集所有结果,再判断是否有真实错误。
// 原先在循环中遇到非 context.Canceled 错误就立即 return,但由于 channel 中的写入顺序是不确定的,
// 存在以下竞态:分支 A 先把 context.Canceled 写入 channel,紧接着分支 B 才把真实错误写入。
// 若此时早返回,真实错误永远不会被读取,调用方可能收到 nil error 而实际上有分支失败了。
// 解决方案:等全部结果到齐后,再过滤出第一个非取消的真实错误。
branches := make([]S, len(targets)) branches := make([]S, len(targets))
var firstErr error
for r := range resultCh { for r := range resultCh {
if r.err != nil { if r.err != nil {
// 过滤掉因取消产生的次生错误,只返回第一个真实错误 // 过滤掉因取消产生的次生错误,只记录第一个真实错误
if !errors.Is(r.err, context.Canceled) { if !errors.Is(r.err, context.Canceled) && firstErr == nil {
return nil, r.err firstErr = r.err
} }
} else { } else {
branches[r.idx] = r.state branches[r.idx] = r.state
} }
} }
if firstErr != nil {
return nil, firstErr
}
return branches, nil return branches, nil
} }
+44 -17
View File
@@ -1,6 +1,9 @@
package GopherGraph package GopherGraph
import "context" import (
"context"
"sync"
)
// Engine 是 CompiledGraph 的增强包装器,在完全兼容原有 API 的基础上, // Engine 是 CompiledGraph 的增强包装器,在完全兼容原有 API 的基础上,
// 提供以下四项生产级能力: // 提供以下四项生产级能力:
@@ -12,6 +15,12 @@ import "context"
// //
// 所有选项均为可选,通过链式调用注册。未注册的选项将退回到 CompiledGraph 的默认行为。 // 所有选项均为可选,通过链式调用注册。未注册的选项将退回到 CompiledGraph 的默认行为。
// //
// # 线程安全说明
//
// Engine 使用 sync.RWMutex 保护配置字段,WithXxx 配置方法和 opts() 读取方法均加锁。
// 因此,允许在 Start/Resume 运行期间并发调用 WithXxx 修改配置(但不推荐这样做)。
// 常见的安全模式:在服务启动时一次性完成链式配置,之后多个 Goroutine 并发调用 Start/Resume。
//
// 用法示例: // 用法示例:
// //
// engine := GopherGraph.NewEngine(compiledGraph). // engine := GopherGraph.NewEngine(compiledGraph).
@@ -30,6 +39,7 @@ import "context"
// //
// thread, err := engine.Start(ctx, "start", initialState) // thread, err := engine.Start(ctx, "start", initialState)
type Engine[S any] struct { type Engine[S any] struct {
mu sync.RWMutex
graph *CompiledGraph[S] graph *CompiledGraph[S]
stateCloner func(S) S stateCloner func(S) S
maxSteps int maxSteps int
@@ -50,6 +60,8 @@ func NewEngine[S any](graph *CompiledGraph[S]) *Engine[S] {
// 若不注册,引擎退化为 Go 默认的值拷贝语义(浅拷贝)。 // 若不注册,引擎退化为 Go 默认的值拷贝语义(浅拷贝)。
// 对于状态中含有引用类型的场景,强烈建议注册此函数。 // 对于状态中含有引用类型的场景,强烈建议注册此函数。
func (e *Engine[S]) WithStateCloner(fn func(S) S) *Engine[S] { func (e *Engine[S]) WithStateCloner(fn func(S) S) *Engine[S] {
e.mu.Lock()
defer e.mu.Unlock()
e.stateCloner = fn e.stateCloner = fn
return e return e
} }
@@ -62,6 +74,8 @@ func (e *Engine[S]) WithStateCloner(fn func(S) S) *Engine[S] {
// 建议根据工作流的预期最大深度合理设置(如 100~1000)。 // 建议根据工作流的预期最大深度合理设置(如 100~1000)。
// 传入 0 或负数表示不限制(与不调用此方法等效)。 // 传入 0 或负数表示不限制(与不调用此方法等效)。
func (e *Engine[S]) WithMaxSteps(n int) *Engine[S] { func (e *Engine[S]) WithMaxSteps(n int) *Engine[S] {
e.mu.Lock()
defer e.mu.Unlock()
e.maxSteps = n e.maxSteps = n
return e return e
} }
@@ -73,10 +87,21 @@ func (e *Engine[S]) WithMaxSteps(n int) *Engine[S] {
// - 记录 OpenTelemetry Span 的开始时间 // - 记录 OpenTelemetry Span 的开始时间
// - 向前端推送"正在执行 xxx 节点"的流式事件 // - 向前端推送"正在执行 xxx 节点"的流式事件
// //
// 支持多次调用:每次调用将新钩子追加到执行链末尾,按注册顺序依次执行,不覆盖已有钩子。
// 钩子函数接收到的 state 是节点执行【前】的状态快照,请勿修改(只读语义)。 // 钩子函数接收到的 state 是节点执行【前】的状态快照,请勿修改(只读语义)。
// 钩子中的 panic 会向上传播,建议在内部做好 recover。 // 钩子中的 panic 会向上传播,建议在内部做好 recover。
func (e *Engine[S]) WithPreNodeHook(fn HookFn[S]) *Engine[S] { func (e *Engine[S]) WithPreNodeHook(fn HookFn[S]) *Engine[S] {
e.mu.Lock()
defer e.mu.Unlock()
if e.preNodeHook == nil {
e.preNodeHook = fn e.preNodeHook = fn
} else {
prev := e.preNodeHook
e.preNodeHook = func(ctx context.Context, name string, s S) {
prev(ctx, name, s)
fn(ctx, name, s)
}
}
return e return e
} }
@@ -87,14 +112,28 @@ func (e *Engine[S]) WithPreNodeHook(fn HookFn[S]) *Engine[S] {
// - 上报节点耗时指标(Prometheus/DataDog) // - 上报节点耗时指标(Prometheus/DataDog)
// - 向前端推送节点执行完毕的流式事件 // - 向前端推送节点执行完毕的流式事件
// //
// 支持多次调用:每次调用将新钩子追加到执行链末尾,按注册顺序依次执行,不覆盖已有钩子。
// 钩子函数接收到的 state 是节点执行【后】更新过的状态,同为只读语义。 // 钩子函数接收到的 state 是节点执行【后】更新过的状态,同为只读语义。
func (e *Engine[S]) WithPostNodeHook(fn HookFn[S]) *Engine[S] { func (e *Engine[S]) WithPostNodeHook(fn HookFn[S]) *Engine[S] {
e.mu.Lock()
defer e.mu.Unlock()
if e.postNodeHook == nil {
e.postNodeHook = fn e.postNodeHook = fn
} else {
prev := e.postNodeHook
e.postNodeHook = func(ctx context.Context, name string, s S) {
prev(ctx, name, s)
fn(ctx, name, s)
}
}
return e return e
} }
// opts 将 Engine 的配置打包成 runOptions,传入底层 run() 调度器。 // opts 将 Engine 的配置快照打包成 runOptions,传入底层 run() 调度器。
// 使用 RLock 保证读取配置时的并发安全。
func (e *Engine[S]) opts() runOptions[S] { func (e *Engine[S]) opts() runOptions[S] {
e.mu.RLock()
defer e.mu.RUnlock()
return runOptions[S]{ return runOptions[S]{
maxSteps: e.maxSteps, maxSteps: e.maxSteps,
stateCloner: e.stateCloner, stateCloner: e.stateCloner,
@@ -115,12 +154,10 @@ func (e *Engine[S]) Start(ctx context.Context, startNode string, initialState S)
// Resume 恢复执行一个被暂停的线程,行为与 CompiledGraph.Resume 完全一致, // Resume 恢复执行一个被暂停的线程,行为与 CompiledGraph.Resume 完全一致,
// 但会附加所有已注册的 Engine 增强选项(Hooks、MaxSteps、StateCloner)。 // 但会附加所有已注册的 Engine 增强选项(Hooks、MaxSteps、StateCloner)。
// 复用 validateResume 进行前提条件校验,与 CompiledGraph.Resume 保持一致。
func (e *Engine[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) { func (e *Engine[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) {
if !thread.IsPaused { if err := validateResume(thread); err != nil {
return nil, errNotPaused return nil, err
}
if thread.IsFinished {
return nil, errAlreadyFinished
} }
thread.State = modifiedState thread.State = modifiedState
@@ -128,13 +165,3 @@ func (e *Engine[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState
return e.graph.run(ctx, thread, e.opts()) return e.graph.run(ctx, thread, e.opts())
} }
// 复用与 CompiledGraph.Resume 中相同的哨兵错误,避免魔法字符串重复
var (
errNotPaused = errStr("cannot resume: thread is not paused")
errAlreadyFinished = errStr("cannot resume: thread is already finished")
)
type errStr string
func (e errStr) Error() string { return string(e) }
+11 -3
View File
@@ -4,9 +4,10 @@ import (
"context" "context"
"fmt" "fmt"
"log" "log"
"sync"
"time" "time"
GopherGraph "github.com/unclesam-LY/GopherGraph" GopherGraph "github.com/unclesam-ly/GopherGraph"
) )
// AgentState 是这个示例工作流的状态 // AgentState 是这个示例工作流的状态
@@ -82,7 +83,10 @@ func main() {
// ============================================================ // ============================================================
// 使用 Engine 包装器,挂载所有增强能力 // 使用 Engine 包装器,挂载所有增强能力
// ============================================================ // ============================================================
var nodeTimings = make(map[string]time.Time) var (
nodeTimings = make(map[string]time.Time)
nodeTimingsMu sync.Mutex
)
engine := GopherGraph.NewEngine(cg). engine := GopherGraph.NewEngine(cg).
// 1. 注册深拷贝函数,消除并发分支对 Messages 切片的数据竞争 // 1. 注册深拷贝函数,消除并发分支对 Messages 切片的数据竞争
@@ -91,13 +95,17 @@ func main() {
WithMaxSteps(50). WithMaxSteps(50).
// 3. Pre Hook:记录节点开始时间 + 向控制台推送"流式进度" // 3. Pre Hook:记录节点开始时间 + 向控制台推送"流式进度"
WithPreNodeHook(func(ctx context.Context, name string, s AgentState) { WithPreNodeHook(func(ctx context.Context, name string, s AgentState) {
nodeTimingsMu.Lock()
nodeTimings[name] = time.Now() nodeTimings[name] = time.Now()
nodeTimingsMu.Unlock()
fmt.Printf(" ⏳ [PRE ] %-15s | 消息数: %d\n", name, len(s.Messages)) fmt.Printf(" ⏳ [PRE ] %-15s | 消息数: %d\n", name, len(s.Messages))
}). }).
// 4. Post Hook:计算并打印节点耗时 // 4. Post Hook:计算并打印节点耗时
WithPostNodeHook(func(ctx context.Context, name string, s AgentState) { WithPostNodeHook(func(ctx context.Context, name string, s AgentState) {
nodeTimingsMu.Lock()
elapsed := time.Since(nodeTimings[name]) elapsed := time.Since(nodeTimings[name])
fmt.Printf(" ✅ [POST] %-15s | 耗时: %v | 消息数: %d\n", name, elapsed.Round(time.Millisecond), len(s.Messages)) nodeTimingsMu.Unlock()
fmt.Printf(" [POST] %-15s | 耗时: %v | 消息数: %d\n", name, elapsed.Round(time.Millisecond), len(s.Messages))
}) })
fmt.Println("🚀 启动工作流...") fmt.Println("🚀 启动工作流...")
+4 -2
View File
@@ -6,8 +6,9 @@ import (
"fmt" "fmt"
"os" "os"
"strings" "strings"
"time"
GopherGraph "github.com/unclesam-LY/GopherGraph" GopherGraph "github.com/unclesam-ly/GopherGraph"
) )
// TranslationState 代表我们工作流的共享状态(State) // TranslationState 代表我们工作流的共享状态(State)
@@ -122,7 +123,8 @@ func main() {
// 包含敏感词“秘密”,会触发人工审核中断: // 包含敏感词“秘密”,会触发人工审核中断:
inputText := "这是一个秘密你好" inputText := "这是一个秘密你好"
fmt.Printf("🚀 启动工作流,输入原文: %q\n", inputText) fmt.Printf("🚀 启动工作流,输入原文: %q\n", inputText)
ctx := context.Background() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
thread, err := cg.Start(ctx, "translator", TranslationState{InputText: inputText}) thread, err := cg.Start(ctx, "translator", TranslationState{InputText: inputText})
if err != nil { if err != nil {
fmt.Printf("运行失败: %v\n", err) fmt.Printf("运行失败: %v\n", err)
+41 -6
View File
@@ -1,6 +1,10 @@
package GopherGraph package GopherGraph
import "context" import (
"context"
"errors"
"fmt"
)
// NodeFn 代表图中的一个执行节点(Agent)。 // NodeFn 代表图中的一个执行节点(Agent)。
// 它接收当前的上下文和状态 S,处理后返回更新后的状态 S,或者返回错误。 // 它接收当前的上下文和状态 S,处理后返回更新后的状态 S,或者返回错误。
@@ -27,6 +31,7 @@ type Graph[S any] struct {
conditional map[string]RouterFn[S] // 条件路由边:源节点 -> 路由函数 conditional map[string]RouterFn[S] // 条件路由边:源节点 -> 路由函数
interrupts map[string]bool // 标记需要中断的节点:执行完该节点后挂起 interrupts map[string]bool // 标记需要中断的节点:执行完该节点后挂起
parallels map[string]parallelStep[S] // 并发分支:源节点 -> 并发步骤配置 parallels map[string]parallelStep[S] // 并发分支:源节点 -> 并发步骤配置
errs []error // 注册阶段的错误积累(重复注册等),在 Compile 时统一报告
} }
// NewGraph 创建并初始化一个全新强类型的图构建器 // NewGraph 创建并初始化一个全新强类型的图构建器
@@ -36,27 +41,43 @@ func NewGraph[S any]() *Graph[S] {
edges: make(map[string]string), edges: make(map[string]string),
conditional: make(map[string]RouterFn[S]), conditional: make(map[string]RouterFn[S]),
interrupts: make(map[string]bool), interrupts: make(map[string]bool),
parallels: make(map[string]parallelStep[S]), // 初始化 parallels parallels: make(map[string]parallelStep[S]),
} }
} }
// AddNode 向图中注册一个节点(Agent) // AddNode 向图中注册一个节点(Agent)。
// 若同名节点已被注册,不会覆盖原有节点,错误将在 Compile() 时统一报告。
func (g *Graph[S]) AddNode(name string, fn NodeFn[S]) { func (g *Graph[S]) AddNode(name string, fn NodeFn[S]) {
if _, exists := g.nodes[name]; exists {
g.errs = append(g.errs, fmt.Errorf("AddNode: node %q is already registered", name))
return
}
g.nodes[name] = fn g.nodes[name] = fn
} }
// AddEdge 建立一条从 from 节点到 to 节点的静态连接线 // AddEdge 建立一条从 from 节点到 to 节点的静态连接线。
// 若 from 节点已有静态出边,不会覆盖,错误将在 Compile() 时统一报告。
func (g *Graph[S]) AddEdge(from, to string) { func (g *Graph[S]) AddEdge(from, to string) {
if _, exists := g.edges[from]; exists {
g.errs = append(g.errs, fmt.Errorf("AddEdge: node %q already has a static edge registered", from))
return
}
g.edges[from] = to g.edges[from] = to
} }
// AddConditionalEdges 建立一条从 from 节点出发的条件路由 // AddConditionalEdges 建立一条从 from 节点出发的条件路由
// 到底去哪,由 router 函数在运行时根据当前 State 动态决定 // 到底去哪,由 router 函数在运行时根据当前 State 动态决定。
// 若 from 节点已有条件出边,不会覆盖,错误将在 Compile() 时统一报告。
func (g *Graph[S]) AddConditionalEdges(from string, router RouterFn[S]) { func (g *Graph[S]) AddConditionalEdges(from string, router RouterFn[S]) {
if _, exists := g.conditional[from]; exists {
g.errs = append(g.errs, fmt.Errorf("AddConditionalEdges: node %q already has a conditional edge registered", from))
return
}
g.conditional[from] = router g.conditional[from] = router
} }
// AddInterrupt 标记在执行 nodeName 节点“之前”进行中断挂起,等待人工确认 // AddInterrupt 标记在执行 nodeName 节点"之前"进行中断挂起,等待人工确认。
// 重复标记同一节点为幂等操作(设置 true 再设置 true,无副作用)。
func (g *Graph[S]) AddInterrupt(nodeName string) { func (g *Graph[S]) AddInterrupt(nodeName string) {
g.interrupts[nodeName] = true g.interrupts[nodeName] = true
} }
@@ -66,7 +87,12 @@ func (g *Graph[S]) AddInterrupt(nodeName string) {
// targets: 需要并发执行的多个目标节点名称(如 "translate_en", "translate_fr") // targets: 需要并发执行的多个目标节点名称(如 "translate_en", "translate_fr")
// next: 所有并发节点运行完并合并后,下一步要去的节点(如 "publisher") // next: 所有并发节点运行完并合并后,下一步要去的节点(如 "publisher")
// merger: 用于合并各个并发分支状态的函数 // merger: 用于合并各个并发分支状态的函数
// 若 from 节点已有并发出边,不会覆盖,错误将在 Compile() 时统一报告。
func (g *Graph[S]) AddParallelEdges(from string, targets []string, next string, merger ParallelMergeFn[S]) { func (g *Graph[S]) AddParallelEdges(from string, targets []string, next string, merger ParallelMergeFn[S]) {
if _, exists := g.parallels[from]; exists {
g.errs = append(g.errs, fmt.Errorf("AddParallelEdges: node %q already has parallel edges registered", from))
return
}
g.parallels[from] = parallelStep[S]{ g.parallels[from] = parallelStep[S]{
targets: targets, targets: targets,
next: next, next: next,
@@ -78,3 +104,12 @@ func (g *Graph[S]) AddParallelEdges(from string, targets []string, next string,
// nodeName 为当前执行的节点名称,state 为该节点执行时的状态快照(只读语义,请勿修改)。 // nodeName 为当前执行的节点名称,state 为该节点执行时的状态快照(只读语义,请勿修改)。
// 可通过此钩子接入日志、链路追踪(OpenTelemetry)、监控指标等能力,而无需修改任何 NodeFn。 // 可通过此钩子接入日志、链路追踪(OpenTelemetry)、监控指标等能力,而无需修改任何 NodeFn。
type HookFn[S any] func(ctx context.Context, nodeName string, state S) type HookFn[S any] func(ctx context.Context, nodeName string, state S)
// errorsJoin 兼容多个错误合并为一个,避免 errors.Join 输出格式不够友好。
// 仅供内部 graph.errs 汇总使用。
func errorsJoin(errs []error) error {
if len(errs) == 0 {
return nil
}
return errors.Join(errs...)
}
+307 -1
View File
@@ -3,6 +3,7 @@ package GopherGraph
import ( import (
"context" "context"
"errors" "errors"
"os"
"strings" "strings"
"testing" "testing"
"time" "time"
@@ -256,7 +257,7 @@ func TestParallelExecution(t *testing.T) {
func TestFileCheckpointer(t *testing.T) { func TestFileCheckpointer(t *testing.T) {
// 创建测试临时文件夹(测试结束后 Go 会自动清理) // 创建测试临时文件夹(测试结束后 Go 会自动清理)
tmpDir := t.TempDir() tmpDir := t.TempDir()
fc, err := NewFileCheckpointer[TestState](tmpDir) fc, err := NewFileCheckpointer[TestState](tmpDir, WithPrettyJSON())
if err != nil { if err != nil {
t.Fatalf("创建文件检查点管理器失败: %v", err) t.Fatalf("创建文件检查点管理器失败: %v", err)
} }
@@ -316,6 +317,28 @@ func TestFileCheckpointer(t *testing.T) {
if !thread.IsFinished || thread.State.Value != 700 { if !thread.IsFinished || thread.State.Value != 700 {
t.Errorf("执行结果错误,最终值: %d, 是否结束: %t", thread.State.Value, thread.IsFinished) t.Errorf("执行结果错误,最终值: %d, 是否结束: %t", thread.State.Value, thread.IsFinished)
} }
// 5. 保存已完结的线程,Load 后应附带 ErrAlreadyFinished 提示
if err := fc.Save(ctx, sessionID, thread); err != nil {
t.Fatalf("保存已完结线程失败: %v", err)
}
_, loadErr := fc.Load(ctx, sessionID)
if !errors.Is(loadErr, ErrAlreadyFinished) {
t.Errorf("Load 已完结线程应附带 ErrAlreadyFinished,实际: %v", loadErr)
}
// 6. Delete 后文件应不存在
if err := fc.Delete(ctx, sessionID); err != nil {
t.Fatalf("删除检查点失败: %v", err)
}
_, err = fc.Load(ctx, sessionID)
if err == nil {
t.Error("删除后 Load 应报错(文件不存在)")
}
// 幂等:重复 Delete 不应报错
if err := fc.Delete(ctx, sessionID); err != nil {
t.Errorf("重复 Delete 应幂等不报错,实际: %v", err)
}
} }
// TestMaxStepsCircuitBreaker 验证 Engine.WithMaxSteps 能在步数超限时熔断,防止死循环 // TestMaxStepsCircuitBreaker 验证 Engine.WithMaxSteps 能在步数超限时熔断,防止死循环
@@ -452,3 +475,286 @@ func TestLifecycleHooks(t *testing.T) {
} }
} }
} }
// TestResumeErrors 验证 Resume 对非暂停/已结束线程的错误路径
func TestResumeErrors(t *testing.T) {
g := NewGraph[TestState]()
g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) {
return s, nil
})
cg, err := g.Compile()
if err != nil {
t.Fatalf("编译图失败: %v", err)
}
// 1. Resume 非暂停的线程
thread := &Thread[TestState]{NextNode: "A"}
_, err = cg.Resume(context.Background(), thread, TestState{})
if !errors.Is(err, ErrNotPaused) {
t.Errorf("期望 ErrNotPaused,实际: %v", err)
}
// 2. Resume 已结束的线程
thread2 := &Thread[TestState]{IsPaused: true, IsFinished: true}
_, err = cg.Resume(context.Background(), thread2, TestState{})
if !errors.Is(err, ErrAlreadyFinished) {
t.Errorf("期望 ErrAlreadyFinished,实际: %v", err)
}
// 3. Engine.Resume 也应返回相同的哨兵错误
engine := NewEngine(cg)
_, err = engine.Resume(context.Background(), &Thread[TestState]{NextNode: "A"}, TestState{})
if !errors.Is(err, ErrNotPaused) {
t.Errorf("Engine.Resume: 期望 ErrNotPaused,实际: %v", err)
}
}
// TestCompileValidationErrors 验证 Compile 对各种非法图结构的校验
func TestCompileValidationErrors(t *testing.T) {
// 1. 空图
g1 := NewGraph[TestState]()
_, err := g1.Compile()
if err == nil || !strings.Contains(err.Error(), "no nodes") {
t.Errorf("空图应报错: %v", err)
}
// 2. 边指向不存在的节点
g2 := NewGraph[TestState]()
g2.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
g2.AddEdge("A", "nonexistent")
_, err = g2.Compile()
if err == nil || !strings.Contains(err.Error(), "destination") {
t.Errorf("边指向不存在节点应报 destination 错误: %v", err)
}
// 3. 中断节点不存在
g3 := NewGraph[TestState]()
g3.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
g3.AddInterrupt("ghost")
_, err = g3.Compile()
if err == nil || !strings.Contains(err.Error(), "interrupt") {
t.Errorf("不存在的中断节点应报错: %v", err)
}
// 4. 重复注册节点和边
g4 := NewGraph[TestState]()
g4.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
g4.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil }) // 重复节点
g4.AddEdge("A", "B")
g4.AddEdge("A", "C") // 重复静态边
g4.AddConditionalEdges("A", func(ctx context.Context, s TestState) (string, error) { return "B", nil })
g4.AddConditionalEdges("A", func(ctx context.Context, s TestState) (string, error) { return "C", nil }) // 重复条件边
g4.AddParallelEdges("A", []string{"B"}, "", func(ctx context.Context, parent TestState, branches []TestState) (TestState, error) { return parent, nil })
g4.AddParallelEdges("A", []string{"C"}, "", func(ctx context.Context, parent TestState, branches []TestState) (TestState, error) { return parent, nil }) // 重复并发边
_, err = g4.Compile()
if err == nil {
t.Fatal("重复注册应报错")
}
errMsg := err.Error()
if !strings.Contains(errMsg, "already registered") {
t.Errorf("应包含节点重复错误: %v", err)
}
if !strings.Contains(errMsg, "static edge registered") {
t.Errorf("应包含静态边重复错误: %v", err)
}
if !strings.Contains(errMsg, "conditional edge registered") {
t.Errorf("应包含条件边重复错误: %v", err)
}
if !strings.Contains(errMsg, "parallel edges registered") {
t.Errorf("应包含并发边重复错误: %v", err)
}
}
// TestCompileEdgeConflict 验证同一节点不能同时存在于多种出边类型中
func TestCompileEdgeConflict(t *testing.T) {
g := NewGraph[TestState]()
g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
g.AddNode("B", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
// 同时为 A 设置静态边和条件路由
g.AddEdge("A", "B")
g.AddConditionalEdges("A", func(ctx context.Context, s TestState) (string, error) {
return "B", nil
})
_, err := g.Compile()
if err == nil || !strings.Contains(err.Error(), "conflicting") {
t.Errorf("出边冲突应报错: %v", err)
}
}
// TestRouterReturnsNonExistentNode 验证路由函数返回不存在的节点名时的运行时错误
func TestRouterReturnsNonExistentNode(t *testing.T) {
g := NewGraph[TestState]()
g.AddNode("start", func(ctx context.Context, s TestState) (TestState, error) { return s, nil })
g.AddConditionalEdges("start", func(ctx context.Context, s TestState) (string, error) {
return "nonexistent_node", nil
})
cg, err := g.Compile()
if err != nil {
t.Fatalf("编译图失败: %v", err)
}
_, err = cg.Start(context.Background(), "start", TestState{})
if err == nil || !strings.Contains(err.Error(), "not found") {
t.Errorf("路由到不存在节点应返回 runtime error: %v", err)
}
}
// TestParallelWithStateCloner 验证 StateCloner 在并发分支中正确隔离引用类型
func TestParallelWithStateCloner(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
})
// 两个分支都向 Log 切片追加元素
g.AddNode("branch1", func(ctx context.Context, s TestState) (TestState, error) {
s.Log = append(s.Log, "branch1")
s.Value = 10
return s, nil
})
g.AddNode("branch2", func(ctx context.Context, s TestState) (TestState, error) {
s.Log = append(s.Log, "branch2")
s.Value = 20
return s, nil
})
g.AddParallelEdges("start", []string{"branch1", "branch2"}, "",
func(ctx context.Context, parent TestState, branches []TestState) (TestState, error) {
for _, b := range branches {
parent.Value += b.Value
}
return parent, nil
},
)
cg, err := g.Compile()
if err != nil {
t.Fatalf("编译图失败: %v", err)
}
// 使用 Engine + StateCloner 运行
engine := NewEngine(cg).WithStateCloner(func(s TestState) TestState {
clone := s
clone.Log = append([]string{}, s.Log...)
return clone
})
thread, err := engine.Start(context.Background(), "start", TestState{})
if err != nil {
t.Fatalf("启动图失败: %v", err)
}
if thread.State.Value != 30 {
t.Errorf("期望 Value 为 30,实际为 %d", thread.State.Value)
}
}
// TestHookChaining 验证多次调用 WithPreNodeHook/WithPostNodeHook 时钩子按顺序叠加
func TestHookChaining(t *testing.T) {
g := NewGraph[TestState]()
g.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) {
return s, nil
})
cg, _ := g.Compile()
var log []string
engine := NewEngine(cg).
WithPreNodeHook(func(ctx context.Context, name string, s TestState) {
log = append(log, "pre1:"+name)
}).
WithPreNodeHook(func(ctx context.Context, name string, s TestState) {
log = append(log, "pre2:"+name)
}).
WithPostNodeHook(func(ctx context.Context, name string, s TestState) {
log = append(log, "post1:"+name)
}).
WithPostNodeHook(func(ctx context.Context, name string, s TestState) {
log = append(log, "post2:"+name)
})
engine.Start(context.Background(), "A", TestState{})
expected := []string{"pre1:A", "pre2:A", "post1:A", "post2:A"}
if len(log) != len(expected) {
t.Fatalf("期望 %d 次 hook 触发,实际 %d 次: %v", len(expected), len(log), log)
}
for i, entry := range expected {
if log[i] != entry {
t.Errorf("第 %d 次 hook 期望 %q,实际 %q", i, entry, log[i])
}
}
}
// TestCheckpointerThreadIDValidation 验证 FileCheckpointer 对恶意 threadID 的防御
func TestCheckpointerThreadIDValidation(t *testing.T) {
tmpDir := t.TempDir()
fc, err := NewFileCheckpointer[TestState](tmpDir)
if err != nil {
t.Fatalf("创建 FileCheckpointer 失败: %v", err)
}
thread := &Thread[TestState]{State: TestState{Value: 1}}
ctx := context.Background()
// 空 ID
if err := fc.Save(ctx, "", thread); err == nil {
t.Error("空 threadID 应报错")
}
// 路径穿越
if err := fc.Save(ctx, "../escape", thread); err == nil {
t.Error("包含 .. 的 threadID 应报错")
}
// 目录分隔符
if err := fc.Save(ctx, "sub/dir", thread); err == nil {
t.Error("包含 / 的 threadID 应报错")
}
// Load 同样校验
_, err = fc.Load(ctx, "../escape")
if err == nil {
t.Error("Load 包含 .. 的 threadID 应报错")
}
// 合法 ID 应成功
if err := fc.Save(ctx, "valid-id_123", thread); err != nil {
t.Errorf("合法 threadID 不应报错: %v", err)
}
// Delete 校验
if err := fc.Delete(ctx, "../escape"); err == nil {
t.Error("Delete 包含 .. 的 threadID 应报错")
}
}
// TestFileCheckpointerCompactJSON 验证默认格式为紧凑 JSON(不带缩进)
func TestFileCheckpointerCompactJSON(t *testing.T) {
tmpDir := t.TempDir()
// 默认紧凑格式
fc, _ := NewFileCheckpointer[TestState](tmpDir)
thread := &Thread[TestState]{State: TestState{Value: 42}, NextNode: "A"}
_ = fc.Save(context.Background(), "compact", thread)
data, _ := os.ReadFile(tmpDir + "/compact.json")
if len(data) == 0 {
t.Fatal("文件为空")
}
// 紧凑格式不应包含换行符
if strings.Contains(string(data), "\n") {
t.Error("默认模式应为紧凑 JSON,不应包含换行符")
}
// WithPrettyJSON 模式应包含换行符
fcPretty, _ := NewFileCheckpointer[TestState](tmpDir, WithPrettyJSON())
_ = fcPretty.Save(context.Background(), "pretty", thread)
dataPretty, _ := os.ReadFile(tmpDir + "/pretty.json")
if !strings.Contains(string(dataPretty), "\n") {
t.Error("WithPrettyJSON 模式应包含换行符")
}
}
+5
View File
@@ -0,0 +1,5 @@
module testmod
go 1.26.2
require github.com/unclesam-ly/GopherGraph v1.1.1 // indirect
+2
View File
@@ -0,0 +1,2 @@
github.com/unclesam-ly/GopherGraph v1.1.1 h1:1jou9QFtsom/eK69Evx+J3HLkZTjKjnuEuRvkir8qi4=
github.com/unclesam-ly/GopherGraph v1.1.1/go.mod h1:+99WeqLyTIYfOs2OSob/SQ3Pnw+R7pxW8+o6D7S4m6g=