messages

Class info

Classes

Name	Children	Inherits
AgentResponse llmling_agent.messaging.messages Result from an agent's execution.		Generic
ChatMessage llmling_agent.messaging.messages Common message format for all UI types.		Generic
TeamResponse llmling_agent.messaging.messages Results from a team execution.		list Generic
TokenCost llmling_agent.messaging.messages Combined token and cost tracking.		object
TokenUsage llmling_agent.messaging.messages Token usage statistics from model responses.		dict
ToolCallInfo llmling_agent.models.tools Information about an executed tool call.		BaseModel

🛈 DocStrings

Message and token usage models.

AgentResponse dataclass

Result from an agent's execution.

Source code in src/llmling_agent/messaging/messages.py

@dataclass
class AgentResponse[TResult]:
    """Result from an agent's execution."""

    agent_name: str
    """Name of the agent that produced this result"""

    message: ChatMessage[TResult] | None
    """The actual message with content and metadata"""

    timing: float | None = None
    """Time taken by this agent in seconds"""

    error: str | None = None
    """Error message if agent failed"""

    @property
    def success(self) -> bool:
        """Whether the agent completed successfully."""
        return self.error is None

    @property
    def response(self) -> TResult | None:
        """Convenient access to message content."""
        return self.message.content if self.message else None

agent_name instance-attribute

agent_name: str

Name of the agent that produced this result

error class-attribute instance-attribute

error: str | None = None

Error message if agent failed

message instance-attribute

message: ChatMessage[TResult] | None

The actual message with content and metadata

response property

response: TResult | None

Convenient access to message content.

success property

success: bool

Whether the agent completed successfully.

timing class-attribute instance-attribute

timing: float | None = None

Time taken by this agent in seconds

ChatMessage dataclass

Common message format for all UI types.

Generically typed with: ChatMessage[Type of Content] The type can either be str or a BaseModel subclass.

Source code in src/llmling_agent/messaging/messages.py

@dataclass
class ChatMessage[TContent]:
    """Common message format for all UI types.

    Generically typed with: ChatMessage[Type of Content]
    The type can either be str or a BaseModel subclass.
    """

    content: TContent
    """Message content, typed as TContent (either str or BaseModel)."""

    role: MessageRole
    """Role of the message sender (user/assistant/system)."""

    model: str | None = None
    """Name of the model that generated this message."""

    metadata: JsonObject = field(default_factory=dict)
    """Additional metadata about the message."""

    timestamp: datetime = field(default_factory=datetime.now)
    """When this message was created."""

    cost_info: TokenCost | None = None
    """Token usage and costs for this specific message if available."""

    message_id: str = field(default_factory=lambda: str(uuid4()))
    """Unique identifier for this message."""

    conversation_id: str | None = None
    """ID of the conversation this message belongs to."""

    response_time: float | None = None
    """Time it took the LLM to respond."""

    tool_calls: list[ToolCallInfo] = field(default_factory=list)
    """List of tool calls made during message generation."""

    associated_messages: list[ChatMessage[Any]] = field(default_factory=list)
    """List of messages which were generated during the the creation of this messsage."""

    name: str | None = None
    """Display name for the message sender in UI."""

    forwarded_from: list[str] = field(default_factory=list)
    """List of agent names (the chain) that forwarded this message to the sender."""

    provider_extra: dict[str, Any] = field(default_factory=dict)
    """Provider specific metadata / extra information."""

    def forwarded(self, previous_message: ChatMessage[Any]) -> Self:
        """Create new message showing it was forwarded from another message.

        Args:
            previous_message: The message that led to this one's creation

        Returns:
            New message with updated chain showing the path through previous message
        """
        from_ = [*previous_message.forwarded_from, previous_message.name or "unknown"]
        return replace(self, forwarded_from=from_)

    def to_text_message(self) -> ChatMessage[str]:
        """Convert this message to a text-only version."""
        return dataclasses.replace(self, content=str(self.content))  # type: ignore

    def _get_content_str(self) -> str:
        """Get string representation of content."""
        match self.content:
            case str():
                return self.content
            case BaseModel():
                return self.content.model_dump_json(indent=2)
            case _:
                msg = f"Unexpected content type: {type(self.content)}"
                raise ValueError(msg)

    @property
    def data(self) -> TContent:
        """Get content as typed data. Provides compat to RunResult."""
        return self.content

    def format(
        self,
        style: FormatStyle = "simple",
        *,
        template: str | None = None,
        variables: dict[str, Any] | None = None,
        show_metadata: bool = False,
        show_costs: bool = False,
    ) -> str:
        """Format message with configurable style.

        Args:
            style: Predefined style or "custom" for custom template
            template: Custom Jinja template (required if style="custom")
            variables: Additional variables for template rendering
            show_metadata: Whether to include metadata
            show_costs: Whether to include cost information

        Raises:
            ValueError: If style is "custom" but no template provided
                    or if style is invalid
        """
        from jinja2 import Environment
        import yamling

        env = Environment(trim_blocks=True, lstrip_blocks=True)
        env.filters["to_yaml"] = yamling.dump_yaml

        match style:
            case "custom":
                if not template:
                    msg = "Custom style requires a template"
                    raise ValueError(msg)
                template_str = template
            case _ if style in MESSAGE_TEMPLATES:
                template_str = MESSAGE_TEMPLATES[style]
            case _:
                msg = f"Invalid style: {style}"
                raise ValueError(msg)

        template_obj = env.from_string(template_str)
        vars_ = {**asdict(self), "show_metadata": show_metadata, "show_costs": show_costs}
        if variables:
            vars_.update(variables)

        return template_obj.render(**vars_)

