generated from coulomb/repo-seed
tegwick
019f6e7dc7
Wire LLMConnectAdapter behind the existing LLMAdapter seam with config-driven selection, graceful degradation, --offline mode, and bounded session context. Add unit tests, integration docs, and update README/SCOPE/AGENTS.
73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
"""Prompt construction for llm-connect delegation (CYA-WP-0008)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from cya.llm.adapter import AssistanceRequest
|
|
|
|
_DEFAULT_SYSTEM = """You are cya, a console-native assistant for practical local work from the shell.
|
|
|
|
Help the user with command-line tasks: repository inspection, file workflows, command
|
|
suggestion, command explanation, and local context summarization.
|
|
|
|
Be concise and practical. When suggesting shell commands, explain risks briefly.
|
|
Do not claim to have executed anything — the user runs commands themselves.
|
|
Reference the provided context when it is relevant."""
|
|
|
|
|
|
def default_system_prompt() -> str:
|
|
return _DEFAULT_SYSTEM
|
|
|
|
|
|
def build_assistance_prompt(request: AssistanceRequest, *, system_prompt: str | None = None) -> tuple[str, str]:
|
|
"""Return (system_prompt, user_prompt) for llm-connect execute_prompt."""
|
|
system = system_prompt or default_system_prompt()
|
|
parts: list[str] = []
|
|
|
|
context = request.context or {}
|
|
session_turns = context.get("session_turns")
|
|
if session_turns:
|
|
parts.append("## Recent conversation")
|
|
for turn in session_turns:
|
|
parts.append(f"User: {turn.get('user', '')}")
|
|
parts.append(f"Assistant: {turn.get('assistant', '')}")
|
|
|
|
envelope = {k: v for k, v in context.items() if k not in ("session_turns", "memory")}
|
|
if envelope:
|
|
parts.append("## Local context")
|
|
parts.append(_summarize_context(envelope))
|
|
|
|
memory = context.get("memory")
|
|
if isinstance(memory, dict) and memory.get("items"):
|
|
parts.append("## Activated memory")
|
|
for item in memory["items"][:8]:
|
|
parts.append(f"- [{item.get('kind', '?')}] {item.get('key', '?')}: {item.get('value', '')}")
|
|
|
|
parts.append("## Current request")
|
|
parts.append(request.user_request.strip())
|
|
|
|
return system, "\n\n".join(parts)
|
|
|
|
|
|
def _summarize_context(envelope: dict[str, Any]) -> str:
|
|
"""Compact, JSON-safe context summary to stay within prompt budget."""
|
|
summary: dict[str, Any] = {}
|
|
if envelope.get("cwd"):
|
|
summary["cwd"] = envelope["cwd"]
|
|
if envelope.get("git"):
|
|
git = envelope["git"]
|
|
summary["git"] = {
|
|
k: git[k]
|
|
for k in ("branch", "status_short", "workdir", "is_repo")
|
|
if k in git
|
|
}
|
|
if envelope.get("top_level"):
|
|
names = [e.get("name") for e in envelope["top_level"][:30] if e.get("name")]
|
|
summary["top_level"] = names
|
|
if envelope.get("env"):
|
|
summary["env"] = envelope["env"]
|
|
if envelope.get("notes"):
|
|
summary["notes"] = envelope["notes"][:5]
|
|
return json.dumps(summary, indent=2, default=str) |