Conversation Intelligence

tac.intelligence

Conversation Intelligence event processing module.

Webhook processing for Twilio Conversational Intelligence.

ConversationIntelligenceConfig

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.

from_env classmethod

from_env() -> ConversationIntelligenceConfig | None

Create ConversationIntelligenceConfig from CONVERSATION_INTELLIGENCE_* env vars.

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 observations or summaries in Conversation Memory based on the event type.

Events are filtered by: - Configuration ID matching the provided config - Operator SID matching observation or 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...",
    observation_operator_sid="LY...",
    summary_operator_sid="LY...",
)
processor = OperatorResultProcessor(conversation_memory_client, config)

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

Initialize the CI event processor.

Parameters:

Name	Type	Description	Default
conversation_memory_client	MemoryClient	MemoryClient instance for creating observations/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: extracts profile IDs, generates content 5. Creates observations or 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

Bases: BaseModel

Result of processing a Conversation Intelligence webhook event.