←back to tutorials

Multi-Stage Data Extraction with ChatBotKit Complete API

Learn how to perform progressive data extraction at multiple stages during conversation processing using ChatBotKit's complete API, enabling real-time structured data capture beyond traditional end-of-conversation extraction.

This tutorial demonstrates how to perform multi-stage data extraction using ChatBotKit's complete API, a powerful capability that distinguishes ChatBotKit from native APIs provided by OpenAI and other providers.

Understanding the Difference

Traditional Approach (OpenAI & Others)

Most AI providers, including OpenAI, perform data extraction only at the very end of the conversation flow:

  1. The conversation runs to completion
  2. At the end, structured output is extracted using a single schema
  3. You get one extraction result when everything is done

This approach works for simple scenarios but has limitations:

  • No progressive data capture during multi-step processes
  • Cannot extract intermediate results from sub-agents or parallel tasks
  • Limited visibility into extraction progress

ChatBotKit's Multi-Stage Approach

ChatBotKit enables progressive extraction at multiple stages during conversation processing:

  1. Define inline functions with extraction schemas
  2. Functions are called automatically during conversation flow
  3. Monitor function calls to capture structured data as they happen
  4. Extract data from multiple sources (orchestrator, sub-agents, parallel tasks)
  5. Compile results progressively rather than waiting for completion

This is achieved through:

  • Inline functions with JSON schemas that act as extraction points
  • Pre-canned function results that allow the conversation to continue seamlessly
  • Event streaming that lets you monitor and capture function calls in real-time

Real-World Use Case: Priority Gathering System

Let's build a system that gathers priorities from multiple AI agents and extracts them progressively. This example demonstrates multi-stage extraction in action.

Architecture Overview

flowchart TD
    A[User Request] --> B[conversation.complete with<br/>extraction functions]
    B --> C[Stream events immediately]
    C --> D{Event Type?}
    D -->|message| E{Message Type?}
    D -->|token| F[Stream response text]
    E -->|activity| G{Activity Type?}
    E -->|other| H[Display message]
    G -->|request| I[Capture extracted data<br/>from function arguments]
    I --> J[Store/Process data]
    F --> J
    H --> J
    J --> K[Return compiled results]

Step 1: Define the Extraction Schema

First, define the data structure you want to extract. This becomes the schema for your inline function:

// Priority item structure
const prioritySchema = {
  type: 'object',
  properties: {
    title: {
      type: 'string',
      description: 'Brief title of the priority',
    },
    description: {
      type: 'string',
      description: 'Detailed description of why this is important and what action to take',
    },
    importance: {
      type: 'string',
      enum: ['critical', 'high', 'medium', 'low'],
      description: 'The importance level of this priority',
    },
    source: {
      type: 'string',
      description: 'The exact name of the agent that identified this priority',
    },
  },
  required: ['title', 'description', 'importance', 'source'],
}

// Complete extraction function parameters
const extractionFunctionParameters = {
  type: 'object',
  properties: {
    priorities: {
      type: 'array',
      items: prioritySchema,
      description: 'The compiled list of priorities sorted by importance',
    },
  },
  required: ['priorities'],
}

Step 2: Set Up the Conversation with Extraction Function

Create a conversation that includes an inline function for data extraction. The key is providing a pre-canned result so the conversation continues without waiting for your application:

import { ChatBotKit } from '@chatbotkit/sdk'

const client = new ChatBotKit({
  secret: process.env.CHATBOTKIT_API_SECRET,
})

const SUBMIT_PRIORITIES_FUNCTION = 'submit_priorities'

const extractedPriorities = []

