Overview TypeAgent uses a universal message system that supports transcripts, podcasts, chats, emails, forums, and other conversation types. All message classes implement the IMessage protocol.
ConversationMessage from typeagent.knowpro.universal_message import ConversationMessage
Universal message type for any conversation. This is the primary message class used in TypeAgent.
Class Definition @dataclass class ConversationMessage ( IMessage ): text_chunks: list[ str ] metadata: ConversationMessageMeta tags: list[ str ] = [] timestamp: str | None = None
Fields The text content of the message, split into one or more chunks. Most messages have a single chunk. Example: ["Hello, how can I help you today?"]
metadata
ConversationMessageMeta
required
Optional tags attached to this specific message. These are indexed and searchable. Example: ["question", "technical", "urgent"]
ISO 8601 datetime when the message occurred/was sent (UTC timezone). Must include ‘Z’ suffix to explicitly indicate UTC. Format: "2024-03-06T10:30:00Z"Special case: If no timestamp is provided during ingestion, the Unix epoch ("1970-01-01T00:00:00Z") is used as a sentinel value.
Methods get_knowledge def get_knowledge ( self ) -> KnowledgeResponse
Extract structured knowledge from the message metadata. Delegates to metadata.get_knowledge().
Structured knowledge including entities (speakers, recipients) and actions (communication).
add_timestamp def add_timestamp ( self , timestamp : str ) -> None
Set or update the message timestamp.
ISO 8601 timestamp with ‘Z’ suffix.
serialize / deserialize def serialize ( self ) -> ConversationMessageData @ staticmethod def deserialize(message_data: ConversationMessageData) -> ConversationMessage
Convert to/from JSON-serializable dictionary format for persistence.
Example Usage from typeagent.knowpro.universal_message import ( ConversationMessage, ConversationMessageMeta ) # Simple message msg = ConversationMessage( text_chunks = [ "Let's schedule a meeting for next Tuesday." ], metadata = ConversationMessageMeta( speaker = "Alice" , recipients = [ "Bob" , "Charlie" ] ), tags = [ "scheduling" , "meeting" ], timestamp = "2024-03-06T14:30:00Z" ) # Multi-chunk message (rare) multi_chunk = ConversationMessage( text_chunks = [ "First, we need to review the requirements." , "Then, we can start implementation." , "Finally, we'll write tests." ], metadata = ConversationMessageMeta( speaker = "Alice" ) )
from typeagent.knowpro.universal_message import ConversationMessageMeta
Metadata for conversation messages, supporting all conversation types.
Class Definition @dataclass class ConversationMessageMeta ( IKnowledgeSource , IMessageMetadata ): speaker: str | None = None recipients: list[ str ] = []
Fields The primary source/sender of the message. Interpretation varies by conversation type:
Transcript: Speaker name from WebVTT or detected patternPodcast: Participant speaking this turnEmail: Sender address or nameChat: Username or display nameForum: Post author Example: "Alice Smith" or "[email protected] "Intended recipients/listeners of the message. Also called “listeners” in serialization.
Transcript: Empty (broadcast medium)Podcast: Other participants in conversationEmail: To/CC recipientsChat: @-mentioned users or room membersForum: Empty (public) or @-mentioned users Example: ["[email protected] ", "[email protected] "]Properties Alias for speaker. Implements IMessageMetadata.source.
Returns recipients if non-empty, else None. Implements IMessageMetadata.dest.
Methods get_knowledge def get_knowledge ( self ) -> KnowledgeResponse
Extract structured knowledge from metadata:
Person entities for speaker and recipientsCommunication actions linking speaker to recipients
Show Knowledge Extraction Rules
With recipients:
Creates entity for speaker
Creates entity for each recipient
Creates “say” action from speaker to each recipient (past tense)
Without recipients:
Creates entity for speaker only
Creates “say/speak” action with no specific audience (past tense)
No speaker:
Returns empty knowledge (no entities or actions)
Example Output: meta = ConversationMessageMeta( speaker = "Alice" , recipients = [ "Bob" , "Charlie" ] ) knowledge = meta.get_knowledge() # Returns: # - entities: [Person("Alice"), Person("Bob"), Person("Charlie")] # - actions: [ # Action(verbs=["say"], subject="Alice", object="Bob"), # Action(verbs=["say"], subject="Alice", object="Charlie") # ]
Example Usage # Email-style message email_meta = ConversationMessageMeta( speaker = "[email protected] " , recipients = [ "[email protected] " , "[email protected] " ] ) # Transcript-style message (no recipients) transcript_meta = ConversationMessageMeta( speaker = "Alice Smith" ) # Chat with mentions chat_meta = ConversationMessageMeta( speaker = "alice_dev" , recipients = [ "@bob" , "@charlie" ] )
TranscriptMessage from typeagent.transcripts.transcript import TranscriptMessage
Alias for ConversationMessage. Use this when working with transcript-specific code for clarity.
TranscriptMessage = ConversationMessage
Example from typeagent.transcripts.transcript import TranscriptMessage, TranscriptMessageMeta # Semantically identical to ConversationMessage msg = TranscriptMessage( text_chunks = [ "Welcome to today's presentation." ], metadata = TranscriptMessageMeta( speaker = "Alice" ), timestamp = "2024-03-06T09:00:00Z" )
from typeagent.transcripts.transcript import TranscriptMessageMeta
Alias for ConversationMessageMeta. Use with TranscriptMessage for transcript processing.
TranscriptMessageMeta = ConversationMessageMeta
from typeagent.knowpro.universal_message import format_timestamp_utc def format_timestamp_utc ( dt : datetime) -> str
Format a datetime as ISO 8601 with explicit ‘Z’ suffix for UTC.
Datetime to format. Must be timezone-aware . Will be converted to UTC if needed.
ISO 8601 string with ‘Z’ suffix, e.g., "2024-01-01T12:34:56Z"
Raises: ValueError if datetime is timezone-naive.Example: from datetime import datetime, timezone from typeagent.knowpro.universal_message import format_timestamp_utc # Create timezone-aware datetime dt = datetime( 2024 , 3 , 6 , 14 , 30 , 0 , tzinfo = timezone.utc) timestamp = format_timestamp_utc(dt) print (timestamp) # "2024-03-06T14:30:00Z" # Convert from other timezone from datetime import timedelta est = timezone(timedelta( hours =- 5 )) dt_est = datetime( 2024 , 3 , 6 , 9 , 30 , 0 , tzinfo = est) timestamp = format_timestamp_utc(dt_est) print (timestamp) # "2024-03-06T14:30:00Z" (converted to UTC) # Error: naive datetime try : naive_dt = datetime( 2024 , 3 , 6 , 14 , 30 , 0 ) format_timestamp_utc(naive_dt) # Raises ValueError except ValueError as e: print ( f "Error: { e } " )
Complete Example from datetime import datetime, timezone from typeagent import create_conversation from typeagent.knowpro.universal_message import ( ConversationMessage, ConversationMessageMeta, format_timestamp_utc ) # Create conversation conv = await create_conversation( dbname = "team_chat.db" , message_type = ConversationMessage, name = "engineering_discussion" ) # Create messages with proper timestamps messages = [ ConversationMessage( text_chunks = [ "We need to refactor the authentication module." ], metadata = ConversationMessageMeta( speaker = "Alice" , recipients = [ "Bob" , "Charlie" ] ), tags = [ "refactoring" , "security" ], timestamp = format_timestamp_utc( datetime( 2024 , 3 , 6 , 10 , 0 , 0 , tzinfo = timezone.utc) ) ), ConversationMessage( text_chunks = [ "I agree. Let's use JWT tokens instead of sessions." ], metadata = ConversationMessageMeta( speaker = "Bob" , recipients = [ "Alice" ] ), tags = [ "security" , "architecture" ], timestamp = format_timestamp_utc( datetime( 2024 , 3 , 6 , 10 , 2 , 30 , tzinfo = timezone.utc) ) ) ] # Add to conversation result = await conv.add_messages_with_indexing(messages) print ( f "Processed { result.messages_added } messages" ) print ( f "Extracted { result.semrefs_added } knowledge items" ) # Query about the conversation answer = await conv.query( "What security improvements were discussed?" ) print (answer)
