Chapter 6: Callback and Trace (Observability)

Goal of this chapter: understand the Callback mechanism and integrate CozeLoop for tracing and observability.

Code Location

• Entry code: cmd/ch06/main.go

Prerequisites

Same as Chapter 1: you need a configured and available ChatModel (OpenAI or Ark). Also, like Chapter 4, set PROJECT_ROOT:

export PROJECT_ROOT=/path/to/eino  # Eino core library root directory (defaults to current directory if not set)

Optional: configure CozeLoop for tracing:

export COZELOOP_WORKSPACE_ID=your_workspace_id
export COZELOOP_API_TOKEN=your_token

Running

In the examples/quickstart/chatwitheino directory:

# Set project root directory
export PROJECT_ROOT=/path/to/your/project

# Optional: configure CozeLoop
export COZELOOP_WORKSPACE_ID=your_workspace_id
export COZELOOP_API_TOKEN=your_token

go run ./cmd/ch06

Example output:

[trace] starting session: 083d16da-6b13-4fe6-afb0-c45d8f490ce1
you> Hello
[trace] chat_model_generate: model=gpt-4.1-mini tokens=150
[trace] tool_call: name=list_files duration=23ms
[assistant] Hello! How can I help you?

From Black Box to White Box: Why We Need Callbacks

In previous chapters, the Agent we built was a “black box”: questions go in, answers come out, but what happened in between was opaque.

Problems with a black box:

• Don’t know how many times the model was called
• Don’t know how long Tool execution took
• Don’t know how many tokens were consumed
• Hard to locate the root cause when issues arise

Callback’s role:

• Callback is Eino’s sidecar mechanism: consistent from component to compose (discussed below) to ADK
• Callback triggers at fixed points: 5 key moments in a component’s lifecycle
• Callback extracts real-time information: inputs, outputs, errors, streaming data, etc.
• Callback has broad applications: observability, logging, metrics, tracing, debugging, auditing, etc.

Simple analogy:

• Agent = “business logic” (main path)
• Callback = “sidecar hooks” (extract information at fixed points)

Key Concepts

Handler Interface

Handler is the core interface in Eino for defining callback handlers:

type Handler interface {
    // Non-streaming input (before component starts processing)
    OnStart(ctx context.Context, info *RunInfo, input CallbackInput) context.Context
    
    // Non-streaming output (after component returns successfully)
    OnEnd(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context
    
    // Error (when component returns an error)
    OnError(ctx context.Context, info *RunInfo, err error) context.Context
    
    // Streaming input (when component receives streaming input)
    OnStartWithStreamInput(ctx context.Context, info *RunInfo, 
        input *schema.StreamReader[CallbackInput]) context.Context
    
    // Streaming output (when component returns streaming output)
    OnEndWithStreamOutput(ctx context.Context, info *RunInfo, 
        output *schema.StreamReader[CallbackOutput]) context.Context
}

Design philosophy:

• Sidecar mechanism: does not interfere with the main flow, extracts information at fixed points
• Full coverage: all components from component to compose to ADK support callbacks
• State passing: the same Handler’s OnStart→OnEnd can pass state via context
• Performance optimization: implement the TimingChecker interface to skip unnecessary timings

RunInfo structure:

type RunInfo struct {
    Name      string        // Business name (node name or user-specified)
    Type      string        // Implementation type (e.g., "OpenAI")
    Component string        // Component type (e.g., "ChatModel")
}

Important notes:

• Streaming callbacks must close the StreamReader, otherwise goroutine leaks will occur
• Do not modify Input/Output — they are shared with all downstream consumers
• RunInfo may be nil; check before use

CozeLoop

CozeLoop is ByteDance’s open-source AI application observability platform, providing:

• Tracing: complete call chain visualization
• Metrics monitoring: latency, token consumption, error rates, etc.
• Log aggregation: centralized log management
• Debug support: online viewing and debugging

Integration:

import (
    clc "github.com/cloudwego/eino-ext/callbacks/cozeloop"
    "github.com/cloudwego/eino/callbacks"
    "github.com/coze-dev/cozeloop-go"
)

// Create CozeLoop client
client, err := cozeloop.NewClient(
    cozeloop.WithAPIToken(apiToken),
    cozeloop.WithWorkspaceID(workspaceID),
)

// Register as global Callback
callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))

Callback Trigger Timings

Callbacks trigger at 5 key moments in a component’s lifecycle. The Timing* constants in the table below are Eino internal constant names (used for the TimingChecker interface); the corresponding Handler interface methods are shown on the right:

Example: ChatModel call flow

┌─────────────────────────────────────────┐
│  ChatModel.Generate(ctx, messages)      │
└─────────────────────────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  OnStart             │  ← Input: CallbackInput (messages)
        └──────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  Model processing    │
        └──────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  OnEnd               │  ← Output: CallbackOutput (response)
        └──────────────────────┘

