LZH-YS1998
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""Runtime artifact rendering helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any, Iterable
|
|
|
|
from .deltas import changed_artifacts
|
|
from .types import RuntimeArtifact
|
|
|
|
|
|
RUNTIME_ARTIFACT_HEADER = "## Runtime Artifact:"
|
|
RUNTIME_ARTIFACT_DELTA_HEADER = "## Runtime Artifact Delta:"
|
|
|
|
|
|
def artifact_content_hash(content: str, metadata: dict[str, Any] | None = None) -> str:
|
|
raw = json.dumps(
|
|
{
|
|
"content": str(content or ""),
|
|
"metadata": dict(metadata or {}),
|
|
},
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
default=str,
|
|
)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def build_runtime_artifact_record(
|
|
artifact_type: str,
|
|
title: str,
|
|
content: str,
|
|
*,
|
|
scope: str = "runtime",
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
content_hash = artifact_content_hash(content, metadata)
|
|
return RuntimeArtifact(
|
|
artifact_type=artifact_type,
|
|
title=title,
|
|
content=str(content or "").strip(),
|
|
scope=scope,
|
|
metadata={"content_hash": content_hash, **dict(metadata or {})},
|
|
).to_record(content_hash=content_hash)
|
|
|
|
|
|
def build_runtime_artifact_manifest(artifacts: Iterable[RuntimeArtifact]) -> tuple[list[dict[str, Any]], dict[str, str]]:
|
|
manifest: list[dict[str, Any]] = []
|
|
hashes: dict[str, str] = {}
|
|
for artifact in artifacts:
|
|
content_hash = artifact_content_hash(artifact.content, artifact.metadata)
|
|
record = artifact.to_record(content_hash=content_hash)
|
|
manifest.append(record)
|
|
hashes[artifact.artifact_type] = content_hash
|
|
return manifest, hashes
|
|
|
|
|
|
def render_runtime_artifact_messages(
|
|
artifacts: Iterable[RuntimeArtifact],
|
|
*,
|
|
previous_hashes: dict[str, str] | None = None,
|
|
emit_delta_messages: bool = True,
|
|
) -> list[dict[str, str]]:
|
|
messages: list[dict[str, str]] = []
|
|
for artifact, is_delta in changed_artifacts(artifacts, previous_hashes):
|
|
header = RUNTIME_ARTIFACT_DELTA_HEADER if emit_delta_messages and is_delta else RUNTIME_ARTIFACT_HEADER
|
|
messages.append({
|
|
"role": "system",
|
|
"content": f"{header} {artifact.title}\n{artifact.content}".strip(),
|
|
})
|
|
return messages
|
|
|
|
|
|
def is_runtime_artifact_message(message: dict[str, Any]) -> bool:
|
|
if str(message.get("role", "") or "") != "system":
|
|
return False
|
|
content = str(message.get("content", "") or "")
|
|
return content.startswith(RUNTIME_ARTIFACT_HEADER) or content.startswith(RUNTIME_ARTIFACT_DELTA_HEADER)
|
|
|
|
|
|
def strip_runtime_artifact_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return [message for message in messages if not is_runtime_artifact_message(message)]
|