associated_messages class-attribute instance-attribute

associated_messages: list[ChatMessage[Any]] = field(default_factory=list)

List of messages which were generated during the the creation of this messsage.

content instance-attribute

content: TContent

Message content, typed as TContent (either str or BaseModel).

conversation_id class-attribute instance-attribute

conversation_id: str | None = None

ID of the conversation this message belongs to.

cost_info class-attribute instance-attribute

cost_info: TokenCost | None = None

Token usage and costs for this specific message if available.

data property

data: TContent

Get content as typed data. Provides compat to RunResult.

forwarded_from class-attribute instance-attribute

forwarded_from: list[str] = field(default_factory=list)

List of agent names (the chain) that forwarded this message to the sender.

message_id class-attribute instance-attribute

message_id: str = field(default_factory=lambda: str(uuid4()))

Unique identifier for this message.

metadata class-attribute instance-attribute

metadata: JsonObject = field(default_factory=dict)

Additional metadata about the message.

model class-attribute instance-attribute

model: str | None = None

Name of the model that generated this message.

name class-attribute instance-attribute

name: str | None = None

Display name for the message sender in UI.

provider_extra class-attribute instance-attribute

provider_extra: dict[str, Any] = field(default_factory=dict)

Provider specific metadata / extra information.

response_time class-attribute instance-attribute

response_time: float | None = None

Time it took the LLM to respond.

role instance-attribute

role: MessageRole

Role of the message sender (user/assistant/system).

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now)

When this message was created.

tool_calls class-attribute instance-attribute

tool_calls: list[ToolCallInfo] = field(default_factory=list)

List of tool calls made during message generation.

_get_content_str

_get_content_str() -> str

Get string representation of content.

Source code in src/llmling_agent/messaging/messages.py

def _get_content_str(self) -> str:
    """Get string representation of content."""
    match self.content:
        case str():
            return self.content
        case BaseModel():
            return self.content.model_dump_json(indent=2)
        case _:
            msg = f"Unexpected content type: {type(self.content)}"
            raise ValueError(msg)

format

format(
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_metadata: bool = False,
    show_costs: bool = False,
) -> str

Format message with configurable style.

Parameters:

Name	Type	Description	Default
style	FormatStyle	Predefined style or "custom" for custom template	'simple'
template	str | None	Custom Jinja template (required if style="custom")	None
variables	dict[str, Any] | None	Additional variables for template rendering	None
show_metadata	bool	Whether to include metadata	False
show_costs	bool	Whether to include cost information	False

Raises:

Type	Description
ValueError	If style is "custom" but no template provided or if style is invalid

Source code in src/llmling_agent/messaging/messages.py

