Observability

The Doclo SDK provides comprehensive observability through lifecycle hooks, metrics aggregation, and cloud integration. Monitor every step of your document processing pipelines.

Local Observability

Configure observability hooks when creating a flow:

import { createFlow, extract } from '@doclo/flows';

const flow = createFlow({
  observability: {
    onFlowStart: (ctx) => {
      console.log(`Flow ${ctx.flowId} started`);
    },
    onFlowEnd: (ctx) => {
      console.log(`Flow completed in ${ctx.duration}ms`);
      console.log(`Cost: $${ctx.stats?.totalCostUSD.toFixed(4)}`);
    },
    onFlowError: (ctx) => {
      console.error(`Flow failed: ${ctx.error.message}`);
    },
    onStepStart: (ctx) => {
      console.log(`Step ${ctx.stepId} starting...`);
    },
    onStepEnd: (ctx) => {
      console.log(`Step ${ctx.stepId}: ${ctx.duration}ms, $${ctx.cost?.toFixed(4)}`);
    },
    onStepError: (ctx) => {
      console.error(`Step ${ctx.stepId} failed: ${ctx.error.message}`);
    }
  }
})
  .step('extract', extract({ provider, schema }))
  .build();

Available Hooks

Flow-Level Hooks

Hook	Trigger	Context
`onFlowStart`	Flow execution begins	`flowId`, `executionId`, `input`, `metadata`
`onFlowEnd`	Flow completes successfully	`duration`, `output`, `stats`, `traceContext`
`onFlowError`	Flow fails	`error`, `errorCode`, `failedAtStepIndex`

Step-Level Hooks

Hook	Trigger	Context
`onStepStart`	Step execution begins	`stepId`, `stepIndex`, `stepType`, `provider`, `model`
`onStepEnd`	Step completes successfully	`duration`, `output`, `usage`, `cost`
`onStepError`	Step fails	`error`, `willRetry`, `retryAttempt`

Consensus Hooks

Hook	Trigger	Context
`onConsensusStart`	Consensus voting begins	`runsPlanned`, `strategy`
`onConsensusRunComplete`	Individual run completes	`runIndex`, `output`, `status`
`onConsensusComplete`	All runs complete	`agreement`, `agreedOutput`, `totalCost`

Batch/forEach Hooks

Hook	Trigger	Context
`onBatchStart`	forEach begins	`totalItems`, `batchId`
`onBatchItemEnd`	Single item processed	`itemIndex`, `result`, `status`
`onBatchEnd`	All items processed	`successfulItems`, `failedItems`

Provider-Level Hooks

Hook	Trigger	Context
`onProviderRequest`	API request sent	`provider`, `model`, `input`
`onProviderResponse`	API response received	`output`, `usage`, `cost`, `httpStatusCode`
`onProviderRetry`	Request will retry	`error`, `attemptNumber`, `nextRetryDelay`

Cloud Observability

Send execution events to the Doclo Cloud dashboard for monitoring and analytics.

Using createCloudObservability

import { createCloudObservability } from '@doclo/client';
import { createFlow, extract } from '@doclo/flows';

// Create cloud observability transport
const cloudObs = createCloudObservability({
  client: docloClient,
  flowId: 'invoice-extraction',
  flowVersion: '1.0.0',
  mode: 'stream',           // Real-time events
  flushIntervalMs: 5000,    // Flush every 5 seconds
  includeInputs: false,     // Privacy: exclude input data
  includeOutputs: true      // Include extraction results
});

// Create flow with cloud observability
const flow = createFlow({ observability: cloudObs })
  .step('extract', extract({ provider, schema }))
  .build();

// Run the flow
const result = await flow.run(input);

// Ensure all events are sent
await cloudObs.flush();

Configuration Options

Option	Type	Default	Description
`client`	`DocloClient`	Required	Doclo client instance
`flowId`	`string`	Required	Flow ID for tracking
`flowVersion`	`string`	-	Flow version
`mode`	`'stream' \| 'batch-at-end'`	`'stream'`	When to send events
`flushIntervalMs`	`number`	`5000`	Flush interval (0 = immediate)
`batchSize`	`number`	`50`	Events before auto-flush
`maxRetries`	`number`	`3`	Retry attempts for failed sends
`retryDelayMs`	`number`	`1000`	Base retry delay
`maxBufferSize`	`number`	`1000`	Max buffered events
`includeInputs`	`boolean`	`false`	Include input data in events
`includeOutputs`	`boolean`	`false`	Include output data in events
`onError`	`function`	-	Called when events are dropped