// Start conversation with extraction capability and stream results
for await (const { type, data } of client.conversation
  .complete(null, {
    backstory: `You are a priority orchestrator. Your job is to:
1. Gather priorities from all available agents
2. Analyze and consolidate the results
3. Submit the final list using the ${SUBMIT_PRIORITIES_FUNCTION} function`,

    messages: [
      {
        type: 'user',
        text: 'Please gather the top 5 priorities from all agents and compile them.',
      },
    ],

    // Define the extraction function
    functions: [
      {
        name: SUBMIT_PRIORITIES_FUNCTION,
        description: 'Submit the final compiled list of priorities. Call this function after gathering all priorities.',
        parameters: extractionFunctionParameters,

        // Pre-canned result: allows conversation to continue
        // The AI sees success, but we capture the actual data from the function call
        result: {
          data: { status: 'ok' },
        },
      },
    ],

    // Optional: provide abilities for calling sub-agents
    extensions: {
      skillsets: [
        {
          name: 'Agent Orchestration',
          description: 'Skills for calling sub-agents to gather information',
          abilities: [
            // Define sub-agent calling abilities here
          ],
        },
      ],
    },
  })
  .stream()) {

  // Monitor for function calls - this is where extraction happens!
  if (type === 'message' && data.type === 'activity') {
    const { meta } = data
    const activity = meta?.activity

    // Check if this is a request activity (function call)
    if (activity?.type === 'request') {
      const functionName = activity.function?.name
      const functionArgs = activity.function?.arguments

      // Capture extraction when our function is called
      if (functionName === SUBMIT_PRIORITIES_FUNCTION) {
        console.log('✓ Extraction function called!')

        // Parse and store extracted data
        const { priorities } = functionArgs
        extractedPriorities.push(...priorities)

        console.log(`Extracted ${priorities.length} priorities:`)
        priorities.forEach((priority, index) => {
          console.log(`  ${index + 1}. [${priority.importance}] ${priority.title}`)
          console.log(`     Source: ${priority.source}`)
        })
      }
    }
  }

  // Monitor other event types for visibility
  if (type === 'token') {
    process.stdout.write(data.token)
  }
}

console.log('\\nFinal extracted data:', extractedPriorities)

Step 3: Complete Example with Multiple Extraction Points

Here's a complete example showing multiple extraction stages:

import { ChatBotKit } from '@chatbotkit/sdk'

const client = new ChatBotKit({
  secret: process.env.CHATBOTKIT_API_SECRET,
})

async function performMultiStageExtraction() {
  const extractedData = {
    intermediateResults: [],
    finalResults: [],
  }

  // Start conversation with multiple extraction functions and stream results
  for await (const { type, data } of client.conversation
    .complete(null, {
      backstory: 'You are a data gathering orchestrator.',

      messages: [
        {
          type: 'user',
          text: 'Analyze the situation and provide structured insights.',
        },
      ],

      functions: [
        // Stage 1: Intermediate extraction
        {
          name: 'report_intermediate_findings',
          description: 'Report findings from a sub-task',
          parameters: {
            type: 'object',
            properties: {
              findings: {
                type: 'array',
                items: {
                  type: 'object',
                  properties: {
                    insight: { type: 'string' },
                    confidence: { type: 'number' },
                  },
                },
              },
            },
          },
          result: { data: { status: 'recorded' } },
        },

        // Stage 2: Final extraction
        {
          name: 'submit_final_analysis',
          description: 'Submit the complete analysis',
          parameters: {
            type: 'object',
            properties: {
              summary: { type: 'string' },
              recommendations: {
                type: 'array',
                items: { type: 'string' },
              },
              priority: {
                type: 'string',
                enum: ['low', 'medium', 'high'],
              },
            },
          },
          result: { data: { status: 'completed' } },
        },
      ],
    })
    .stream()) {

    // Monitor and extract at each stage
    if (type === 'message' && data.type === 'activity') {
      const { meta } = data
      const activity = meta?.activity

      // Check if this is a request activity (function call)
      if (activity?.type === 'request') {
        const functionName = activity.function?.name
        const functionArgs = activity.function?.arguments

        // Stage 1 extraction
        if (functionName === 'report_intermediate_findings') {
          console.log('✓ Stage 1: Intermediate extraction')
          extractedData.intermediateResults.push(functionArgs.findings)
        }

        // Stage 2 extraction
        if (functionName === 'submit_final_analysis') {
          console.log('✓ Stage 2: Final extraction')
          extractedData.finalResults.push({
            summary: functionArgs.summary,
            recommendations: functionArgs.recommendations,
            priority: functionArgs.priority,
          })
        }
      }
    }

    // Display streaming tokens
    if (type === 'token') {
      process.stdout.write(data.token)
    }
  }

  return extractedData
}

