mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 22:04:20 +00:00
86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
"""
|
|
Script to upload local prompts to Langfuse.
|
|
|
|
Uploads the project prompts with the agreed names so that
|
|
prompt_manager can retrieve them via get_prompt_from_langfuse().
|
|
|
|
Usage:
|
|
uv run python scripts/upload_prompts_to_langfuse.py
|
|
"""
|
|
|
|
# Import settings BEFORE the Langfuse SDK to ensure env vars are set
|
|
from src.core.config import settings
|
|
settings.setup_langfuse()
|
|
|
|
from langfuse import Langfuse
|
|
from src.core.logging import setup_logging
|
|
|
|
logger = setup_logging(log_level=settings.LOG_LEVEL, log_format=settings.LOG_FORMAT)
|
|
|
|
# Mapping: Langfuse prompt name → local prompt variable
|
|
from src.agent.local_prompts.canceling_analysis import canceling_analysis_pt
|
|
from src.agent.local_prompts.ticket_reclassification import ticket_reclassification_pt
|
|
from src.agent.local_prompts.tim_complaint_analysis import tim_complaint_analysis_pt
|
|
from src.agent.local_prompts.emulator.response_emulator_generation import response_emulator_generation_pt
|
|
|
|
PROMPTS_TO_UPLOAD = [
|
|
{
|
|
"name": "ticket_reclassification_pt",
|
|
"prompt": ticket_reclassification_pt,
|
|
"description": "Unified reclassification prompt that recieves motives AND decides between (tratamento / reclassificar).",
|
|
},
|
|
{
|
|
"name": "ticket_canceling_analysis_pt",
|
|
"prompt": canceling_analysis_pt,
|
|
"description": "Policy violation cancellation check prompt (alternative flow).",
|
|
},
|
|
{
|
|
"name": "tim_complaint_analysis_pt",
|
|
"prompt": tim_complaint_analysis_pt,
|
|
"description": "TIM-specific complaint analysis prompt to determine if the complaint is about TIM, a different operator, or is undefined.",
|
|
},
|
|
{
|
|
"name": "response_emulator_generation_pt",
|
|
"prompt": response_emulator_generation_pt,
|
|
"description": "Anatel response generation prompt for the emulator (opening/body/ombudsman structure).",
|
|
# Tunable history window read by generate_response_node via config.
|
|
"config": {"max_regeneration_history": 4},
|
|
},
|
|
]
|
|
|
|
# Upload
|
|
def main():
|
|
logger.info(f"Connecting to Langfuse at: {settings.LANGFUSE_BASE_URL}")
|
|
client = Langfuse(
|
|
public_key=settings.LANGFUSE_PUBLIC_KEY,
|
|
secret_key=settings.LANGFUSE_SECRET_KEY,
|
|
host=settings.LANGFUSE_BASE_URL,
|
|
)
|
|
|
|
for item in PROMPTS_TO_UPLOAD:
|
|
name = item["name"]
|
|
prompt_text = item["prompt"].strip()
|
|
description = item["description"]
|
|
config = {"description": description, **item.get("config", {})}
|
|
|
|
logger.info(f"Uploading prompt '{name}'...")
|
|
try:
|
|
result = client.create_prompt(
|
|
name=name,
|
|
prompt=prompt_text,
|
|
labels=["production"],
|
|
config=config,
|
|
type="text",
|
|
)
|
|
logger.info(f"Prompt '{name}' created/updated (version={result.version})")
|
|
except Exception as e:
|
|
logger.error(f"Failed to upload prompt '{name}': {e}")
|
|
|
|
logger.info("Upload complete. Check the Langfuse dashboard to verify the prompts.")
|
|
client.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|