| """Build multimodal messages for llama-cpp-python chat completion.""" | |
| import base64, io, uuid, datetime | |
| from typing import Optional | |
| from PIL import Image | |
| SYSTEM_PROMPT = """You are Elysium — a persistent agentic civilization. | |
| You ALWAYS respond with a single valid JSON object exactly matching the | |
| ElysiumResponse schema v1.0.0. No preamble. No markdown fences. JSON only. | |
| Decide complexity dynamically: | |
| - SIMPLE_REPLY: trivial Q — no agents (council_deliberation.agent_outputs = []) | |
| - QUERY / MORNING_BRIEFING / EVENING_REPORT: spawn 1–5 agents in agent_outputs, | |
| each with thinking + stance + tts_speech_text + tts_voice_design{voice_id,pace,tone} | |
| - TOOL_REQUIRED: populate tool_calls when external data is needed | |
| - SPECIATION_EVENT: only on unresolved cross-domain tension | |
| - Always populate ui_directives (camera_focus_node_id, pulses, threads) | |
| - All node_id and edge_id values must be unique strings | |
| """ | |
| def _img_to_data_uri(img: Image.Image) -> str: | |
| buf = io.BytesIO() | |
| img.convert("RGB").save(buf, format="JPEG", quality=88) | |
| return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode() | |
| def build_messages(user_text: str, image: Optional[Image.Image], hg_context: str = ""): | |
| user_content = [] | |
| if image is not None: | |
| user_content.append({"type": "image_url", "image_url": {"url": _img_to_data_uri(image)}}) | |
| ctx = f"\n\n[Hypergraph context]\n{hg_context}" if hg_context else "" | |
| user_content.append({"type": "text", "text": user_text + ctx}) | |
| return [ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": user_content}, | |
| ] | |
| def new_session_meta(): | |
| return { | |
| "session_id": str(uuid.uuid4()), | |
| "timestamp_utc": datetime.datetime.utcnow().isoformat() + "Z", | |
| } | |