Add dual-prefix AI command

app/ai/__init__.py

@@ -0,0 +1,9 @@
+from app.ai.parser import AIInvocation, parse_ai_invocation
+from app.ai.service import AIResponse, AIService
+
+__all__ = [
+    "AIInvocation",
+    "AIResponse",
+    "AIService",
+    "parse_ai_invocation",
+]

app/ai/parser.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import shlex
+
+
+@dataclass(slots=True)
+class AIInvocation:
+    context_messages: int
+    model_flags: list[str]
+    prompt: str
+
+
+def parse_ai_invocation(text: str, *, default_context_messages: int) -> AIInvocation:
+    try:
+        tokens = shlex.split(text)
+    except ValueError:
+        tokens = text.split()
+
+    context_messages = default_context_messages
+    model_flags: list[str] = []
+
+    if tokens:
+        first = tokens[0]
+        if first.isdigit():
+            context_messages = int(tokens.pop(0))
+
+    while tokens and tokens[0].startswith("--"):
+        flag = tokens.pop(0)[2:]
+        if "=" in flag:
+            flag = flag.split("=", 1)[0]
+        flag = flag.strip().lower()
+        if flag:
+            model_flags.append(flag)
+
+    prompt = " ".join(tokens).strip()
+    if not prompt:
+        raise ValueError("Usage: ..ai [N] [--model] prompt")
+
+    return AIInvocation(
+        context_messages=context_messages,
+        model_flags=model_flags,
+        prompt=prompt,
+    )

app/ai/prompts.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+
+DEFAULT_SYSTEM_PROMPT = "You are a concise, accurate Telegram assistant."
+
+
+@dataclass(slots=True)
+class PromptBook:
+    root: Path
+    system_file: str = "system.md"
+    parts_dir: str = "parts"
+
+    def _read_file(self, path: Path) -> str:
+        return path.read_text(encoding="utf-8").strip()
+
+    def load_system_prompt(self) -> str:
+        sections: list[str] = []
+        system_path = self.root / self.system_file
+        if system_path.exists():
+            text = self._read_file(system_path)
+            if text:
+                sections.append(text)
+
+        parts_root = self.root / self.parts_dir
+        if parts_root.exists():
+            for path in sorted(parts_root.rglob("*.md")):
+                if path.is_file():
+                    text = self._read_file(path)
+                    if text:
+                        sections.append(text)
+
+        if not sections:
+            return DEFAULT_SYSTEM_PROMPT
+        return "\n\n".join(sections).strip()

