bubo

README

Bubo

Bubo is a powerful Go library for building and orchestrating AI agents with a focus on reliability, extensibility, and maintainable workflows.

Overview

Bubo provides a robust foundation for creating AI-powered applications by offering:

• Agent Orchestration: Coordinate multiple AI agents working together
• Event-Driven Architecture: Built on a reliable event system for agent communication
• Provider Abstraction: Flexible integration with AI providers
• Tool System: Extensible framework for adding custom capabilities to agents
• Memory Management: Built-in short-term memory system for context retention
• Workflow Engine: Integration with Temporal for reliable workflow execution
• Message Broker: NATS integration for scalable message handling

Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│     Agent       │     │    Provider     │     │      Tool       │
│  Orchestration  │◄────┤   Integration   │◄────┤    System       │
└─────────────────┘     └─────────────────┘     └─────────────────┘
         ▲                      ▲                       ▲
         │                      │                       │
         ▼                      ▼                       ▼
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│     Event       │     │     Memory      │     │    Workflow     │
│     System      │◄────┤   Management    │◄────┤     Engine      │
└─────────────────┘     └─────────────────┘     └─────────────────┘

Core Components

• Agent: Manages AI agent lifecycle and coordination
• Provider: Abstracts AI provider integration (e.g., OpenAI)
• Tool: Extensible system for adding capabilities to agents
• Events: Reliable event system for agent communication
• Memory: Short-term memory management for context retention
• Workflow: Temporal integration for reliable execution

Installation

Prerequisites

• Go 1.20 or later
• NATS Server (for message broker)
• Temporal (for workflow engine)

Using Go Modules

go get github.com/casualjim/bubo

Basic Usage

Agent Handoff Example

This example demonstrates how to create agents that can transfer control between each other based on language:

package main

import (
    "context"
    "log/slog"
    "os"
    "time"

    // Ensure API Key is loaded
    _ "github.com/joho/godotenv/autoload"

    "github.com/casualjim/bubo"
    "github.com/casualjim/bubo/api"
    "github.com/casualjim/bubo/examples/internal/msgfmt"
    "github.com/casualjim/bubo/agent"
    "github.com/casualjim/bubo/provider/openai"
)

// Define agents with specific language capabilities
var (
    englishAgent = agent.New(
        agent.Name("English Agent"),
        agent.Model(openai.GPT4oMini()),
        agent.Instructions("You only speak English, so you only reply in english."),
        agent.Tools(transferToSpanishAgentTool),
    )
    spanishAgent = agent.New(
        agent.Name("Spanish Agent"),
        agent.Model(openai.GPT4oMini()),
        agent.Instructions("You only speak Spanish, so you only reply in spanish."),
    )
)

// Transfer spanish speaking users immediately
//
// bubo:agentTool
func transferToSpanishAgent() api.Agent { return spanishAgent }

func main() {
    ctx := context.Background()

    // Configure console output
    hook, result := msgfmt.Console[string](ctx, os.Stdout)

    // Create and run workflow
    p := bubo.New(
        bubo.Agents(englishAgent),
        bubo.Steps(
            bubo.Step(englishAgent.Name(), "Hola. ¿Como estás?"),
        ),
    )

    if err := p.Run(ctx, bubo.Local(hook)); err != nil {
        slog.Error("error running agent", "error", err)
        return
    }

    <-result
}

See the examples directory for more usage patterns including:

• Basic examples
  • Agent handoff
  • Context variables
  • Function calling
  • Structured output
• Temporal workflow integration
• Triage system implementation

Component Relationships

Agent ↔ Provider

Agents use providers to interact with AI models. The provider abstraction allows for easy integration of different AI services.

Agent ↔ Tool

Tools extend agent capabilities by providing specific functionalities. Tools can be generated using the bubo-tool-gen command for marker comments like:

// bubo:agentTool
func transferToSpanishAgent() api.Agent { return spanishAgent }

Agent ↔ Memory

The memory system helps agents maintain context across interactions and share information.

Provider ↔ Tool

Providers can utilize tools to enhance AI model capabilities and provide additional functionalities.

Contributing

Development Setup

Clone the repository:

git clone https://github.com/casualjim/bubo.git
cd bubo

Install dependencies:

go mod download

Run tests:

go test ./...

Guidelines

1. Code Style

   • Follow Go best practices and idioms
   • Use 2 spaces for indentation
   • Run golangci-lint run before submitting PRs

