import enum
import logging
import pydantic
import textwrap
import typing
logger = logging.getLogger(__name__)
[docs]
class Kind(enum.StrEnum):
"""The different types of log content that are recognized."""
def __new__(cls, value: str, doc: str):
self = str.__new__(cls, value)
self._value_ = value
self.__doc__ = textwrap.dedent(doc)
return self
System = (
"system",
"""
System refers to messages that are generated by (none other than) the system or application.
In agent frameworks, these messages are typically templated and instantiated with application-defined
objectives / instructions.
""",
)
ToolCall = (
"tool-call",
"""
ToolCall refers to messages that contain (typically LLM generated) arguments for invoking a tool.
These logs are not to be confused with *ToolResult* messages which contain the results of invoking a tool.
""",
)
ToolResult = (
"tool-result",
"""
ToolResult refers to messages containing the results of invoking a tool.
These logs are not to be confused with *ToolCall* messages which are (typically) generated by an LLM.
""",
)
ChatCompletion = (
"chat-completion",
"""
ChatCompletion refers to messages that are generated using a language model.
Ideally, these messages should be captured immediately after generation (without any post-processing).
""",
)
RequestHeader = (
"request-header",
"""
RequestHeader refers to messages that *specifically* capture tools and output types used in a request to
a language model.
""",
)
User = (
"user",
"""
User refers to messages that are directly sent by (none other than) the user.
If the application uses prompt templates, these messages refer to the raw user input (not the templated text).
""",
)
Assistant = (
"assistant",
"""
Assistant refers to messages that are directly served back to the user.
These messages exclude any *ModelOutput* or *System* messages that are used internally by the application.
""",
)
Begin = (
"begin",
"""
Begin refers to marker messages that are used to indicate the start of a span (e.g., a task, agent, state,
etc...).
These messages are typically used to trace how one unit of work mutates some state, and are application
specific.
""",
)
End = (
"end",
"""
End refers to marker messages that are used to indicate the end of a span (e.g., a task, agent, state, etc...).
These messages are typically used to trace how one unit of work mutates some state, and are application
specific.
""",
)
Edge = (
"edge",
"""
Edge refers to marker messages that are used to indicate one unit of work (essentially) invoking another unit
of work.
These messages can be used to trace 'handoffs' and 'agent-to-agent' communication.
""",
)
KeyValue = (
"key-value",
"""
KeyValue refers to messages that contain user-specified data that are to be logged under some span.
""",
)
class BaseContent(pydantic.BaseModel):
model_config = pydantic.ConfigDict(frozen=True, use_enum_values=True)
extra: typing.Optional[dict] = pydantic.Field(
description="Additional data that is associated with the content. This field is optional.", default=None
)
@staticmethod
def _safe_serialize(obj):
"""Source available at: https://stackoverflow.com/a/74923639"""
def _safe_serialize_impl(inner_obj):
if isinstance(inner_obj, list):
result = list()
for element in inner_obj:
result.append(_safe_serialize_impl(element))
return result
elif isinstance(inner_obj, dict):
result = dict()
for key, value in inner_obj.items():
result[key] = _safe_serialize_impl(value)
return result
elif hasattr(inner_obj, "__dict__"):
if hasattr(inner_obj, "__repr__"):
result = inner_obj.__repr__()
else:
# noinspection PyBroadException
try:
result = inner_obj.__class__.__name__
except:
result = "object"
return result
else:
return inner_obj
return _safe_serialize_impl(obj)
@pydantic.field_serializer("extra", when_used="json")
def _serialize_extra(self, extra: dict, _info):
return self._safe_serialize(extra)
[docs]
class SystemContent(BaseContent):
"""System refers to messages that are generated by (none other than) the system or application.
.. card:: Class Description
In agent frameworks, these messages are typically templated and instantiated with application-defined
objectives / instructions.
In our integration packages (e.g., :py:class:`agentc_langchain.chat.Callback`), system messages are commonly
used to record the contents used to generate a chat-completion or tool-call message.
A system message has a single required field, ``value``, which contains the content of the system message.
If extra data is associated with the system message (e.g., LangGraph's ``run_id`` fields), it can be stored in
the optional ``extra`` field.
"""
kind: typing.Literal[Kind.System] = Kind.System
value: str = pydantic.Field(description="The content of the system message.")
[docs]
class ToolCallContent(BaseContent):
"""ToolCall refers to messages that contain (typically LLM generated) arguments for invoking a tool.
.. card:: Class Description
LLMs enable the invocation of standard Python functions (i.e., "tools") by a) "selecting" tools from some larger
tool-set and b) generating arguments to these tools.
This type of content is not to be confused with :py:class:`ToolResult` messages which contain the results of
invoking a tool (i.e., the output).
A tool call message has two required fields: ``tool_name`` and ``tool_args``.
The ``tool_name`` field refers to the name of the tool that is being called.
The ``tool_args`` field contains the arguments that are going to be passed to the tool (represented as a
dictionary keyed by parameter names whose entries are the parameter values).
Optional fields for a tool call message include ``tool_call_id``, ``status``, ``meta``, and ``extra``.
The ``tool_call_id`` field is an optional unique identifier associated with a tool call instance and is used
to correlate the call to the result of its execution (i.e., the :py:class:`ToolResult` message) by the
application.
The ``status`` field is an optional field that indicates the status of *generating* the tool call message
(e.g., the generated output does not adhere to the function signature).
To capture the breadth of LLM-provider metadata, tool call messages may also contain a ``meta`` field
(used to capture the raw response associated with the tool call message).
If extra data is associated with the tool call (e.g., the log-probabilities), it can be stored in the optional
``extra`` field.
.. tip::
If ``tool_call_id`` is not specified *but* your application calls and executes tools sequentially, you can still
link the tool call to the corresponding tool result by using the order of the messages in the log.
This is the approach taken by the :sql:`ToolInvocations` view (more information can be found
`here <analysis.html#toolinvocations-view>`__).
"""
kind: typing.Literal[Kind.ToolCall] = Kind.ToolCall
tool_name: str = pydantic.Field(
description="The name of the tool that is being called. If this tool is indexed with Agent Catalog, this field "
"should refer to the tool's 'name' field."
)
tool_args: dict[str, typing.Any] = pydantic.Field(
description="The arguments that are going to be passed to the tool. This field should be JSON-serializable."
)
tool_call_id: str = pydantic.Field(
description="The unique identifier associated with a tool call instance. "
"This field is (typically) parsed from a LLM response and is used to correlate / JOIN this "
"message with the corresponding ToolResult message."
)
status: typing.Optional[typing.Literal["success", "error"]] = "success"
meta: typing.Optional[dict] = pydantic.Field(
description="The raw response associated with the tool call. This must be JSON-serializable.",
default=None,
)
@pydantic.field_serializer("tool_args", when_used="json")
def _serialize_tool_args(self, tool_args: dict[str, typing.Any], _info):
return self._safe_serialize(tool_args)
[docs]
class ToolResultContent(BaseContent):
"""ToolResult refers to messages containing the results of invoking a tool.
.. card:: Class Description
Tool result messages are used to capture the output of invoking a tool (i.e., the result of the tool call).
This type of content is not to be confused with :py:class:`ToolCall` messages which contain the arguments for
invoking a tool.
A tool result message has a single required field, ``tool_result``, which contains a JSON-serializable object
representing the result of executing the tool.
Optional fields for a tool result message include ``tool_call_id``, ``status``, and ``extra``.
The ``tool_call_id`` field is an optional unique identifier associated with a tool call instance and is used
here to correlate the execution of a tool to its call generation.
``status`` is an optional field that indicates the status of the tool invocation (i.e., "success", or "error"
if the tool itself raised an exception).
Finally, if extra data is associated with the tool result (e.g., the error message on unsuccessful tool
invocations), it can be stored in the optional ``extra`` field.
"""
kind: typing.Literal[Kind.ToolResult] = Kind.ToolResult
tool_call_id: typing.Optional[str] = pydantic.Field(
description="The unique identifier of the tool call. This field will be used to correlate / JOIN this message "
"with the corresponding ToolCall message.",
default=None,
)
tool_result: typing.Any = pydantic.Field(
description="The result of invoking the tool. This field should be JSON-serializable."
)
status: typing.Optional[typing.Literal["success", "error"]] = pydantic.Field(
description="The status of the tool invocation. This field should be one of 'success' or 'error'.",
default="success",
)
@pydantic.field_serializer("tool_result", when_used="json")
def _serialize_tool_result(self, tool_result: typing.Any, _info):
return self._safe_serialize(tool_result)
[docs]
class ChatCompletionContent(BaseContent):
"""ChatCompletion refers to messages that are generated using a language model.
.. card:: Class Description
Chat completion messages refer to the output "predicted" text of a language model.
In the context of "agentic" applications, these messages are distinct from :py:class:`ToolCallContent` messages
(even though both are generated using LLMs).
A chat completion message has one required field: ``output``.
The ``output`` field refers to the unstructured generated text returned by the language model.
To capture the breadth of LLM-provider metadata, chat completion messages may also contain a ``meta`` field
(used to capture the raw response associated with the chat completion).
Finally, any extra data that exists outside the raw response can be stored in the optional ``extra`` field.
"""
kind: typing.Literal[Kind.ChatCompletion] = Kind.ChatCompletion
output: str = pydantic.Field(description="The output of the model.")
meta: typing.Optional[dict] = pydantic.Field(
description="The raw response associated with the chat completion. This must be JSON-serializable.",
default=None,
)
@pydantic.field_serializer("meta", when_used="json")
def _serialize_meta(self, meta: dict, _info):
return self._safe_serialize(meta)
[docs]
class UserContent(BaseContent):
"""User refers to messages that are directly sent by (none other than) the user.
.. card:: Class Description
User messages are used to capture a user's input into your application.
User messages exclude those generated by a prior LLM call for the purpose of "mocking" an intelligent actor
(e.g., multi-agent applications).
A user message has a single required field, ``value``, which contains the direct input given by the user.
If extra data is associated with the user message, it can be stored in the optional ``extra`` field.
"""
kind: typing.Literal[Kind.User] = Kind.User
value: str = pydantic.Field(description="The captured user input.")
user_id: typing.Optional[str] = pydantic.Field(
description="The unique identifier of the user. This field is optional.", default=None
)
[docs]
class AssistantContent(BaseContent):
"""Assistant refers to messages that are directly served back to the user.
.. card:: Class Description
Assistant messages are used to capture the direct output of your application (i.e., the messages served back to
the user).
These messages are *not* strictly :py:class:`ChatCompletionContent` messages, as your application may utilize
multiple LLM calls before returning some message back to the user.
An assistant message has a single required field, ``value``, which contains the direct output back to the user.
If extra data is associated with the assistant message, it can be stored in the optional ``extra`` field.
"""
kind: typing.Literal[Kind.Assistant] = Kind.Assistant
value: str = pydantic.Field(description="The response served back to the user.")
[docs]
class EdgeContent(BaseContent):
"""Edge refers to messages used to capture a caller's intent to 'handoff' some state to another span.
.. card:: Card Description
Edge messages denote the explicit intent to invoke another unit of work.
In the case of multi-agent applications, this type of message can be used to capture how one agent might call
another agent.
Edge messages have two required fields: i) ``source``, the fully qualified name of the source :py:class:`Span`,
and ii) ``dest``, the fully qualified name of the destination :py:class:`Span`.
A ``payload`` field can optionally be recorded in an edge message to model spans as a functional unit of work
(which they are in most cases).
If extra data is associated with the edge message, it can be stored in the optional ``extra`` field.
There are two paradigms around span to span "communication": a) *horizontal* and b) *vertical*.
Horizontal communication refers to (span) graphs that are managed by some orchestrator (e.g., LangGraph, CrewAI,
etc...), while vertical communication refers to (span) graphs that are built by directly invoking other spans
(ultimately building a call stack).
.. note::
In the case of our :py:class:`agentc_langgraph.agent.ReActAgent` helper class, a ``previous_node`` field is
used to help build these edges for horizontal communication.
"""
kind: typing.Literal[Kind.Edge] = Kind.Edge
source: list[str] = pydantic.Field(description="Name of the source span associated with this edge.")
dest: list[str] = pydantic.Field(description="Name of the destination span associated with this edge.")
payload: typing.Optional[typing.Any] = pydantic.Field(
description="A (JSON-serializable) item being sent from the source span to the destination span.", default=None
)
[docs]
class BeginContent(BaseContent):
"""Begin refers to marker messages that are used to indicate the start of a span (e.g., a task, agent, state,
etc...).
.. card:: Card Description
Begin messages denote the start of a span *and* (perhaps just as important) record the entrance state of a span.
In certain applications, log analysts are able to use this information to model how the state of an application
mutates over time.
Begin messages have two optional fields: i) the ``state`` field (used to record the starting state of a span)
and ii) ``extra`` (used to record any extra information pertaining to the start of a span).
"""
kind: typing.Literal[Kind.Begin] = Kind.Begin
state: typing.Optional[typing.Any] = pydantic.Field(
description="The state logged on entering a span.", default=None
)
[docs]
class EndContent(BaseContent):
r"""End refers to marker messages that are used to indicate the end of a span (e.g., a task, agent, state, etc...).
.. card:: Class Description
End messages denote the end of a span *and* (perhaps just as important) record the exit state of a span.
In certain applications, log analysts are able to use this information to model how the state of an application
mutates over time.
End messages have two optional fields: i) the ``state`` field (used to record the ending state of a span)
and ii) ``extra`` (used to record any extra information pertaining to the end of a span).
"""
kind: typing.Literal[Kind.End] = Kind.End
state: typing.Optional[typing.Any] = pydantic.Field(description="The state logged on exiting a span.", default=None)
[docs]
class KeyValueContent(BaseContent):
"""KeyValue refers to messages that contain user-specified data that are to be logged under some span.
.. card:: Class Description
Key-value messages serve as the catch-all container for user-defined data that belong to some span.
We distinguish key-value messages from log-level *annotations*, which are used to attach information to a log
entry with existing content.
Key-value messages have two required fields: i) the ``key`` field (used to record the name of the entry)
and ii) ``value`` (used to record the value of the entry).
Extra data can also be passed in through the optional ``extra`` field (as with other messages).
Using the :python:`[]` syntax with a :py:class:`agentc.span.Span` instance generates one key-value entry per
call.
For example, the following code snippet generates two logs with the same key but different values:
.. code-block:: python
my_span = catalog.Span(name="my_span")
my_span["alpha"] = alpha_value_1
my_span["alpha"] = alpha_value_2
These messages are commonly purposed for recording evaluation data, as seen in our example application
`here <https://github.com/couchbaselabs/agent-catalog/tree/master/examples/with_langgraph>`__.
"""
kind: typing.Literal[Kind.KeyValue] = Kind.KeyValue
key: str = pydantic.Field(description="The name of the key-value pair.")
value: typing.Any = pydantic.Field(
description="The value of the key-value pair. This value should be JSON-serializable."
)
@pydantic.field_serializer("value", when_used="json")
def _serialize_value(self, value: typing.Any, _info):
return self._safe_serialize(value)
Content = typing.Union[
SystemContent,
ToolCallContent,
ToolResultContent,
ChatCompletionContent,
RequestHeaderContent,
UserContent,
AssistantContent,
BeginContent,
EndContent,
EdgeContent,
KeyValueContent,
]