86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from .prompt_release import PromptReleaseService
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PromptRuntimeRenderResult:
|
|
prompt: str
|
|
prompt_key: str
|
|
source: str
|
|
release_id: int | None = None
|
|
format_error: bool = False
|
|
|
|
def routing_meta(self) -> dict[str, Any]:
|
|
return {
|
|
"prompt_key": self.prompt_key,
|
|
"prompt_release_id": self.release_id,
|
|
"prompt_source": self.source,
|
|
"prompt_format_error": self.format_error,
|
|
}
|
|
|
|
|
|
class PromptRuntimeService:
|
|
"""Runtime access layer for tenant-managed prompt releases.
|
|
|
|
The service keeps call sites away from direct prompt release storage details.
|
|
"""
|
|
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
self.db = db
|
|
|
|
async def render(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
prompt_key: str,
|
|
fallback_template: str,
|
|
**variables: Any,
|
|
) -> PromptRuntimeRenderResult:
|
|
active_release = await PromptReleaseService(self.db).get_active_release(
|
|
tenant_id=tenant_id,
|
|
prompt_key=prompt_key,
|
|
)
|
|
active_template = str(active_release.get("body_snapshot") or "") if active_release else ""
|
|
release_id = int(active_release["id"]) if active_release and active_release.get("id") is not None else None
|
|
source = "active_release" if active_template else "default_fallback"
|
|
template = str(active_template or fallback_template or "")
|
|
try:
|
|
prompt = template.format(**variables).strip()
|
|
return PromptRuntimeRenderResult(
|
|
prompt=prompt,
|
|
prompt_key=prompt_key,
|
|
source=source,
|
|
release_id=release_id,
|
|
)
|
|
except Exception:
|
|
prompt = str(fallback_template or "").format(**variables).strip()
|
|
return PromptRuntimeRenderResult(
|
|
prompt=prompt,
|
|
prompt_key=prompt_key,
|
|
source="default_fallback",
|
|
release_id=None,
|
|
format_error=True,
|
|
)
|
|
|
|
async def render_text(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
prompt_key: str,
|
|
fallback_template: str,
|
|
**variables: Any,
|
|
) -> str:
|
|
result = await self.render(
|
|
tenant_id=tenant_id,
|
|
prompt_key=prompt_key,
|
|
fallback_template=fallback_template,
|
|
**variables,
|
|
)
|
|
return result.prompt
|