2. Testing

   • Write unit tests for new functionality
   • Ensure existing tests pass
   • For concurrent tests, use signals/events instead of timing

3. Documentation

   • Update documentation for new features
   • Include godoc examples for public APIs
   • Keep the README updated

4. Pull Requests

   • Create feature branches from main
   • Include tests and documentation
   • Ensure CI passes
   • Request review from maintainers

License

This project is licensed under the LICENSE file in the repository.

Documentation

Overview

Package bubo provides a framework for building and orchestrating AI agents with a focus on reliability, extensibility, and maintainable workflows.

The package implements a robust foundation for creating AI-powered applications through several key abstractions:

• Agents: Autonomous entities that can process tasks and make decisions
• Workflows: Sequences of steps that coordinate agent interactions
• Tools: Extensible capabilities that agents can use
• Events: Communication system between components
• Memory: Context retention across interactions

Basic Usage

A typical workflow involves creating agents, defining their capabilities, and orchestrating their interactions:

englishAgent := agent.New(
	agent.Name("English Agent"),
	agent.Model(openai.GPT4oMini()),
	agent.Instructions("You only speak English"),
	agent.Tools(transferTool),
)

p := bubo.New(
	bubo.Agents(englishAgent),
	bubo.Steps(
		bubo.Step(englishAgent.Name(), "Process this text"),
	),
)

if err := p.Run(ctx, bubo.Local(hook)); err != nil {
	// Handle error
}

Architecture

The package is built around several core concepts:

1. Execution Engine (execution.go)

• Manages the lifecycle of workflows
• Handles state transitions and error recovery
• Coordinates between agents and tools

2. Hooks (hook.go)

• Provides extension points for workflow customization
• Enables monitoring and logging
• Allows integration with external systems

3. Promises (promise.go)

• Implements asynchronous operation handling
• Manages concurrent agent tasks
• Provides error propagation

4. Tasks (task.go)

• Represents units of work
• Encapsulates agent instructions
• Manages execution context

5. Knots (knot.go)

• Implements workflow synchronization points
• Manages agent handoffs
• Coordinates multi-agent interactions

Integration

Bubo integrates with several backend systems:

• NATS for message brokering
• Temporal for workflow orchestration
• Various AI providers (e.g., OpenAI)

Examples

The examples directory contains various usage patterns:

• Basic agent interactions
• Context variable handling
• Function calling
• Agent handoff scenarios
• Temporal workflow integration
• Triage system implementation

See the examples directory for complete implementations.

Best Practices

1. Agent Design

• Keep agent instructions focused and specific
• Use tools for extending agent capabilities
• Implement proper error handling

2. Workflow Management

• Break complex workflows into manageable steps
• Use hooks for monitoring and debugging
• Implement proper context cancellation

3. Tool Implementation

• Keep tools simple and focused
• Document tool capabilities clearly
• Use the bubo-tool-gen for consistent generation

4. Error Handling

• Always check for errors from Run()
• Implement proper cleanup in hooks
• Use context for cancellation

Thread Safety

The package is designed to be thread-safe when used correctly:

• Agents can be shared across goroutines
• Hooks should be implemented in a thread-safe manner
• Context is used for cancellation and deadlines

For more information about specific components, see their respective documentation:

• agent.Agent for agent implementation details
• provider.Provider for AI provider integration
• tool.Tool for implementing custom capabilities

Package bubo provides a framework for building conversational AI agents that can interact in a structured manner. It supports multi-agent conversations, structured output, and flexible execution contexts.

The execution package is a core component that handles:

• Conversation workflow execution
• Context and state management
• Structured output handling
• Event hooks and promises
• Resource cleanup

Key concepts:

• ExecutionContext: Manages the execution environment and configuration
• Promises: Handles asynchronous result processing
• Hooks: Provides extension points for monitoring and control
• Structured Output: Enables type-safe response handling

Package bubo provides a framework for building conversational AI agents that can interact in a structured manner. It supports multi-agent conversations, structured output, and flexible execution contexts.

Package bubo provides a framework for building conversational AI agents that can interact in a structured manner. It supports multi-agent conversations, structured output, and flexible execution contexts.

Package bubo provides a framework for building conversational AI agents that can interact in a structured manner. It supports multi-agent conversations, structured output, and flexible execution contexts.

Package bubo provides a framework for building conversational AI agents that can interact in a structured manner. It supports multi-agent conversations, structured output, and flexible execution contexts.