// Run the extraction
performMultiStageExtraction()
  .then(data => {
    console.log('\\n=== Multi-Stage Extraction Complete ===')
    console.log('Intermediate results:', data.intermediateResults.length)
    console.log('Final results:', data.finalResults.length)
    console.log(JSON.stringify(data, null, 2))
  })
  .catch(console.error)

Key Advantages of Multi-Stage Extraction

1. Progressive Data Capture

Extract data as the conversation unfolds, not just at the end:

// Extract from multiple sub-agents in parallel
Agent 1 → Extract priorities → Store
Agent 2 → Extract priorities → Store
Agent 3 → Extract priorities → Store
Orchestrator → Compile → Final extraction

2. Better User Experience

Show extraction progress to users in real-time:

if (event.type === 'function') {
  updateUI(`Processing: ${event.data.name}`)
  const extracted = event.data.args
  displayExtractedData(extracted)
}

3. Flexible Data Structures

Define different schemas for different extraction stages:

functions: [
  { name: 'extract_user_info', parameters: userSchema },
  { name: 'extract_preferences', parameters: preferencesSchema },
  { name: 'extract_final_request', parameters: requestSchema },
]

4. Error Recovery

Handle failures at specific stages without losing all data:

try {
  for await (const event of subscription.stream()) {
    if (event.type === 'function') {
      saveExtractionToDatabase(event.data.args)
    }
  }
} catch (error) {
  // You still have previously extracted data
  console.log('Extracted before error:', extractedItems)
}

Common Use Cases

1. Multi-Agent Systems

Extract results from each agent as they complete their tasks:

  • Sales agent extracts customer info
  • Support agent extracts issue details
  • Product agent extracts feature requests

2. Form Filling

Extract form fields progressively during a conversation:

  • Stage 1: Extract name and email
  • Stage 2: Extract preferences
  • Stage 3: Extract final submission data

3. Data Aggregation

Compile data from multiple sources:

  • Query multiple databases/APIs via agents
  • Extract results from each source
  • Aggregate into final report

4. Quality Assurance

Validate extracted data at each stage:

  • Extract data
  • Validate completeness
  • Request clarification if needed
  • Final extraction with validated data

Troubleshooting

Function Not Being Called

Problem: The AI doesn't call your extraction function.

Solution: Make the function description more explicit:

{
  name: 'submit_data',
  description: 'IMPORTANT: You MUST call this function to submit your findings. Call it when you have completed the analysis.',
  // ...
}

Missing Data in Extraction

Problem: Some fields are undefined in the extracted data.

Solution: Make fields required and add clear descriptions:

parameters: {
  type: 'object',
  properties: {
    title: {
      type: 'string',
      description: 'Required: A clear, concise title (cannot be empty)',
    },
  },
  required: ['title'], // Enforce required fields
}

Pre-canned Results Not Working

Problem: Conversation hangs waiting for function result.

Solution: Always provide a result field:

{
  name: 'extract_data',
  // ...
  result: {
    data: { status: 'ok' }, // AI sees this result immediately
  },
}

Conclusion

Multi-stage data extraction with ChatBotKit's complete API offers significant advantages over traditional end-of-conversation extraction:

  • Real-time extraction as the conversation progresses
  • Multiple extraction points for complex workflows
  • Better visibility into the extraction process
  • Flexible schemas for different stages

By using inline functions with pre-canned results and monitoring function calls through event streaming, you can build sophisticated data extraction pipelines that capture structured data progressively, enabling more robust and interactive AI applications.

extractdataAPIfunctions