def format(
    self,
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_metadata: bool = False,
    show_costs: bool = False,
) -> str:
    """Format message with configurable style.

    Args:
        style: Predefined style or "custom" for custom template
        template: Custom Jinja template (required if style="custom")
        variables: Additional variables for template rendering
        show_metadata: Whether to include metadata
        show_costs: Whether to include cost information

    Raises:
        ValueError: If style is "custom" but no template provided
                or if style is invalid
    """
    from jinja2 import Environment
    import yamling

    env = Environment(trim_blocks=True, lstrip_blocks=True)
    env.filters["to_yaml"] = yamling.dump_yaml

    match style:
        case "custom":
            if not template:
                msg = "Custom style requires a template"
                raise ValueError(msg)
            template_str = template
        case _ if style in MESSAGE_TEMPLATES:
            template_str = MESSAGE_TEMPLATES[style]
        case _:
            msg = f"Invalid style: {style}"
            raise ValueError(msg)

    template_obj = env.from_string(template_str)
    vars_ = {**asdict(self), "show_metadata": show_metadata, "show_costs": show_costs}
    if variables:
        vars_.update(variables)

    return template_obj.render(**vars_)

forwarded

forwarded(previous_message: ChatMessage[Any]) -> Self

Create new message showing it was forwarded from another message.

Parameters:

Name	Type	Description	Default
previous_message	ChatMessage[Any]	The message that led to this one's creation	required

Returns:

Type	Description
Self	New message with updated chain showing the path through previous message

Source code in src/llmling_agent/messaging/messages.py

def forwarded(self, previous_message: ChatMessage[Any]) -> Self:
    """Create new message showing it was forwarded from another message.

    Args:
        previous_message: The message that led to this one's creation

    Returns:
        New message with updated chain showing the path through previous message
    """
    from_ = [*previous_message.forwarded_from, previous_message.name or "unknown"]
    return replace(self, forwarded_from=from_)

to_text_message

to_text_message() -> ChatMessage[str]

Convert this message to a text-only version.

Source code in src/llmling_agent/messaging/messages.py

def to_text_message(self) -> ChatMessage[str]:
    """Convert this message to a text-only version."""
    return dataclasses.replace(self, content=str(self.content))  # type: ignore

TeamResponse

Bases: list[AgentResponse[Any]]

Results from a team execution.

Source code in src/llmling_agent/messaging/messages.py

class TeamResponse[TMessageContent](list[AgentResponse[Any]]):
    """Results from a team execution."""

    def __init__(
        self,
        responses: list[AgentResponse[TMessageContent]],
        start_time: datetime | None = None,
        errors: dict[str, Exception] | None = None,
    ):
        super().__init__(responses)
        self.start_time = start_time or datetime.now()
        self.end_time = datetime.now()
        self.errors = errors or {}

    @property
    def duration(self) -> float:
        """Get execution duration in seconds."""
        return (self.end_time - self.start_time).total_seconds()

    @property
    def success(self) -> bool:
        """Whether all agents completed successfully."""
        return not bool(self.errors)

    @property
    def failed_agents(self) -> list[str]:
        """Names of agents that failed."""
        return list(self.errors.keys())

    def by_agent(self, name: str) -> AgentResponse[TMessageContent] | None:
        """Get response from specific agent."""
        return next((r for r in self if r.agent_name == name), None)

    def format_durations(self) -> str:
        """Format execution times."""
        parts = [f"{r.agent_name}: {r.timing:.2f}s" for r in self if r.timing is not None]
        return f"Individual times: {', '.join(parts)}\nTotal time: {self.duration:.2f}s"

    # TODO: could keep TResultContent for len(messages) == 1
    def to_chat_message(self) -> ChatMessage[str]:
        """Convert team response to a single chat message."""
        # Combine all responses into one structured message
        content = "\n\n".join(
            f"[{response.agent_name}]: {response.message.content}"
            for response in self
            if response.message
        )
        meta = {
            "type": "team_response",
            "agents": [r.agent_name for r in self],
            "duration": self.duration,
            "success_count": len(self),
        }
        return ChatMessage(content=content, role="assistant", metadata=meta)  # type: ignore

duration property

duration: float

Get execution duration in seconds.

failed_agents property

failed_agents: list[str]

Names of agents that failed.

success property

success: bool

Whether all agents completed successfully.

by_agent

by_agent(name: str) -> AgentResponse[TMessageContent] | None

Get response from specific agent.

Source code in src/llmling_agent/messaging/messages.py

def by_agent(self, name: str) -> AgentResponse[TMessageContent] | None:
    """Get response from specific agent."""
    return next((r for r in self if r.agent_name == name), None)

format_durations

format_durations() -> str

Format execution times.

Source code in src/llmling_agent/messaging/messages.py

def format_durations(self) -> str:
    """Format execution times."""
    parts = [f"{r.agent_name}: {r.timing:.2f}s" for r in self if r.timing is not None]
    return f"Individual times: {', '.join(parts)}\nTotal time: {self.duration:.2f}s"