app/ai/service.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+from typing import TYPE_CHECKING, Any
+
+from openai import AsyncOpenAI
+
+from app.ai.parser import AIInvocation
+from app.ai.prompts import PromptBook
+from app.config import AIConfig, AIModelConfig, AIProviderConfig
+
+if TYPE_CHECKING:
+    from app.commands.core import CommandContext
+
+
+log = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class AIResponse:
+    text: str
+    model: AIModelConfig
+    provider: AIProviderConfig
+
+
+@dataclass(slots=True)
+class ResolvedAIModel:
+    alias: str
+    provider: AIProviderConfig
+    model: AIModelConfig
+
+
+class AIService:
+    def __init__(self, config: AIConfig) -> None:
+        self.config = config
+        self.prompt_book = PromptBook(config.prompt.root)
+        self._clients: dict[str, AsyncOpenAI] = {}
+        self._system_prompt: str | None = None
+
+    @property
+    def enabled(self) -> bool:
+        return True
+
+    def available_model_aliases(self) -> list[str]:
+        return sorted(self.config.models.keys())
+
+    def describe_models(self) -> str:
+        lines = [f"default={self.config.default_model}"]
+        for alias in self.available_model_aliases():
+            model = self.config.models[alias]
+            provider = self.config.providers[model.provider]
+            lines.append(f"{alias} -> {provider.name}:{model.model}")
+        return "; ".join(lines)
+
+    def select_model_alias(self, model_flags: list[str]) -> str:
+        for flag in model_flags:
+            if flag in self.config.models:
+                return flag
+        if model_flags:
+            raise ValueError(
+                "Unknown model flag: " + ", ".join(f"--{flag}" for flag in model_flags)
+            )
+        return self.config.default_model
+
+    def resolve_model(self, alias: str | None) -> ResolvedAIModel:
+        selected_alias = alias or self.config.default_model
+        model = self.config.models.get(selected_alias)
+        if model is None:
+            raise ValueError(f"Unknown AI model alias: {selected_alias}")
+        provider = self.config.providers.get(model.provider)
+        if provider is None:
+            raise ValueError(
+                f"AI model {selected_alias} references unknown provider: {model.provider}"
+            )
+        return ResolvedAIModel(alias=selected_alias, provider=provider, model=model)
+
+    def _get_client(self, provider: AIProviderConfig) -> AsyncOpenAI:
+        client = self._clients.get(provider.name)
+        if client is None:
+            client = AsyncOpenAI(
+                api_key=provider.api_key,
+                base_url=provider.base_url,
+                timeout=provider.timeout,
+            )
+            self._clients[provider.name] = client
+        return client
+
+    def load_system_prompt(self) -> str:
+        if self._system_prompt is None:
+            self._system_prompt = self.prompt_book.load_system_prompt()
+        return self._system_prompt
+
+    async def build_messages(
+        self,
+        ctx: "CommandContext",
+        *,
+        prompt: str,
+        context_messages: int,
+    ) -> list[dict[str, str]]:
+        messages: list[dict[str, str]] = [{"role": "system", "content": self.load_system_prompt()}]
+        history = await self.collect_history(ctx, limit=context_messages)
+        messages.extend(history)
+        messages.append({"role": "user", "content": prompt})
+        return messages
+
+    async def collect_history(self, ctx: "CommandContext", *, limit: int) -> list[dict[str, str]]:
+        if limit <= 0:
+            return []
+
+        collected: list[dict[str, str]] = []
+        current_message_id = getattr(ctx.message, "id", 0)
+        owner_id = ctx.owner_id
+
+        async for message in ctx.client.get_chat_history(
+            ctx.message.chat.id,
+            limit=limit + 5,
+            offset_id=current_message_id,
+        ):
+            message_id = getattr(message, "id", 0)
+            if message_id >= current_message_id:
+                continue
+
+            text = self._extract_message_text(message)
+            if not text:
+                continue
+
+            role = "assistant" if self._is_own_message(message, owner_id) else "user"
+            collected.append({"role": role, "content": text})
+            if len(collected) >= limit:
+                break
+
+        collected.reverse()
+        return collected
+
+    def _extract_message_text(self, message: Any) -> str:
+        text = getattr(message, "text", None) or getattr(message, "caption", None)
+        if not isinstance(text, str):
+            return ""
+        return text.strip()
+
+    def _is_own_message(self, message: Any, owner_id: int | None) -> bool:
+        user = getattr(message, "from_user", None)
+        user_id = getattr(user, "id", None)
+        if owner_id is None:
+            return bool(getattr(message, "outgoing", False))
+        return user_id == owner_id or bool(getattr(message, "outgoing", False))
+
+    async def generate(self, ctx: "CommandContext", invocation: AIInvocation) -> AIResponse:
+        model = self.resolve_model(invocation.model_flags[0] if invocation.model_flags else None)
+        client = self._get_client(model.provider)
+
+        messages = await self.build_messages(
+            ctx,
+            prompt=invocation.prompt,
+            context_messages=invocation.context_messages,
+        )
+
+        try:
+            response = await client.chat.completions.create(
+                model=model.model.model,
+                messages=messages,
+            )
+        except Exception as exc:
+            log.exception("AI request failed for model %s", model.alias)
+            raise RuntimeError(f"AI request failed: {exc}") from exc
+
+        text = self._extract_response_text(response)
+        if not text:
+            raise RuntimeError("AI model returned an empty response")
+
+        return AIResponse(text=text, model=model.model, provider=model.provider)
+
+    def _extract_response_text(self, response: Any) -> str:
+        choices = getattr(response, "choices", None) or []
+        if not choices:
+            return ""
+        choice = choices[0]
+        message = getattr(choice, "message", None)
+        if message is None:
+            return ""
+        content = getattr(message, "content", None)
+        if isinstance(content, str):
+            return content.strip()
+        if isinstance(content, list):
+            parts = []
+            for part in content:
+                text = getattr(part, "text", None)
+                if isinstance(text, str):
+                    parts.append(text)
+            return "\n".join(parts).strip()
+        return ""