PocketFlowGo

English | 中文

The Go implementation of PocketFlow — a minimalist LLM Agents framework in just ~100 lines of code.

Why PocketFlowGo?

Why Go for LLM Agents?

While Python dominates the AI/ML ecosystem, Go offers compelling advantages for production LLM agent deployments:

Aspect	Python	Go
Deployment	Requires runtime, virtualenv, dependencies	Single static binary
Performance	Interpreter overhead, GIL limitations	Native compilation, true parallelism
Concurrency	asyncio complexity, thread limitations	Goroutines + channels (simple & powerful)
Memory	Higher footprint	Efficient memory usage
Type Safety	Runtime errors	Compile-time guarantees
Ops Integration	Separate tooling	Native fit for K8s, Docker, cloud infrastructure

When to Choose PocketFlowGo

• Building production microservices that orchestrate LLM calls
• Deploying agents in containerized environments (K8s, Docker)
• Need high concurrency with thousands of simultaneous agent workflows
• Teams with Go-based infrastructure wanting consistent tech stack
• Edge deployment where binary size and startup time matter
• Building CLI tools that leverage LLM capabilities

Design Philosophy

Like the original PocketFlow, we believe:

"The best framework is the one you can fully understand."

PocketFlowGo maintains the same minimalist philosophy:

• ~100 lines of core code — easy to understand, modify, and extend
• Zero external dependencies — just the Go standard library
• No magic — explicit control flow, no hidden behaviors

Installation

go get github.com/lalolv/PocketFlowGo

Quick Start

package main

import (
    "fmt"
    pf "github.com/lalolv/PocketFlowGo"
)

func main() {
    // Create nodes
    node1 := pf.NewFuncNode(func(prepRes any) (any, error) {
        fmt.Println("Node 1 executing")
        return "result1", nil
    })

    node2 := pf.NewFuncNode(func(prepRes any) (any, error) {
        fmt.Println("Node 2 executing")
        return "result2", nil
    })

    // Chain nodes (equivalent to Python's node1 >> node2)
    node1.Then(node2)

    // Create and run flow
    flow := pf.NewFlow(node1)
    shared := pf.Shared{}
    flow.Run(shared)
}

Syntax Comparison: Python vs Go

Python PocketFlow	Go PocketFlowGo
node_a >> node_b	nodeA.Then(nodeB)
node_a - "action" >> node_b	nodeA.On("action").Then(nodeB)
async def exec_async()	func(ctx context.Context, prepRes any)
asyncio.gather()	goroutines + sync.WaitGroup

Node Types

Type	Description
BaseNode	Foundation with params/successors management
RetryNode	Adds retry logic with configurable attempts and wait
FuncNode	Convenient inline function node
BatchNode	Processes collections sequentially
AsyncNode	Async execution with context support
AsyncBatchNode	Async batch processing (sequential)
AsyncParallelBatchNode	Async batch processing (parallel with goroutines)

Flow Types

Type	Description
Flow	Basic node orchestration
BatchFlow	Process multiple parameter sets
AsyncFlow	Async node orchestration
AsyncBatchFlow	Async batch (sequential)
AsyncParallelBatchFlow	Async batch (parallel)

Examples

Conditional Routing

validator := pf.NewFuncNodeFull(
    func(shared pf.Shared) (any, error) {
        return shared["value"], nil
    },
    func(prepRes any) (any, error) {
        return prepRes.(int), nil
    },
    func(shared pf.Shared, prepRes, execRes any) (string, error) {
        if execRes.(int) > 100 {
            return "too_high", nil
        }
        return "valid", nil
    },
)

validator.On("valid").Then(successNode)
validator.On("too_high").Then(errorNode)

Retry with Fallback

node := pf.NewRetryNode(
    pf.WithMaxRetries(3),
    pf.WithWait(time.Second),
)
node.ExecFunc = func(prepRes any) (any, error) {
    return fetchData(prepRes)
}
node.ExecFallbackFunc = func(prepRes any, err error) (any, error) {
    return "fallback data", nil
}

Parallel Processing

node := pf.NewAsyncParallelBatchNode()
node.PrepFunc = func(ctx context.Context, shared pf.Shared) (any, error) {
    return []any{"url1", "url2", "url3"}, nil
}
node.ExecFunc = func(ctx context.Context, prepRes any) (any, error) {
    return fetch(ctx, prepRes.(string))
}

flow := pf.NewAsyncFlow(node)
flow.RunAsync(context.Background(), shared)

Run Examples

go run ./examples/basic/
go run ./examples/retry/
go run ./examples/conditional/
go run ./examples/parallel/

Related Projects

• PocketFlow — Original Python implementation
• PocketFlow Cookbook — Example patterns and recipes

License

MIT