Skip to content

Conversation Intelligence

tac.intelligence

Conversation Intelligence event processing module.

Webhook processing for Twilio Conversational Intelligence.

ConversationIntelligenceConfig pydantic-model

Bases: BaseModel

Configuration for Conversation Intelligence webhook filtering.

This config specifies which CI configuration and operators to process. Events that don't match are filtered out.

Config:

  • json_schema_extra: {'example': {'configuration_id': 'your_ci_configuration_id', 'summary_operator_sid': 'LY00000000000000000000000000000002'}}

Fields:

configuration_id pydantic-field 

configuration_id: str

Conversation Intelligence Configuration ID

summary_operator_sid pydantic-field 

summary_operator_sid: str | None = None

Operator SID for summary extraction (e.g., LY...)

from_env classmethod 

from_env() -> ConversationIntelligenceConfig | None

Create ConversationIntelligenceConfig from CONVERSATION_INTELLIGENCE_* env vars.

Blank or whitespace-only operator SIDs are treated as not configured.

OperatorResultProcessor 

OperatorResultProcessor(
    conversation_memory_client: MemoryClient,
    config: ConversationIntelligenceConfig,
)

Processor for Conversation Intelligence webhook events.

This processor handles incoming CI webhook payloads, validates them, and creates conversation summaries in Conversation Memory based on the event type.

Events are filtered by: - Configuration ID matching the provided config - Operator SID matching the summary operator SID in config

Example usage 
from tac.context.memory import MemoryClient
from tac.core.config import ConversationIntelligenceConfig
from tac.intelligence import OperatorResultProcessor

conversation_memory_client = MemoryClient(...)
config = ConversationIntelligenceConfig(
    configuration_id="GA...",
    summary_operator_sid="LY...",
)
processor = OperatorResultProcessor(conversation_memory_client, config)

result = await processor.process_event(webhook_payload)
if result.skipped:
    print(f"Skipped: {result.skip_reason}")
elif result.success:
    print(f"Created {result.created_count} {result.event_type}(s)")
else:
    print(f"Error: {result.error}")

Initialize the CI event processor.

Parameters:

Name Type Description Default
conversation_memory_client MemoryClient

MemoryClient instance for creating summaries

 required
config ConversationIntelligenceConfig

ConversationIntelligenceConfig for filtering events by configuration ID and operator SIDs

 required

process_event async 

process_event(
    payload: dict[str, Any],
) -> OperatorProcessingResult

Process a CI webhook payload.

This method: 1. Parses the payload into an OperatorResultEvent (Pydantic validates required fields) 2. Applies filtering logic based on intelligence configuration ID and operator SIDs 3. Iterates over operator_results array 4. For each operator result: filters by operator SID, then generates content and extracts profile IDs 5. Creates conversation summaries in Conversation Memory

Parameters:

Name Type Description Default
payload dict[str, Any]

The raw webhook payload dictionary

 required

Returns:

Type Description
OperatorProcessingResult

OperatorProcessingResult with status and details

OperatorProcessingResult pydantic-model

Bases: BaseModel

Result of processing a Conversation Intelligence webhook event.

Config:

  • default: {'populate_by_name': True}

Fields:

success pydantic-field 

success: bool

Whether processing completed successfully

event_type pydantic-field 

event_type: str | None = None

Type of event processed: 'observation', 'summary', or None if filtered/failed

skipped pydantic-field 

skipped: bool = False

True if event was filtered out (not an error)

skip_reason pydantic-field 

skip_reason: str | None = None

Reason for skipping (e.g., 'non-conversation-memory event')

error pydantic-field 

error: str | None = None

Error message if processing failed

created_count pydantic-field 

created_count: int = 0

Number of observations/summaries created