Transport Modes

Stream Mode (Default)

Events are sent periodically during execution for real-time dashboard updates:

createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'stream',
  flushIntervalMs: 5000,  // Flush every 5 seconds
  batchSize: 50           // Or when 50 events buffered
});

Set flushIntervalMs: 0 for immediate per-event sending:

createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'stream',
  flushIntervalMs: 0  // Send immediately
});

Batch-at-End Mode

Events are collected and sent only when flush() is called:

const obs = createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'batch-at-end'
});

await flow.run(input);
await obs.flush();  // Send all events at once

Privacy Controls

Control what data is sent to the cloud:

createCloudObservability({
  client,
  flowId: 'my-flow',
  includeInputs: false,   // Don't send document content
  includeOutputs: true    // Send extraction results
});

Error Handling

Handle dropped events and transport errors:

createCloudObservability({
  client,
  flowId: 'my-flow',
  onError: (error, droppedCount) => {
    console.error(`Dropped ${droppedCount} events:`, error.message);
  }
});

Metrics

Every flow execution returns aggregated metrics:

const result = await flow.run(input);

console.log('Metrics:', result.aggregated);
// {
//   totalDurationMs: 2500,
//   totalCostUSD: 0.0042,
//   totalInputTokens: 1500,
//   totalOutputTokens: 200,
//   stepCount: 2
// }

// Per-step metrics
console.log('Step metrics:', result.metrics);
// [
//   { step: 'parse', ms: 1200, costUSD: 0.0010 },
//   { step: 'extract', ms: 1300, costUSD: 0.0032 }
// ]

Distributed Tracing

The SDK supports W3C trace context for distributed tracing:

const flow = createFlow({
  observability: {
    onFlowStart: (ctx) => {
      // W3C trace context available
      console.log('Trace ID:', ctx.traceContext.traceId);
      console.log('Span ID:', ctx.traceContext.spanId);
    }
  }
});

Propagating Trace Context

Pass trace context from upstream services:

const result = await flow.run(input, {
  metadata: {
    traceParent: req.headers['traceparent']
  }
});

Combining Local and Cloud

Use both local logging and cloud observability:

const cloudObs = createCloudObservability({
  client,
  flowId: 'my-flow'
});

const flow = createFlow({
  observability: {
    ...cloudObs,
    // Add local logging on top of cloud
    onStepEnd: (ctx) => {
      cloudObs.onStepEnd?.(ctx);
      console.log(`Step ${ctx.stepId}: ${ctx.duration}ms`);
    }
  }
});

Getting Started

Concepts

SDK

Doclo Cloud

Guides

Resources

Observability

Local Observability

Available Hooks

Flow-Level Hooks

Step-Level Hooks

Consensus Hooks

Batch/forEach Hooks

Provider-Level Hooks

Cloud Observability

Using createCloudObservability

Configuration Options

Transport Modes

Stream Mode (Default)

Batch-at-End Mode

Privacy Controls

Error Handling

Metrics

Distributed Tracing

Propagating Trace Context

Combining Local and Cloud

Next Steps

Hybrid Client

Error Handling

​Local Observability

​Available Hooks

​Flow-Level Hooks

​Step-Level Hooks

​Consensus Hooks

​Batch/forEach Hooks

​Provider-Level Hooks

​Cloud Observability

​Using createCloudObservability

​Configuration Options

​Transport Modes

​Stream Mode (Default)

​Batch-at-End Mode

​Privacy Controls

​Error Handling

​Metrics

​Distributed Tracing

​Propagating Trace Context

​Combining Local and Cloud

​Next Steps

Hybrid Client

Error Handling

Local Observability

Available Hooks

Flow-Level Hooks

Step-Level Hooks

Consensus Hooks

Batch/forEach Hooks

Provider-Level Hooks

Cloud Observability

Using createCloudObservability

Configuration Options

Transport Modes

Stream Mode (Default)

Batch-at-End Mode

Privacy Controls

Error Handling

Metrics

Distributed Tracing

Propagating Trace Context

Combining Local and Cloud

Next Steps