Index

• Variables
• func Agents(agent api.Agent, extraAgents ...api.Agent) opts.Option[Knot]
• func Steps(step ConversationStep, extraSteps ...ConversationStep) opts.Option[Knot]
• func StructuredOutput[T any](name, description string) opts.Option[ExecutionContext]
• type ConversationStep
  • func Step[T Task](agentName string, tsk T) ConversationStep
• type ExecutionContext
  • func Local[T any](hook Hook[T], options ...opts.Option[ExecutionContext]) ExecutionContext
• type Future
• type Hook
• type Knot
  • func New(options ...opts.Option[Knot]) *Knot
  • func (p *Knot) Run(ctx context.Context, rc ExecutionContext) error
• type Task

Variables

var (
	// WithContextVars is an option to set context variables for the execution context.
	// These variables are made available to agents during execution and can be used
	// to pass dynamic configuration or state information.
	//
	// Example:
	//  Local(hook, WithContextVars(types.ContextVars{
	//      "user_id": "123",
	//      "preferences": map[string]string{"lang": "en"},
	//  }))
	WithContextVars = opts.ForName[ExecutionContext, types.ContextVars]("contextVars")

	// Streaming is an option to enable/disable response streaming.
	// When enabled, responses are sent incrementally as they become available.
	// When disabled, responses are sent only after completion.
	//
	// Example:
	//  Local(hook, Streaming(true))
	Streaming = opts.ForName[ExecutionContext, bool]("stream")

	// WithMaxTurns is an option to set the maximum number of conversation turns.
	// This helps prevent infinite loops and control resource usage.
	// A turn consists of one complete agent interaction cycle.
	//
	// Example:
	//  Local(hook, WithMaxTurns(5))
	WithMaxTurns = opts.ForName[ExecutionContext, int]("maxTurns")
)

var Name = opts.ForName[Knot, string]("name")

Name is an option to set the name of the conversation initiator.

Functions

func Agents

func Agents(agent api.Agent, extraAgents ...api.Agent) opts.Option[Knot]

Agents creates an option to register one or more agents with the Knot. It requires at least one agent and can accept additional agents as variadic arguments.

func Steps

func Steps(step ConversationStep, extraSteps ...ConversationStep) opts.Option[Knot]

Steps creates an option to add one or more conversation steps to the Knot. Each step represents a single interaction in the conversation flow.

func StructuredOutput

func StructuredOutput[T any](name, description string) opts.Option[ExecutionContext]

StructuredOutput creates an option to configure structured output for responses. It generates a JSON schema for type T and associates it with the given name and description. The schema is used to validate and structure the conversation output.

This option enables type-safe handling of conversation results by:

• Generating a JSON schema from the Go type
• Validating responses against the schema
• Providing structured data to result handlers

Example usage:

type Response struct {
    Status  string `json:"status"`
    Message string `json:"message"`
}

ctx := Local[Response](hook,
    StructuredOutput[Response](
        "status_response",
        "Response with status and message",
    ),
)

Parameters:

• name: Identifier for this output format
• description: Human-readable explanation of the format

Type Parameters:

• T: The Go type to generate a schema for

Types

type ConversationStep

type ConversationStep struct {
	// contains filtered or unexported fields
}

ConversationStep represents a single interaction step in a conversation workflow. It pairs an agent with a specific task to be executed.

func Step

func Step[T Task](agentName string, tsk T) ConversationStep

Step creates a new ConversationStep with the specified agent and task. It accepts either a string or a Message[UserMessage] as the task input. The function will panic if an invalid task type is provided.

type ExecutionContext

type ExecutionContext struct {
	// contains filtered or unexported fields
}

ExecutionContext holds the configuration and state for executing conversation steps. It manages the execution environment, event hooks, promises for handling results, and various execution parameters.

The context is responsible for:

• Managing the execution environment
• Handling event hooks and promises
• Configuring response formats
• Managing conversation state
• Enforcing execution limits
• Coordinating cleanup

Thread Safety: ExecutionContext is designed to be used by a single conversation workflow. While its components (hooks, promises) may be thread-safe, the context itself should not be shared across multiple concurrent conversations.

func Local

func Local[T any](hook Hook[T], options ...opts.Option[ExecutionContext]) ExecutionContext

Local creates a new ExecutionContext configured for local execution. It sets up a future-based promise system with the provided hook for handling results of type T. The context can be further customized using the provided options.

