sagnik
4645ff737b
The complete code integration is done. Co-authored-by: Sagnik <sagnik7896@gmail.com> Reviewed-on: #18
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
_PROMPT_DIR = Path(__file__).resolve().parent.parent / "nemoclaw_prompts"
|
|
_PLACEHOLDER_PATTERN = re.compile(r"\{(\w+)\}")
|
|
_TEMPLATE_HINTS = {
|
|
"pipeline": ["tpl_pipeline_board_v2", "tpl_followup_queue_v1"],
|
|
"kanban": ["tpl_pipeline_board_v2"],
|
|
"map": ["tpl_geo_investor_heat_v2"],
|
|
"geo": ["tpl_geo_investor_heat_v2"],
|
|
"trend": ["tpl_absorption_trend_v1", "tpl_campaign_lead_line_v1"],
|
|
"quota": ["tpl_quota_gauge_v1", "tpl_kpi_pipeline_health_v1"],
|
|
"broker": ["tpl_broker_performance_v1"],
|
|
"source": ["tpl_qd_source_compare_v1", "tpl_bar_source_quality_v3"],
|
|
"follow": ["tpl_followup_queue_v1", "tpl_followup_gap_v1"],
|
|
"campaign": ["tpl_campaign_lead_line_v1"],
|
|
}
|
|
|
|
|
|
class PersonaService:
|
|
def __init__(self) -> None:
|
|
self.prompt_files = {
|
|
"qd_calculator": _PROMPT_DIR / "qd_calculator.md",
|
|
"lead_tagger": _PROMPT_DIR / "lead_tagger.md",
|
|
"cctv_profiler": _PROMPT_DIR / "cctv_profiler.md",
|
|
}
|
|
|
|
async def health(self) -> dict[str, Any]:
|
|
loaded = {}
|
|
for key, path in self.prompt_files.items():
|
|
loaded[key] = path.exists() and path.read_text(encoding="utf-8").strip() != ""
|
|
return {
|
|
"status": "healthy" if all(loaded.values()) else "degraded",
|
|
"prompts": loaded,
|
|
}
|
|
|
|
async def render_prompt(
|
|
self,
|
|
*,
|
|
prompt_name: str,
|
|
variables: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
path = self.prompt_files.get(prompt_name)
|
|
if path is None or not path.exists():
|
|
raise FileNotFoundError(f"Unknown prompt '{prompt_name}'.")
|
|
template = path.read_text(encoding="utf-8")
|
|
rendered = template
|
|
for key, value in variables.items():
|
|
rendered = rendered.replace(f"{{{key}}}", json.dumps(value) if isinstance(value, (dict, list)) else str(value))
|
|
unresolved = sorted(set(_PLACEHOLDER_PATTERN.findall(rendered)))
|
|
return {
|
|
"promptName": prompt_name,
|
|
"templatePath": str(path),
|
|
"renderedPrompt": rendered,
|
|
"unresolvedVariables": unresolved,
|
|
}
|
|
|
|
async def plan_for_prompt(
|
|
self,
|
|
*,
|
|
prompt: str,
|
|
tenant_id: str,
|
|
actor_role: str,
|
|
) -> dict[str, Any]:
|
|
lower_prompt = prompt.lower()
|
|
recommended: list[str] = []
|
|
for token, template_ids in _TEMPLATE_HINTS.items():
|
|
if token in lower_prompt:
|
|
recommended.extend(template_ids)
|
|
if not recommended:
|
|
recommended = ["tpl_kpi_pipeline_health_v1", "tpl_qd_source_compare_v1"]
|
|
recommended = list(dict.fromkeys(recommended))
|
|
return {
|
|
"tenantId": tenant_id,
|
|
"actorRole": actor_role,
|
|
"recommendedTemplates": recommended,
|
|
"canvasBlocks": [
|
|
{
|
|
"type": "textCanvas",
|
|
"widthMode": "full",
|
|
"minHeightPx": 180,
|
|
"content": (
|
|
"Oracle planned a mixed response: query the CRM, reuse matching component templates, "
|
|
"and synthesize missing visualization blocks if a direct template is unavailable."
|
|
),
|
|
}
|
|
],
|
|
"workflowIntent": "comfy_oracle_canvas",
|
|
}
|
|
|
|
|
|
persona_service = PersonaService()
|