Skip to content

chatomon.observability

Tracing and content-capture helpers shared by chatomon modules.

chatomon depends only on opentelemetry-api: without an SDK configured by the application, every span is a no-op.

Classes:

Name Description
ContentPolicy

Decides how much message content the turn reports record.

Functions:

Name Description
get_tracer

Return chatomon's tracer.
stable_json

Serialize a value to canonical JSON: sorted keys, no spaces.
sha256_text

Hash a text with SHA-256.

Attributes:

Name Type Description
TRACER_NAME

Instrumentation scope name of every chatomon span.
REDACTION_FAILED

Recorded instead of a text when the redactor raises or returns no text.
UNSERIALIZABLE

Recorded instead of structured data that cannot be serialized to JSON.

TRACER_NAME module-attribute 

TRACER_NAME = 'chatomon'

Instrumentation scope name of every chatomon span.

REDACTION_FAILED module-attribute 

REDACTION_FAILED = '[redaction failed]'

Recorded instead of a text when the redactor raises or returns no text.

UNSERIALIZABLE module-attribute 

UNSERIALIZABLE = '[unserializable]'

Recorded instead of structured data that cannot be serialized to JSON.

ContentPolicy 

ContentPolicy(
    mode: CaptureContent = "none",
    redactor: Callable[[str], str] | None = None,
)

Decides how much message content the turn reports record.

Log records and spans carry labels only, whatever the mode.

Parameters:

Name Type Description Default
mode CaptureContent

"none" records nothing, "hash" records SHA-256 digests, "full" records the text passed through redactor.

 'none'
redactor Callable[[str], str] | None

Function applied to every text recorded in "full" mode.

 None

Raises:

Type Description
ChatomonConfigError

If mode is not one of the three values.

Examples:

from chatomon.observability import ContentPolicy

policy = ContentPolicy("full", redactor=lambda text: text.replace("@", " at "))
assert policy.text("mail me at a@b.it") == "mail me at a at b.it"
assert ContentPolicy().text("anything") is None

Methods:

Name Description
text

Return what may be recorded for a text.
data

Return what may be recorded for structured data, as JSON.

Attributes:

Name Type Description
mode CaptureContent

The capture mode.
Source code in src/chatomon/observability.py 
110
111
112
113
114
115
116
117
118
def __init__(
    self, mode: CaptureContent = "none", redactor: Callable[[str], str] | None = None
) -> None:
    if mode not in _MODES:
        raise ChatomonConfigError(
            f"capture_content must be one of {list(_MODES)}, got {mode!r}"
        )
    self._mode: CaptureContent = mode
    self._redactor = redactor

mode property 

mode: CaptureContent

The capture mode.

text 

text(value: str | None) -> str | None

Return what may be recorded for a text.

Never raises: if the redactor fails, or returns something that is not a text, the marker "[redaction failed]" is recorded instead of the text.

Parameters:

Name Type Description Default
value str | None

The text, or None.

 required

Returns:

Type Description
str | None

None, a digest or the (redacted) text, depending on the mode.
Source code in src/chatomon/observability.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def text(self, value: str | None) -> str | None:
    """Return what may be recorded for a text.

    Never raises: if the redactor fails, or returns something that is not a
    text, the marker `"[redaction failed]"` is recorded instead of the text.

    Args:
        value: The text, or `None`.

    Returns:
        `None`, a digest or the (redacted) text, depending on the mode.
    """
    if value is None or self._mode == "none":
        return None
    if self._mode == "hash":
        return sha256_text(value)
    if self._redactor is None:
        return value
    try:
        redacted = self._redactor(value)
    except Exception:
        return REDACTION_FAILED
    # Whatever is not a text cannot be recorded: the redactor failed there too.
    return redacted if isinstance(redacted, str) else REDACTION_FAILED

data 

data(value: Any) -> str | None

Return what may be recorded for structured data, as JSON.

Parameters:

Name Type Description Default
value Any

Any JSON-serializable value (strings are recorded as-is).

 required

Returns:

Type Description
str | None

None, a digest or the (redacted) JSON text, depending on the mode;
str | None

"[unserializable]" if the value cannot be serialized.
Source code in src/chatomon/observability.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def data(self, value: Any) -> str | None:
    """Return what may be recorded for structured data, as JSON.

    Args:
        value: Any JSON-serializable value (strings are recorded as-is).

    Returns:
        `None`, a digest or the (redacted) JSON text, depending on the mode;
        `"[unserializable]"` if the value cannot be serialized.
    """
    if value is None or self._mode == "none":
        return None
    if isinstance(value, str):
        return self.text(value)
    try:
        serialized = stable_json(value)
    except Exception:
        return UNSERIALIZABLE
    return self.text(serialized)

get_tracer 

get_tracer(
    tracer_provider: TracerProvider | None = None,
) -> Tracer

Return chatomon's tracer.

Parameters:

Name Type Description Default
tracer_provider TracerProvider | None

Provider to use; the global one when None.

 None

Returns:

Type Description
Tracer

A tracer named chatomon, versioned with the library version.
Source code in src/chatomon/observability.py 
43
44
45
46
47
48
49
50
51
52
def get_tracer(tracer_provider: trace.TracerProvider | None = None) -> trace.Tracer:
    """Return chatomon's tracer.

    Args:
        tracer_provider: Provider to use; the global one when `None`.

    Returns:
        A tracer named `chatomon`, versioned with the library version.
    """
    return trace.get_tracer(TRACER_NAME, __version__, tracer_provider=tracer_provider)

stable_json 

stable_json(value: Any) -> str

Serialize a value to canonical JSON: sorted keys, no spaces.

Pydantic models and dataclasses are converted to JSON types; other objects fall back to their repr.

Parameters:

Name Type Description Default
value Any

The value to serialize.

 required

Returns:

Type Description
str

The JSON text.
Source code in src/chatomon/observability.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def stable_json(value: Any) -> str:
    """Serialize a value to canonical JSON: sorted keys, no spaces.

    Pydantic models and dataclasses are converted to JSON types; other objects
    fall back to their `repr`.

    Args:
        value: The value to serialize.

    Returns:
        The JSON text.
    """
    return json.dumps(
        to_jsonable_python(value, fallback=repr),
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )

sha256_text 

sha256_text(text: str) -> str

Hash a text with SHA-256.

Parameters:

Name Type Description Default
text str

The text to hash.

 required

Returns:

Type Description
str

"sha256:" followed by the hex digest.
Source code in src/chatomon/observability.py 
75
76
77
78
79
80
81
82
83
84
def sha256_text(text: str) -> str:
    """Hash a text with SHA-256.

    Args:
        text: The text to hash.

    Returns:
        `"sha256:"` followed by the hex digest.
    """
    return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()