to_chat_message

to_chat_message() -> ChatMessage[str]

Convert team response to a single chat message.

Source code in src/llmling_agent/messaging/messages.py

def to_chat_message(self) -> ChatMessage[str]:
    """Convert team response to a single chat message."""
    # Combine all responses into one structured message
    content = "\n\n".join(
        f"[{response.agent_name}]: {response.message.content}"
        for response in self
        if response.message
    )
    meta = {
        "type": "team_response",
        "agents": [r.agent_name for r in self],
        "duration": self.duration,
        "success_count": len(self),
    }
    return ChatMessage(content=content, role="assistant", metadata=meta)  # type: ignore

TokenCost dataclass

Combined token and cost tracking.

Source code in src/llmling_agent/messaging/messages.py

@dataclass(frozen=True)
class TokenCost:
    """Combined token and cost tracking."""

    token_usage: TokenUsage
    """Token counts for prompt and completion"""
    total_cost: float
    """Total cost in USD"""

    @classmethod
    async def from_usage(
        cls,
        usage: tokonomics.Usage | None,
        model: str,
        prompt: str,
        completion: str,
    ) -> TokenCost | None:
        """Create result from usage data.

        Args:
            usage: Token counts from model response
            model: Name of the model used
            prompt: The prompt text sent to model
            completion: The completion text received

        Returns:
            TokenCost if usage data available, None otherwise
        """
        import tokonomics

        if not (
            usage
            and usage.total_tokens is not None
            and usage.request_tokens is not None
            and usage.response_tokens is not None
        ):
            logger.debug("Missing token counts in Usage object")
            return None

        token_usage = TokenUsage(
            total=usage.total_tokens,
            prompt=usage.request_tokens,
            completion=usage.response_tokens,
        )
        logger.debug("Token usage: %s", token_usage)

        cost = await tokonomics.calculate_token_cost(
            model,
            usage.request_tokens,
            usage.response_tokens,
        )
        total_cost = cost.total_cost if cost else 0.0

        return cls(token_usage=token_usage, total_cost=total_cost)

token_usage instance-attribute

token_usage: TokenUsage

Token counts for prompt and completion

total_cost instance-attribute

total_cost: float

Total cost in USD

from_usage async classmethod

from_usage(
    usage: Usage | None, model: str, prompt: str, completion: str
) -> TokenCost | None

Create result from usage data.

Parameters:

Name	Type	Description	Default
usage	Usage | None	Token counts from model response	required
model	str	Name of the model used	required
prompt	str	The prompt text sent to model	required
completion	str	The completion text received	required

Returns:

Type	Description
TokenCost | None	TokenCost if usage data available, None otherwise

Source code in src/llmling_agent/messaging/messages.py

@classmethod
async def from_usage(
    cls,
    usage: tokonomics.Usage | None,
    model: str,
    prompt: str,
    completion: str,
) -> TokenCost | None:
    """Create result from usage data.

    Args:
        usage: Token counts from model response
        model: Name of the model used
        prompt: The prompt text sent to model
        completion: The completion text received

    Returns:
        TokenCost if usage data available, None otherwise
    """
    import tokonomics

    if not (
        usage
        and usage.total_tokens is not None
        and usage.request_tokens is not None
        and usage.response_tokens is not None
    ):
        logger.debug("Missing token counts in Usage object")
        return None

    token_usage = TokenUsage(
        total=usage.total_tokens,
        prompt=usage.request_tokens,
        completion=usage.response_tokens,
    )
    logger.debug("Token usage: %s", token_usage)

    cost = await tokonomics.calculate_token_cost(
        model,
        usage.request_tokens,
        usage.response_tokens,
    )
    total_cost = cost.total_cost if cost else 0.0

    return cls(token_usage=token_usage, total_cost=total_cost)

TokenUsage

Bases: TypedDict

Token usage statistics from model responses.

Source code in src/llmling_agent/messaging/messages.py

class TokenUsage(TypedDict):
    """Token usage statistics from model responses."""

    total: int
    """Total tokens used"""
    prompt: int
    """Tokens used in the prompt"""
    completion: int
    """Tokens used in the completion"""

completion instance-attribute

completion: int

Tokens used in the completion

prompt instance-attribute

prompt: int

Tokens used in the prompt

total instance-attribute

total: int

Total tokens used