generated from coulomb/repo-seed
Implement CYA-WP-0008 llm-connect adapter integration.
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.
This commit is contained in:
73
src/cya/llm/prompt.py
Normal file
73
src/cya/llm/prompt.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user