Skip to content

neurotrace.core.schema

neurotrace.core.schema

Schema definitions for neurotrace core components.

This module defines the data structures used for managing neuromorphic memory components including messages, metadata, and emotion tags using Pydantic models.

Note

All models inherit from Pydantic BaseModel for data validation.

EmotionTag

Bases: BaseModel

Represents the emotional context and intensity of a message.

Attributes:

Name Type Description
sentiment Optional[Literal['positive', 'neutral', 'negative']]

The emotional tone of the message. Defaults to None.
intensity Optional[float]

A value indicating the strength of the emotion. Defaults to None.
Source code in neurotrace/core/schema.py 
24
25
26
27
28
29
30
31
32
33
34
35
class EmotionTag(BaseModel):
    """Represents the emotional context and intensity of a message.

    Attributes:
        sentiment (Optional[Literal["positive", "neutral", "negative"]]): The emotional
            tone of the message. Defaults to None.
        intensity (Optional[float]): A value indicating the strength of the emotion.
            Defaults to None.
    """

    sentiment: Optional[Literal["positive", "neutral", "negative"]] = None
    intensity: Optional[float] = None

Message

Bases: BaseModel

Message represents a single communication in the system. It includes the sender's role, content, timestamp, and metadata. Each message has a unique identifier generated as a UUID.