The function is generic over type T, which represents the expected result type of the conversation. This enables type-safe handling of conversation outputs.

Example usage:

hook := &MyHook[string]{} // Custom hook implementation
ctx := Local[string](hook,
    WithContextVars(types.ContextVars{"user": "alice"}),
    Streaming(true),
    WithMaxTurns(5),
    StructuredOutput[MyResponse]("response", "Structured response format"),
)

Parameters:

• hook: Implementation of Hook[T] for handling results and lifecycle events
• options: Optional configuration parameters for the execution context

The created context includes:

• Local executor for running commands
• Future-based promise system
• Event hook integration
• Automatic cleanup on completion

type Future

type Future[T any] interface {
	// Get retrieves the value once it's available.
	// Returns the value of type T and any error that occurred during computation.
	Get() (T, error)
}

Future represents a value of type T that will be available in the future. It provides a way to retrieve the value once it's ready. Note: This interface cannot be type aliased yet due to the type parameter T.

type Hook

type Hook[T any] interface {
	// events.Hook provides the base event handling interface:
	//   - OnStart: Called when a conversation starts
	//   - OnError: Called when an error occurs
	//   - OnMessage: Called for each message in the conversation
	//   - OnToolCall: Called when a tool is invoked
	//   - OnToolResponse: Called when a tool returns a result
	events.Hook // Embeds the base Hook interface for common event handling

	// OnResult is called when a conversation step produces a result of type T.
	// This method allows for type-safe handling of conversation outputs.
	//
	// The context.Context parameter can be used for cancellation and timeouts.
	// The type parameter T represents the expected result type, which must match
	// the type parameter used when implementing the Hook interface.
	//
	// This method should:
	//   - Process results quickly to avoid blocking the conversation
	//   - Handle results in a thread-safe manner if shared across goroutines
	//   - Respect context cancellation
	//   - Not modify the result parameter as it may be shared
	OnResult(context.Context, T)

	// OnClose is called when a conversation workflow completes.
	// It provides an opportunity for cleanup and resource release.
	//
	// This method is called exactly once per conversation, regardless of whether
	// the conversation completed successfully or encountered an error. It should:
	//   - Release any resources acquired during the conversation
	//   - Close any open connections or files
	//   - Flush any buffered data
	//   - Complete any final logging or metrics
	//
	// The context.Context parameter may be used for timeout management during cleanup,
	// but implementations should attempt to clean up resources even if the context
	// is cancelled.
	OnClose(context.Context)
}

Hook extends the events.Hook interface to provide type-safe result handling and cleanup functionality. It's parameterized by type T which represents the expected type of conversation results.

Hooks serve several important purposes in the system:

• Monitoring conversation progress
• Handling intermediate and final results
• Managing resources and cleanup
• Implementing custom logging or metrics
• Integrating with external systems

Example implementation:

type LoggingHook[T any] struct {
    events.BaseHook // Provides default implementations
    logger *log.Logger
}

func (h *LoggingHook[T]) OnResult(ctx context.Context, result T) {
    h.logger.Printf("Received result: %v", result)
}

func (h *LoggingHook[T]) OnClose(ctx context.Context) {
    h.logger.Print("Conversation completed")
}

Usage:

hook := &LoggingHook[string]{
    logger: log.New(os.Stdout, "", log.LstdFlags),
}

p := bubo.New(
    bubo.Agents(agent),
    bubo.Steps(step),
)

if err := p.Run(ctx, bubo.Local(hook)); err != nil {
    // Handle error
}

type Knot

type Knot struct {
	// contains filtered or unexported fields
}

Knot represents a conversational workflow that coordinates multiple AI agents through a series of predefined steps. It manages agent registration and execution of conversation steps in sequence.

func New

func New(options ...opts.Option[Knot]) *Knot

New creates a new Knot instance with the provided options. It initializes a default name of "User" and an empty agent registry. Options can be used to customize the name, add agents, and define conversation steps.

func (*Knot) Run

func (p *Knot) Run(ctx context.Context, rc ExecutionContext) error

Run executes the conversation workflow defined by the Knot's steps. It processes each step sequentially using the provided execution context. The last step's output can be structured according to the response schema if specified.

type Task

type Task interface {
	~string | messages.Message[messages.UserMessage]
}

Task is a type constraint interface that defines valid task types that can be used to create conversation steps. It allows either string literals or structured messages as valid task inputs.