可观测性：建立模型可见上下文的纯度量模型

新增独立的上下文度量模块，按 System、Human、AI、Tool 和其他消息类型拆分每轮模型输入，并记录消息数量、稳定序列化字节数、文本体积与 SHA-256。

度量同时覆盖历史工具调用、ToolMessage、模型工具 schema 和多模态图片；对于 data URL 记录编码与可确定的解码字节数，对于外部引用只保留来源类型和哈希。

通过当前消息链确定上一轮输入范围和已报告 token，计算消息、字节、工具与图片维度的增长量。所有输出仅包含类型、大小、计数和哈希，不返回提示词正文、图片内容、URL 或本地路径。

新增中文文本、多模态图片、工具 schema、上一轮增长量、稳定哈希和敏感正文排除测试。

production_build_agents/observability/context_metrics.py

@@ -0,0 +1,597 @@
+"""模型可见上下文的纯度量函数。
+
+本模块只计算类型、大小、计数和 hash，不保存消息正文，也不依赖
+OB Agent、LangGraph、文件系统或业务合同。
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import json
+from collections import Counter
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+from langchain_core.messages import AIMessage
+
+
+CONTEXT_METRICS_SCHEMA = "1.0"
+_MESSAGE_ROLES = ("system", "human", "ai", "tool", "other")
+
+
+def _canonical_bytes(value: object) -> bytes:
+    try:
+        rendered = json.dumps(
+            value,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        )
+    except (TypeError, ValueError):
+        rendered = str(value)
+    return rendered.encode("utf-8")
+
+
+def _hash_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def value_size_and_hash(value: object) -> dict[str, Any]:
+    """返回稳定序列化后的字节数和 hash，不返回原值。"""
+
+    raw = _canonical_bytes(value)
+    return {
+        "serialized_bytes": len(raw),
+        "sha256": _hash_bytes(raw),
+    }
+
+
+def _message_role(message: object) -> str:
+    raw = str(
+        getattr(message, "type", None)
+        or getattr(message, "role", None)
+        or ""
+    ).lower()
+    aliases = {
+        "assistant": "ai",
+        "user": "human",
+        "function": "tool",
+    }
+    role = aliases.get(raw, raw)
+    return role if role in _MESSAGE_ROLES else "other"
+
+
+def _data_url_metrics(value: str) -> dict[str, Any] | None:
+    if not value.startswith("data:image/") or "," not in value:
+        return None
+    header, encoded = value.split(",", 1)
+    if ";base64" not in header.lower():
+        return {
+            "source_kind": "data_url",
+            "encoded_bytes": len(encoded.encode("utf-8")),
+            "decoded_bytes": None,
+            "sha256": _hash_bytes(value.encode("utf-8")),
+        }
+    try:
+        decoded = base64.b64decode(encoded, validate=True)
+    except (binascii.Error, ValueError):
+        decoded = b""
+        decoded_bytes: int | None = None
+    else:
+        decoded_bytes = len(decoded)
+    return {
+        "source_kind": "data_url",
+        "encoded_bytes": len(encoded.encode("ascii", errors="ignore")),
+        "decoded_bytes": decoded_bytes,
+        "sha256": _hash_bytes(value.encode("utf-8")),
+    }
+
+
+def _image_source(block: Mapping[str, Any]) -> object:
+    image_url = block.get("image_url")
+    if isinstance(image_url, Mapping):
+        return image_url.get("url")
+    if image_url is not None:
+        return image_url
+    for key in ("data", "url", "source", "path", "file_path"):
+        if block.get(key) is not None:
+            return block.get(key)
+    return None
+
+
+def _image_metrics(
+    source: object,
+    *,
+    message_index: int,
+    block_index: int,
+) -> dict[str, Any]:
+    raw = str(source or "")
+    data_url = _data_url_metrics(raw)
+    if data_url is not None:
+        metrics = data_url
+    else:
+        source_kind = (
+            "remote_reference"
+            if raw.startswith(("http://", "https://"))
+            else "local_reference"
+            if raw
+            else "unknown"
+        )
+        metrics = {
+            "source_kind": source_kind,
+            "encoded_bytes": None,
+            "decoded_bytes": None,
+            "sha256": _hash_bytes(raw.encode("utf-8")),
+        }
+    return {
+        "message_index": message_index,
+        "block_index": block_index,
+        **metrics,
+    }
+
+
+def _content_metrics(
+    content: object,
+    *,
+    message_index: int,
+) -> dict[str, Any]:
+    raw = _canonical_bytes(content)
+    text_bytes = 0
+    kinds: Counter[str] = Counter()
+    images: list[dict[str, Any]] = []
+
+    if isinstance(content, str):
+        kinds["text"] += 1
+        text_bytes = len(content.encode("utf-8"))
+    elif isinstance(content, (list, tuple)):
+        for block_index, block in enumerate(content):
+            if isinstance(block, str):
+                kinds["text"] += 1
+                text_bytes += len(block.encode("utf-8"))
+                continue
+            if not isinstance(block, Mapping):
+                kinds["other"] += 1
+                continue
+            block_type = str(block.get("type") or "other").lower()
+            if block_type in {"image", "image_url", "input_image"}:
+                kinds["image"] += 1
+                images.append(
+                    _image_metrics(
+                        _image_source(block),
+                        message_index=message_index,
+                        block_index=block_index,
+                    )
+                )
+                continue
+            if block_type in {"text", "input_text", "output_text"}:
+                kinds["text"] += 1
+                text = block.get("text")
+                if isinstance(text, str):
+                    text_bytes += len(text.encode("utf-8"))
+                continue
+            kinds[block_type] += 1
+    elif content is None:
+        kinds["empty"] += 1
+    else:
+        kinds["other"] += 1
+
+    return {
+        "content_kinds": sorted(kinds),
+        "content_bytes": text_bytes,
+        "serialized_bytes": len(raw),
+        "content_sha256": _hash_bytes(raw),
+        "images": images,
+    }
+
+
+def _tool_call_metrics(tool_calls: Iterable[object]) -> list[dict[str, Any]]:
+    metrics: list[dict[str, Any]] = []
+    for index, raw in enumerate(tool_calls):
+        if not isinstance(raw, Mapping):
+            value = value_size_and_hash(raw)
+            metrics.append(
+                {
+                    "index": index,
+                    "name": "unknown",
+                    "arguments_bytes": value["serialized_bytes"],
+                    "arguments_sha256": value["sha256"],
+                }
+            )
+            continue
+        arguments = raw.get("args", raw.get("arguments"))
+        value = value_size_and_hash(arguments)
+        metrics.append(
+            {
+                "index": index,
+                "name": str(raw.get("name") or "unknown")[:160],
+                "arguments_bytes": value["serialized_bytes"],
+                "arguments_sha256": value["sha256"],
+            }
+        )
+    return metrics
+
+
+def _message_metrics(message: object, index: int) -> dict[str, Any]:
+    content = getattr(message, "content", None)
+    content_metrics = _content_metrics(content, message_index=index)
+    tool_calls = _tool_call_metrics(
+        list(getattr(message, "tool_calls", None) or [])
+    )
+    role = _message_role(message)
+    wire_projection = {
+        "role": role,
+        "content": content,
+        "tool_calls": list(getattr(message, "tool_calls", None) or []),
+        "tool_call_id": getattr(message, "tool_call_id", None),
+        "name": getattr(message, "name", None),
+    }
+    wire = _canonical_bytes(wire_projection)
+    return {
+        "index": index,
+        "role": role,
+        "content_kinds": content_metrics["content_kinds"],
+        "content_bytes": content_metrics["content_bytes"],
+        "serialized_bytes": len(wire),
+        "sha256": _hash_bytes(wire),
+        "image_count": len(content_metrics["images"]),
+        "image_decoded_bytes": sum(
+            item["decoded_bytes"] or 0
+            for item in content_metrics["images"]
+        ),
+        "image_unknown_bytes_count": sum(
+            item["decoded_bytes"] is None
+            for item in content_metrics["images"]
+        ),
+        "tool_call_count": len(tool_calls),
+        "tool_call_argument_bytes": sum(
+            item["arguments_bytes"] for item in tool_calls
+        ),
+        "_images": content_metrics["images"],
+        "_tool_calls": tool_calls,
+    }
+
+
+def _tool_schema_metrics(
+    invocation_params: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    tools = (
+        invocation_params.get("tools")
+        if isinstance(invocation_params, Mapping)
+        else None
+    )
+    source = list(tools) if isinstance(tools, (list, tuple)) else []
+    items: list[dict[str, Any]] = []
+    for index, tool in enumerate(source):
+        name = "unknown"
+        if isinstance(tool, Mapping):
+            function = tool.get("function")
+            if isinstance(function, Mapping):
+                name = str(function.get("name") or name)
+            else:
+                name = str(tool.get("name") or name)
+        metrics = value_size_and_hash(tool)
+        items.append(
+            {
+                "index": index,
+                "name": name[:160],
+                "serialized_bytes": metrics["serialized_bytes"],
+                "sha256": metrics["sha256"],
+            }
+        )
+    whole = value_size_and_hash(source)
+    return {
+        "count": len(items),
+        "serialized_bytes": whole["serialized_bytes"],
+        "sha256": whole["sha256"],
+        "items": items,
+    }
+
+
+def _base_context_metrics(
+    messages: list[object],
+    invocation_params: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    private_items = [
+        _message_metrics(message, index)
+        for index, message in enumerate(messages)
+    ]
+    role_counts = Counter(item["role"] for item in private_items)
+    images = [
+        image
+        for item in private_items
+        for image in item.pop("_images")
+    ]
+    history_tool_calls = [
+        call
+        for item in private_items
+        for call in item.pop("_tool_calls")
+    ]
+    tool_schemas = _tool_schema_metrics(invocation_params)
+    context_projection = {
+        "messages": [
+            {
+                "role": _message_role(message),
+                "content": getattr(message, "content", None),
+                "tool_calls": list(
+                    getattr(message, "tool_calls", None) or []
+                ),
+                "tool_call_id": getattr(message, "tool_call_id", None),
+                "name": getattr(message, "name", None),
+            }
+            for message in messages
+        ],
+        "tools": (
+            invocation_params.get("tools", [])
+            if isinstance(invocation_params, Mapping)
+            else []
+        ),
+    }
+    context = _canonical_bytes(context_projection)
+    return {
+        "message_count": len(private_items),
+        "message_role_counts": {
+            role: role_counts.get(role, 0) for role in _MESSAGE_ROLES
+        },
+        "serialized_bytes": len(context),
+        "sha256": _hash_bytes(context),
+        "messages": private_items,
+        "tool_messages": {
+            "count": role_counts.get("tool", 0),
+            "serialized_bytes": sum(
+                item["serialized_bytes"]
+                for item in private_items
+                if item["role"] == "tool"
+            ),
+            "content_bytes": sum(
+                item["content_bytes"]
+                for item in private_items
+                if item["role"] == "tool"
+            ),
+        },
+        "history_tool_calls": {
+            "count": len(history_tool_calls),
+            "argument_bytes": sum(
+                item["arguments_bytes"] for item in history_tool_calls
+            ),
+        },
+        "tool_schemas": tool_schemas,
+        "images": {
+            "count": len(images),
+            "data_url_count": sum(
+                item["source_kind"] == "data_url" for item in images
+            ),
+            "reference_count": sum(
+                item["source_kind"]
+                in {"remote_reference", "local_reference"}
+                for item in images
+            ),
+            "decoded_bytes": sum(
+                item["decoded_bytes"] or 0 for item in images
+            ),
+            "unknown_bytes_count": sum(
+                item["decoded_bytes"] is None for item in images
+            ),
+            "items": images,
+        },
+    }
+
+
+def _usage_value(message: object, *keys: str) -> int | None:
+    usage = getattr(message, "usage_metadata", None)
+    if not isinstance(usage, Mapping):
+        usage = {}
+    response = getattr(message, "response_metadata", None)
+    response = response if isinstance(response, Mapping) else {}
+    candidates: list[Mapping[str, Any]] = [usage]
+    for name in ("token_usage", "usage"):
+        nested = response.get(name)
+        if isinstance(nested, Mapping):
+            candidates.append(nested)
+    for source in candidates:
+        for key in keys:
+            value = source.get(key)
+            if isinstance(value, int) and value >= 0:
+                return value
+    return None
+
+
+def _previous_input(
+    messages: list[object],
+) -> tuple[list[object] | None, object | None]:
+    for index in range(len(messages) - 1, -1, -1):
+        if isinstance(messages[index], AIMessage):
+            return messages[:index], messages[index]
+    return None, None
+
+
+def _delta(current: Mapping[str, Any], previous: Mapping[str, Any]) -> dict[str, Any]:
+    current_roles = current["message_role_counts"]
+    previous_roles = previous["message_role_counts"]
+    return {
+        "message_count": current["message_count"] - previous["message_count"],
+        "serialized_bytes": (
+            current["serialized_bytes"] - previous["serialized_bytes"]
+        ),
+        "message_role_counts": {
+            role: current_roles[role] - previous_roles[role]
+            for role in _MESSAGE_ROLES
+        },
+        "tool_message_bytes": (
+            current["tool_messages"]["serialized_bytes"]
+            - previous["tool_messages"]["serialized_bytes"]
+        ),
+        "image_count": current["images"]["count"] - previous["images"]["count"],
+        "image_decoded_bytes": (
+            current["images"]["decoded_bytes"]
+            - previous["images"]["decoded_bytes"]
+        ),
+    }
+
+
+def build_context_snapshot(
+    messages: Iterable[object],
+    *,
+    invocation_params: Mapping[str, Any] | None,
+    identity: Mapping[str, Any],
+    physical_call_id: str,
+) -> dict[str, Any]:
+    """构造一次模型调用开始时的无正文上下文快照。"""
+
+    message_list = list(messages)
+    metrics = _base_context_metrics(message_list, invocation_params)
+    logical_call_index = (
+        sum(isinstance(message, AIMessage) for message in message_list) + 1
+    )
+    agent_run_id = str(identity.get("agent_run_id") or "unknown")
+    logical_key = _hash_bytes(
+        (
+            f"{agent_run_id}\0{logical_call_index}\0{metrics['sha256']}"
+        ).encode("utf-8")
+    )
+    previous_messages, previous_output = _previous_input(message_list)
+    previous_metrics = (
+        _base_context_metrics(previous_messages, invocation_params)
+        if previous_messages is not None
+        else None
+    )
+    previous_input_tokens = (
+        _usage_value(previous_output, "input_tokens", "prompt_tokens")
+        if previous_output is not None
+        else None
+    )
+    return {
+        "schema_version": CONTEXT_METRICS_SCHEMA,
+        "event_kind": "model_context_start",
+        "physical_call_id": physical_call_id,
+        "logical_call_index": logical_call_index,
+        "logical_call_key": logical_key,
+        "identity": dict(identity),
+        "input": metrics,
+        "previous": (
+            {
+                "logical_call_index": logical_call_index - 1,
+                "input_tokens": previous_input_tokens,
+                "delta": _delta(metrics, previous_metrics),
+            }
+            if previous_metrics is not None
+            else None
+        ),
+    }
+
+
+def _usage_details(message: object, response: object) -> dict[str, Any]:
+    input_tokens = _usage_value(message, "input_tokens", "prompt_tokens")
+    output_tokens = _usage_value(
+        message,
+        "output_tokens",
+        "completion_tokens",
+    )
+    total_tokens = _usage_value(message, "total_tokens")
+    usage = getattr(message, "usage_metadata", None)
+    usage = usage if isinstance(usage, Mapping) else {}
+    input_details = usage.get("input_token_details")
+    output_details = usage.get("output_token_details")
+    llm_output = getattr(response, "llm_output", None)
+    if total_tokens is None and input_tokens is not None and output_tokens is not None:
+        total_tokens = input_tokens + output_tokens
+    return {
+        "input_tokens": input_tokens,
+        "output_tokens": output_tokens,
+        "total_tokens": total_tokens,
+        "input_token_details": (
+            dict(input_details) if isinstance(input_details, Mapping) else {}
+        ),
+        "output_token_details": (
+            dict(output_details) if isinstance(output_details, Mapping) else {}
+        ),
+        "provider_usage_present": bool(
+            usage or isinstance(llm_output, Mapping)
+        ),
+    }
+
+
+def build_terminal_metrics(
+    snapshot: Mapping[str, Any] | None,
+    *,
+    message: object | None,
+    response: object | None,
+    status: str,
+    duration_ms: int,
+    transport: Mapping[str, Any],
+    error_type: str | None = None,
+) -> dict[str, Any]:
+    """构造成功或失败的调用终态度量。"""
+
+    physical_call_id = (
+        str(snapshot.get("physical_call_id"))
+        if isinstance(snapshot, Mapping)
+        else "unknown"
+    )
+    output = (
+        _message_metrics(message, 0)
+        if message is not None
+        else None
+    )
+    if output is not None:
+        output.pop("_images", None)
+        output.pop("_tool_calls", None)
+    usage = _usage_details(message, response) if message is not None else {
+        "input_tokens": None,
+        "output_tokens": None,
+        "total_tokens": None,
+        "input_token_details": {},
+        "output_token_details": {},
+        "provider_usage_present": False,
+    }
+    previous_tokens = None
+    if isinstance(snapshot, Mapping):
+        previous = snapshot.get("previous")
+        if isinstance(previous, Mapping):
+            previous_tokens = previous.get("input_tokens")
+    token_delta = (
+        usage["input_tokens"] - previous_tokens
+        if isinstance(usage["input_tokens"], int)
+        and isinstance(previous_tokens, int)
+        else None
+    )
+    return {
+        "schema_version": CONTEXT_METRICS_SCHEMA,
+        "event_kind": "model_context_terminal",
+        "physical_call_id": physical_call_id,
+        "logical_call_index": (
+            snapshot.get("logical_call_index")
+            if isinstance(snapshot, Mapping)
+            else None
+        ),
+        "logical_call_key": (
+            snapshot.get("logical_call_key")
+            if isinstance(snapshot, Mapping)
+            else None
+        ),
+        "identity": (
+            dict(snapshot.get("identity") or {})
+            if isinstance(snapshot, Mapping)
+            else {}
+        ),
+        "status": status,
+        "duration_ms": max(0, duration_ms),
+        "error_type": error_type,
+        "usage": {**usage, "input_tokens_delta": token_delta},
+        "output": output,
+        "transport": dict(transport),
+    }
+
+
+def summarize_tool_contribution(content: object) -> dict[str, Any]:
+    """描述一次 ToolMessage 将追加到上下文中的大小，不保存结果正文。"""
+
+    metrics = value_size_and_hash(content)
+    return {
+        "schema_version": CONTEXT_METRICS_SCHEMA,
+        "message_role": "tool",
+        "serialized_bytes": metrics["serialized_bytes"],
+        "sha256": metrics["sha256"],
+    }

tests/observability/test_context_metrics.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import json
+
+from langchain_core.messages import (
+    AIMessage,
+    HumanMessage,
+    SystemMessage,
+    ToolMessage,
+)
+
+from production_build_agents.observability.context_metrics import (
+    build_context_snapshot,
+    build_terminal_metrics,
+    summarize_tool_contribution,
+)
+
+
+def test_context_snapshot_measures_messages_tools_images_and_growth() -> None:
+    previous = AIMessage(
+        content="调用工具",
+        tool_calls=[
+            {
+                "id": "call-1",
+                "name": "inspect_tool",
+                "args": {"工具参数": "不能进入指标正文"},
+            }
+        ],
+        usage_metadata={
+            "input_tokens": 120,
+            "output_tokens": 10,
+            "total_tokens": 130,
+        },
+    )
+    messages = [
+        SystemMessage(content="系统提示词秘密"),
+        HumanMessage(
+            content=[
+                {"type": "text", "text": "中文输入"},
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": "data:image/png;base64,YWJj",
+                    },
+                },
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": "https://secret.example/image.png",
+                    },
+                },
+            ]
+        ),
+        previous,
+        ToolMessage(
+            content='{"secret_result":"不能进入指标正文"}',
+            tool_call_id="call-1",
+            name="inspect_tool",
+        ),
+    ]
+    snapshot = build_context_snapshot(
+        messages,
+        invocation_params={
+            "tools": [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": "inspect_tool",
+                        "description": "工具说明秘密",
+                        "parameters": {"type": "object"},
+                    },
+                }
+            ]
+        },
+        identity={
+            "business_run_id": "run-1",
+            "agent_run_id": "agent-1",
+            "role": "executor",
+        },
+        physical_call_id="physical-1",
+    )
+
+    assert snapshot["logical_call_index"] == 2
+    assert snapshot["input"]["message_role_counts"] == {
+        "system": 1,
+        "human": 1,
+        "ai": 1,
+        "tool": 1,
+        "other": 0,
+    }
+    assert snapshot["input"]["tool_messages"]["count"] == 1
+    assert snapshot["input"]["history_tool_calls"]["count"] == 1
+    assert snapshot["input"]["tool_schemas"]["count"] == 1
+    assert snapshot["input"]["images"]["count"] == 2
+    assert snapshot["input"]["images"]["decoded_bytes"] == 3
+    assert snapshot["input"]["images"]["unknown_bytes_count"] == 1
+    assert snapshot["previous"]["input_tokens"] == 120
+    assert snapshot["previous"]["delta"]["message_count"] == 2
+
+    serialized = json.dumps(snapshot, ensure_ascii=False)
+    for secret in (
+        "系统提示词秘密",
+        "中文输入",
+        "不能进入指标正文",
+        "https://secret.example/image.png",
+        "YWJj",
+        "工具说明秘密",
+    ):
+        assert secret not in serialized
+
+
+def test_first_call_has_no_previous_delta_and_terminal_uses_real_tokens() -> None:
+    snapshot = build_context_snapshot(
+        [SystemMessage(content="system"), HumanMessage(content="hello")],
+        invocation_params={},
+        identity={"agent_run_id": "agent-1"},
+        physical_call_id="physical-1",
+    )
+    output = AIMessage(
+        content="done",
+        usage_metadata={
+            "input_tokens": 25,
+            "output_tokens": 5,
+            "total_tokens": 30,
+            "input_token_details": {"cache_read": 4},
+        },
+    )
+    terminal = build_terminal_metrics(
+        snapshot,
+        message=output,
+        response=object(),
+        status="success",
+        duration_ms=17,
+        transport={
+            "observed": False,
+            "attempt_count": None,
+            "retry_count": None,
+            "attempts": [],
+        },
+    )
+
+    assert snapshot["previous"] is None
+    assert terminal["usage"]["input_tokens"] == 25
+    assert terminal["usage"]["input_tokens_delta"] is None
+    assert terminal["output"]["content_bytes"] == 4
+    assert "done" not in json.dumps(terminal)
+
+
+def test_tool_contribution_only_contains_size_and_hash() -> None:
+    contribution = summarize_tool_contribution(
+        {"secret": "tool result", "success": True}
+    )
+
+    assert contribution["message_role"] == "tool"
+    assert contribution["serialized_bytes"] > 0
+    assert "tool result" not in json.dumps(contribution)