Example Representation: { "id": "", # unique message ID "role": "user" | "ai" | "system", # sender role "content": "string", # message text "timestamp": "ISO 8601", # message time (UTC) "metadata": { "token_count": 32, # optional, for budgeting/compression "embedding": [...], # vector representation (optional in-memory or precomputed) "source": "chat" | "web" | "api", # source of message "tags": ["finance", "personal"], # custom tags for search "thread_id": "conversation_XYZ", # optional thread/conversation tracking "user_id": "abc123", # to associate memory across sessions "related_ids": ["msg_id_1", "msg_id_2"],# links to other related messages (graph edge) "emotions": {"sentiment": "positive", "intensity": 0.85}, # optional emotion tagging "compressed": False # for summarization/compression tracking "session_id": "default" # session identifier for context } }

Source code in neurotrace/core/schema.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class Message(BaseModel):
    """
    Message represents a single communication in the system.
    It includes the sender's role, content, timestamp, and metadata.
    Each message has a unique identifier generated as a UUID.

    Example Representation:
    {
        "id": "<uuid4 or hash>",                    # unique message ID
        "role": "user" | "ai" | "system",           # sender role
        "content": "string",                        # message text
        "timestamp": "ISO 8601",                    # message time (UTC)
        "metadata": {
            "token_count": 32,                      # optional, for budgeting/compression
            "embedding": [...],                     # vector representation (optional in-memory or precomputed)
            "source": "chat" | "web" | "api",       # source of message
            "tags": ["finance", "personal"],        # custom tags for search
            "thread_id": "conversation_XYZ",        # optional thread/conversation tracking
            "user_id": "abc123",                    # to associate memory across sessions
            "related_ids": ["msg_id_1", "msg_id_2"],# links to other related messages (graph edge)
            "emotions": {"sentiment": "positive", "intensity": 0.85},  # optional emotion tagging
            "compressed": False                     # for summarization/compression tracking
            "session_id": "default"                # session identifier for context
        }
    }
    """

    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    role: str
    content: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
    metadata: MessageMetadata = Field(default_factory=MessageMetadata)

    def estimated_token_length(self) -> int:
        """Estimates the number of tokens in the message content.

        Returns:
            int: The estimated token count, either from metadata or word-based count.

        Note:
            Currently uses a simple word-splitting approach if token_count is not set
            in metadata. TODO: Implement a more accurate token counting method.
        """
        # todo: Implement a more accurate token counting method
        return self.metadata.token_count or len(self.content.split())

    def to_langchain_message(self) -> Union[HumanMessage, AIMessage]:
        """Converts this Message to a LangChain compatible format.

        Returns:
            Union[HumanMessage, AIMessage]: A LangChain message object based on the role.

        Raises:
            ValueError: If the role is neither 'human' nor 'ai'.
        """
        if Role.from_string(self.role) is Role.HUMAN:
            return self.to_human_message()
        elif Role.from_string(self.role) is Role.AI:
            return self.to_ai_message()
        else:
            raise ValueError(f"Unsupported role: {self.role}. Use 'human' or 'ai'.")

    def to_human_message(self) -> HumanMessage:
        """Converts this Message to a LangChain HumanMessage format.

        Returns:
            HumanMessage: A LangChain HumanMessage with the message content and metadata.
        """
        return HumanMessage(
            id=self.id, content=self.content, additional_kwargs={"id": self.id, "metadata": self.metadata.model_dump()}
        )

    def to_ai_message(self) -> AIMessage:
        """Converts this Message to a LangChain AIMessage format.

        Returns:
            AIMessage: A LangChain AIMessage with the message content and metadata.
        """
        return AIMessage(
            id=self.id, content=self.content, additional_kwargs={"id": self.id, "metadata": self.metadata.model_dump()}
        )

    def to_document(self) -> Document:
        """Convert Message to LangChain-compatible Document with safe metadata.

        Converts the current Message instance to a LangChain Document format,
        ensuring that the metadata is properly serialized and complex types
        are filtered out.

        Returns:
            Document: A LangChain Document instance containing the message content
                and filtered metadata.
        """
        raw_metadata = self.metadata.model_dump() if self.metadata else {}
        doc = Document(page_content=self.content, metadata={"id": self.id, "role": self.role, **raw_metadata})
        doc = filter_complex_metadata([doc])  # Remove lists, dicts, etc.

        return doc[0]

    @staticmethod
    def from_document(doc: Document) -> "Message":
        """Creates a Message instance from a LangChain Document.

        Extracts content and metadata from a LangChain Document to create
        a new Message instance. The role is extracted from metadata with
        a fallback to HUMAN if not specified.

        Args:
            doc (Document): The LangChain Document to convert from.

        Returns:
            Message: A new Message instance containing the document's content
                and metadata.
        """
        metadata = doc.metadata or {}
        role_str = metadata.pop("role", Role.HUMAN.value)  # fallback to human
        return Message(
            role=Role.from_string(role_str),
            content=doc.page_content,
            metadata=MessageMetadata(**metadata) if metadata else None,
            id=metadata.get("id"),
        )

    def __eq__(self, other):
        """Checks equality between two Message objects.

        Compares two Message instances for equality based on their role,
        content, and metadata. The id field is intentionally excluded from
        the comparison.

        Args:
            other: Another object to compare with.

        Returns:
            bool: True if the messages have the same role, content, and metadata
                (excluding id), False otherwise.
        """
        if not isinstance(other, Message):
            return False

        # not comparing id
        return self.role == other.role and self.content == other.content and self.metadata == other.metadata

    def __repr__(self):
        """String representation of the Message object.

        Returns:
            str: A string representation of the Message, including role, content,
                and metadata.
        """
        return f"[{self.timestamp.date()}] ({self.role}): {self.content}"

    def __str__(self):
        """String representation of the Message object.

        Returns:
            str: A string representation of the Message, including role, content,
                and metadata.
        """
        return f"[{self.timestamp.isoformat()}] ({self.role}): {self.content}"

__eq__(other)

Checks equality between two Message objects.

Compares two Message instances for equality based on their role, content, and metadata. The id field is intentionally excluded from the comparison.

Parameters:

Name Type Description Default
other

Another object to compare with.

 required

Returns:

Name Type Description
bool

True if the messages have the same role, content, and metadata (excluding id), False otherwise.
Source code in neurotrace/core/schema.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def __eq__(self, other):
    """Checks equality between two Message objects.

    Compares two Message instances for equality based on their role,
    content, and metadata. The id field is intentionally excluded from
    the comparison.

    Args:
        other: Another object to compare with.

    Returns:
        bool: True if the messages have the same role, content, and metadata
            (excluding id), False otherwise.
    """
    if not isinstance(other, Message):
        return False

    # not comparing id
    return self.role == other.role and self.content == other.content and self.metadata == other.metadata

__repr__()

String representation of the Message object.

Returns:

Name Type Description
str

A string representation of the Message, including role, content, and metadata.
Source code in neurotrace/core/schema.py 
211
212
213
214
215
216
217
218
def __repr__(self):
    """String representation of the Message object.

    Returns:
        str: A string representation of the Message, including role, content,
            and metadata.
    """
    return f"[{self.timestamp.date()}] ({self.role}): {self.content}"

__str__()

String representation of the Message object.

Returns:

Name Type Description
str

A string representation of the Message, including role, content, and metadata.
Source code in neurotrace/core/schema.py 
220
221
222
223
224
225
226
227
def __str__(self):
    """String representation of the Message object.

    Returns:
        str: A string representation of the Message, including role, content,
            and metadata.
    """
    return f"[{self.timestamp.isoformat()}] ({self.role}): {self.content}"

estimated_token_length()

Estimates the number of tokens in the message content.

Returns:

Name Type Description
int int

The estimated token count, either from metadata or word-based count.
Note

Currently uses a simple word-splitting approach if token_count is not set in metadata. TODO: Implement a more accurate token counting method.

Source code in neurotrace/core/schema.py 
101
102
103
104
105
106
107
108
109
110
111
112
def estimated_token_length(self) -> int:
    """Estimates the number of tokens in the message content.

    Returns:
        int: The estimated token count, either from metadata or word-based count.

    Note:
        Currently uses a simple word-splitting approach if token_count is not set
        in metadata. TODO: Implement a more accurate token counting method.
    """
    # todo: Implement a more accurate token counting method
    return self.metadata.token_count or len(self.content.split())

from_document(doc) staticmethod

Creates a Message instance from a LangChain Document.

Extracts content and metadata from a LangChain Document to create a new Message instance. The role is extracted from metadata with a fallback to HUMAN if not specified.

Parameters:

Name Type Description Default
doc Document

The LangChain Document to convert from.

 required

Returns:

Name Type Description
Message Message

A new Message instance containing the document's content and metadata.
Source code in neurotrace/core/schema.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@staticmethod
def from_document(doc: Document) -> "Message":
    """Creates a Message instance from a LangChain Document.

    Extracts content and metadata from a LangChain Document to create
    a new Message instance. The role is extracted from metadata with
    a fallback to HUMAN if not specified.

    Args:
        doc (Document): The LangChain Document to convert from.

    Returns:
        Message: A new Message instance containing the document's content
            and metadata.
    """
    metadata = doc.metadata or {}
    role_str = metadata.pop("role", Role.HUMAN.value)  # fallback to human
    return Message(
        role=Role.from_string(role_str),
        content=doc.page_content,
        metadata=MessageMetadata(**metadata) if metadata else None,
        id=metadata.get("id"),
    )

to_ai_message()

Converts this Message to a LangChain AIMessage format.

Returns:

Name Type Description
AIMessage AIMessage

A LangChain AIMessage with the message content and metadata.
Source code in neurotrace/core/schema.py 
140
141
142
143
144
145
146
147
148
def to_ai_message(self) -> AIMessage:
    """Converts this Message to a LangChain AIMessage format.

    Returns:
        AIMessage: A LangChain AIMessage with the message content and metadata.
    """
    return AIMessage(
        id=self.id, content=self.content, additional_kwargs={"id": self.id, "metadata": self.metadata.model_dump()}
    )

to_document()

Convert Message to LangChain-compatible Document with safe metadata.

Converts the current Message instance to a LangChain Document format, ensuring that the metadata is properly serialized and complex types are filtered out.

Returns:

Name Type Description
Document Document

A LangChain Document instance containing the message content and filtered metadata.
Source code in neurotrace/core/schema.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def to_document(self) -> Document:
    """Convert Message to LangChain-compatible Document with safe metadata.

    Converts the current Message instance to a LangChain Document format,
    ensuring that the metadata is properly serialized and complex types
    are filtered out.

    Returns:
        Document: A LangChain Document instance containing the message content
            and filtered metadata.
    """
    raw_metadata = self.metadata.model_dump() if self.metadata else {}
    doc = Document(page_content=self.content, metadata={"id": self.id, "role": self.role, **raw_metadata})
    doc = filter_complex_metadata([doc])  # Remove lists, dicts, etc.

    return doc[0]

to_human_message()

Converts this Message to a LangChain HumanMessage format.

Returns:

Name Type Description
HumanMessage HumanMessage

A LangChain HumanMessage with the message content and metadata.
Source code in neurotrace/core/schema.py 
130
131
132
133
134
135
136
137
138
def to_human_message(self) -> HumanMessage:
    """Converts this Message to a LangChain HumanMessage format.

    Returns:
        HumanMessage: A LangChain HumanMessage with the message content and metadata.
    """
    return HumanMessage(
        id=self.id, content=self.content, additional_kwargs={"id": self.id, "metadata": self.metadata.model_dump()}
    )

to_langchain_message()

Converts this Message to a LangChain compatible format.

Returns:

Type Description
Union[HumanMessage, AIMessage]

Union[HumanMessage, AIMessage]: A LangChain message object based on the role.

Raises:

Type Description
ValueError

If the role is neither 'human' nor 'ai'.
Source code in neurotrace/core/schema.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def to_langchain_message(self) -> Union[HumanMessage, AIMessage]:
    """Converts this Message to a LangChain compatible format.

    Returns:
        Union[HumanMessage, AIMessage]: A LangChain message object based on the role.

    Raises:
        ValueError: If the role is neither 'human' nor 'ai'.
    """
    if Role.from_string(self.role) is Role.HUMAN:
        return self.to_human_message()
    elif Role.from_string(self.role) is Role.AI:
        return self.to_ai_message()
    else:
        raise ValueError(f"Unsupported role: {self.role}. Use 'human' or 'ai'.")

MessageMetadata

Bases: BaseModel

Contains additional contextual information about a message.

Attributes:

Name Type Description
token_count Optional[int]

Number of tokens in the associated message.
embedding Optional[List[float]]

Vector representation of the message content.
source Optional[Literal['chat', 'web', 'api', 'system']]

Origin of the message. Defaults to "chat".
tags Optional[List[str]]

List of categorical tags. Defaults to empty list.
thread_id Optional[str]

Unique identifier for the conversation thread.
user_id Optional[str]

Identifier for the message author.
related_ids Optional[List[str]]

References to related message IDs.
emotions Optional[EmotionTag]

Emotional analysis of the message.
compressed Optional[bool]

Whether the message content is compressed. Defaults to False.
session_id Optional[str]

Current session identifier. Defaults to 'default'.
Source code in neurotrace/core/schema.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class MessageMetadata(BaseModel):
    """Contains additional contextual information about a message.

    Attributes:
        token_count (Optional[int]): Number of tokens in the associated message.
        embedding (Optional[List[float]]): Vector representation of the message content.
        source (Optional[Literal["chat", "web", "api", "system"]]): Origin of the message.
            Defaults to "chat".
        tags (Optional[List[str]]): List of categorical tags. Defaults to empty list.
        thread_id (Optional[str]): Unique identifier for the conversation thread.
        user_id (Optional[str]): Identifier for the message author.
        related_ids (Optional[List[str]]): References to related message IDs.
        emotions (Optional[EmotionTag]): Emotional analysis of the message.
        compressed (Optional[bool]): Whether the message content is compressed.
            Defaults to False.
        session_id (Optional[str]): Current session identifier. Defaults to 'default'.
    """

    token_count: Optional[int] = None
    embedding: Optional[List[float]] = None
    source: Optional[Literal["chat", "web", "api", "system"]] = "chat"
    tags: Optional[List[str]] = []
    thread_id: Optional[str] = None
    user_id: Optional[str] = None
    related_ids: Optional[List[str]] = []
    emotions: Optional[EmotionTag] = None
    compressed: Optional[bool] = False
    session_id: Optional[str] = "default"