Example: Streaming output flow

┌─────────────────────────────────────────┐
│  ChatModel.Stream(ctx, messages)        │
└─────────────────────────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  OnStart             │  ← Input: CallbackInput (messages)
        └──────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  Model processing    │
        │  (streaming)         │
        └──────────────────────┘
                   ↓
        ┌──────────────────────────┐
        │  OnEndWithStreamOutput   │  ← Output: StreamReader[CallbackOutput]
        └──────────────────────────┘
                   ↓
        ┌──────────────────────┐
        │  Chunks returned     │
        │  one by one          │
        └──────────────────────┘

Notes:

• Streaming errors (errors mid-stream) do not trigger OnError; they are returned within the StreamReader
• The same Handler’s OnStart→OnEnd can pass state via context
• There is no guaranteed execution order between different Handlers

Callback Implementation

1. Implement a Custom Callback Handler

Fully implementing the Handler interface requires implementing all 5 methods, which is verbose. Eino provides the callbacks.HandlerHelper utility class to simplify implementation:

import "github.com/cloudwego/eino/callbacks"

// Use NewHandlerHelper to register callbacks of interest
handler := callbacks.NewHandlerHelper().
    OnStart(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        log.Printf("[trace] %s/%s start", info.Component, info.Name)
        return ctx
    }).
    OnEnd(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        log.Printf("[trace] %s/%s end", info.Component, info.Name)
        return ctx
    }).
    OnError(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
        log.Printf("[trace] %s/%s error: %v", info.Component, info.Name, err)
        return ctx
    }).
    Handler()

// Register as global Callback
callbacks.AppendGlobalHandlers(handler)

Note: RunInfo may be nil (e.g., top-level calls without RunInfo); check before use.

2. Integrate CozeLoop

// Setup CozeLoop tracing (optional)
// Set COZELOOP_API_TOKEN and COZELOOP_WORKSPACE_ID to enable
cozeloopApiToken := os.Getenv("COZELOOP_API_TOKEN")
cozeloopWorkspaceID := os.Getenv("COZELOOP_WORKSPACE_ID")
if cozeloopApiToken != "" && cozeloopWorkspaceID != "" {
    client, err := cozeloop.NewClient(
        cozeloop.WithAPIToken(cozeloopApiToken),
        cozeloop.WithWorkspaceID(cozeloopWorkspaceID),
    )
    if err != nil {
        log.Fatalf("cozeloop.NewClient failed: %v", err)
    }
    defer func() {
        time.Sleep(5 * time.Second)
        client.Close(ctx)
    }()
    callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
    log.Println("CozeLoop tracing enabled")
} else {
    log.Println("CozeLoop tracing disabled (set COZELOOP_API_TOKEN and COZELOOP_WORKSPACE_ID to enable)")
}

Key code snippet (Note: this is a simplified snippet that cannot run directly; see cmd/ch06/main.go for the full code):

// Setup CozeLoop tracing
cozeloopApiToken := os.Getenv("COZELOOP_API_TOKEN")
cozeloopWorkspaceID := os.Getenv("COZELOOP_WORKSPACE_ID")
if cozeloopApiToken != "" && cozeloopWorkspaceID != "" {
    client, err := cozeloop.NewClient(
        cozeloop.WithAPIToken(cozeloopApiToken),
        cozeloop.WithWorkspaceID(cozeloopWorkspaceID),
    )
    if err != nil {
        log.Fatalf("cozeloop.NewClient failed: %v", err)
    }
    defer func() {
        time.Sleep(5 * time.Second)
        client.Close(ctx)
    }()
    callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
}

The Value of Observability

1. Performance Analysis

With data collected via Callbacks, you can analyze:

• Model call latency distribution
• Tool execution time rankings
• Token consumption trends

2. Error Tracing

When the Agent has problems:

• View the complete call chain
• Identify which step failed
• Analyze the root cause

3. Cost Optimization

Through token consumption data:

• Identify high-consumption conversations
• Optimize prompts to reduce tokens
• Choose more economical models

Chapter Summary

• Callback: Eino’s observability hooks, triggered at key lifecycle points
• CozeLoop: ByteDance’s AI application observability platform
• Global registration: register global Callbacks via callbacks.AppendGlobalHandlers
• Non-invasive: business code doesn’t need modification; Callbacks trigger automatically
• Observability value: performance analysis, error tracing, cost optimization

Extended Thinking

Other Callback implementations:

• OpenTelemetry Callback: integrate with standard observability protocols
• Custom logging Callback: log to local files
• Metrics Callback: integrate with monitoring systems like Prometheus

Advanced usage:

• Implement sampling in Callbacks (only record a subset of requests)
• Implement rate limiting in Callbacks (based on token consumption)
• Implement alerting in Callbacks (notify when error rate is too high)