pith

Pith

Go SDK for building LLM agent applications. Four layers — protocol types, LLM gateway, agent loop, stateful agent — each with a strict dependency boundary and no runtime assumptions.

The built-in OpenAICompatProvider covers any /v1/chat/completions API (OpenAI, Groq, Together, DeepSeek, OpenRouter, Ollama, etc.). Custom providers (e.g., Anthropic Messages API) are built using the ProviderPort interface — see example 11.

Build agents quickly → pith-sdk

Multi-module repo. Import a layer directly: agent · gateway · loop · protocol

Installation

Import the layer you need. The agent module pulls in gateway, loop, and protocol:

go get github.com/chinudotdev/pith/agent@v0.1.4
go get github.com/chinudotdev/pith/gateway@v0.1.4
go get github.com/chinudotdev/pith/loop@v0.1.4
go get github.com/chinudotdev/pith/protocol@v0.1.4

Requires Go 1.24+.

Examples

Examples are standalone programs (not listed in go.work). Copy one into a new module and run it — see examples/README.md:

mkdir my-agent && cd my-agent
go mod init my-agent
cp /path/to/pith/examples/01-minimal/main.go .
go get github.com/chinudotdev/pith/gateway@v0.1.4
OPENAI_API_KEY="sk-..." go run main.go

Quick Reference

import (
    "context"
    "os"

    "github.com/chinudotdev/pith/agent"
    "github.com/chinudotdev/pith/gateway"
    "github.com/chinudotdev/pith/gateway/providers"
    "github.com/chinudotdev/pith/protocol"
)

// --- L1: Gateway ---

gw := gateway.NewLLMGateway()
gw.Providers.Register(providers.NewOpenAICompatProvider(providers.OpenAICompatConfig{
    BaseURL: "https://api.openai.com",
}))
gw.Credentials = gateway.CredentialProviderFunc(func(pid protocol.ProviderId) (protocol.Credential, error) {
    return protocol.ApiKey{Key: os.Getenv("OPENAI_API_KEY")}, nil
})
gw.Catalog.Register("openai", protocol.ModelDescriptor{
    ID: "gpt-4o-mini", API: protocol.ApiOpenAICompletions, Provider: "openai",
    BaseURL: "https://api.openai.com", ContextWindow: 128000, MaxTokens: 4096,
})

model, _ := gw.Catalog.Get("openai", "gpt-4o-mini")

stream, _ := gw.Stream(ctx, model, protocol.Context{
    SystemPrompt: "You are helpful.",
}, opts)
for event := range stream {
    if event.Type == protocol.EventTextDelta {
        fmt.Print(event.Delta)
    }
}

// --- L3: Agent ---

ag := agent.NewAgent(agent.AgentConfig{
    InitialState: &agent.AgentState{
        Model:        model,
        SystemPrompt: "You are helpful.",
        Tools:        myTools,
    },
    StreamFn: func(ctx context.Context, m protocol.ModelDescriptor, pctx protocol.Context, opts protocol.StreamOptions) (<-chan protocol.StreamEvent, error) {
        return gw.Stream(ctx, m, pctx, opts)
    },
})

ag.EventBus().Subscribe(func(e agent.AgentEvent) {
    if e.LoopEvent != nil && e.LoopEvent.Type == "messageUpdate" {
        if e.LoopEvent.StreamEvent != nil && e.LoopEvent.StreamEvent.Type == protocol.EventTextDelta {
            fmt.Print(e.LoopEvent.StreamEvent.Delta)
        }
    }
})

ag.Prompt(ctx, "Hello!")

// --- Compaction ---

if agent.ShouldCompact(ag.State().Messages, agent.CompactionSettings{Enabled: true}, model) {
    tokens := agent.EstimateTokens(ag.State().Messages)
    ag.Compact("Summary of conversation", 2, tokens, nil)
}

// --- Steering & Follow-Up ---

ag.FollowUp(protocol.UserMessage{Role: "user", Content: []protocol.Content{protocol.TextContent{Type: "text", Text: "Now explain that."}}, Timestamp: protocol.Now()})
ag.Prompt(ctx, "What is Go?")

ag.Steer(protocol.UserMessage{Role: "user", Content: []protocol.Content{protocol.TextContent{Type: "text", Text: "Be more concise."}}, Timestamp: protocol.Now()})

// --- Abort ---

go ag.Prompt(ctx, "Long running task...")
time.Sleep(100 * time.Millisecond)
ag.Abort()
ag.WaitForIdle()

API Primitives