diff --git a/devplacepy/services/devii/session/__init__.py b/devplacepy/services/devii/session/__init__.py new file mode 100644 index 00000000..c74cb3b7 --- /dev/null +++ b/devplacepy/services/devii/session/__init__.py @@ -0,0 +1,29 @@ +# retoor + +from __future__ import annotations + +from .core import DeviiSession +from .prompts import ( + BEHAVIOR_HEADER, + DOCS_GREETING, + DOCS_SYSTEM_PROMPT, + DOCS_TOOLS, + LOGIN_REQUEST, + NON_ADMIN_COST_RULE, + TOPIC_CLASSIFIER_PROMPT, + _parse_topic, + _system_prompt_for, +) + +__all__ = [ + "DeviiSession", + "BEHAVIOR_HEADER", + "DOCS_GREETING", + "DOCS_SYSTEM_PROMPT", + "DOCS_TOOLS", + "LOGIN_REQUEST", + "NON_ADMIN_COST_RULE", + "TOPIC_CLASSIFIER_PROMPT", + "_parse_topic", + "_system_prompt_for", +] diff --git a/devplacepy/services/devii/session/_helpers.py b/devplacepy/services/devii/session/_helpers.py new file mode 100644 index 00000000..e4342687 --- /dev/null +++ b/devplacepy/services/devii/session/_helpers.py @@ -0,0 +1,45 @@ +# retoor + +from __future__ import annotations + +from typing import Any + + +def _repair_history(messages: list[dict[str, Any]]) -> None: + last_idx = None + for i in range(len(messages) - 1, -1, -1): + message = messages[i] + if message.get("role") == "assistant" and message.get("tool_calls"): + last_idx = i + break + if last_idx is None: + while messages and messages[-1].get("role") == "tool": + messages.pop() + return + declared = { + call.get("id") + for call in (messages[last_idx].get("tool_calls") or []) + if call.get("id") + } + answered = { + messages[j].get("tool_call_id") + for j in range(last_idx + 1, len(messages)) + if messages[j].get("role") == "tool" + } + if not declared or not declared.issubset(answered): + del messages[last_idx:] + + +def _now_iso() -> str: + from datetime import datetime, timezone + + return datetime.now(timezone.utc).isoformat() + + +def _format_offset(offset: Any) -> str: + if offset is None: + return "+00:00" + total_minutes = int(offset.total_seconds() // 60) + sign = "+" if total_minutes >= 0 else "-" + total_minutes = abs(total_minutes) + return f"{sign}{total_minutes // 60:02d}:{total_minutes % 60:02d}" diff --git a/devplacepy/services/devii/session.py b/devplacepy/services/devii/session/core.py similarity index 80% rename from devplacepy/services/devii/session.py rename to devplacepy/services/devii/session/core.py index a81be1c7..a56f3a7b 100644 --- a/devplacepy/services/devii/session.py +++ b/devplacepy/services/devii/session/core.py @@ -8,21 +8,32 @@ from typing import Any import uuid_utils -from .actions import Dispatcher -from .agent import Agent, SYSTEM_PROMPT -from .agentic import AgenticController, LessonStore -from .avatar import AvatarController -from .behavior import BehaviorController -from .chunks import ChunkStore -from .client import ClientController -from .config import Settings -from .cost import CostTracker -from .cost.tracker import Pricing -from .http_client import PlatformClient -from .llm import LLMClient -from .registry import CATALOG -from .tasks import Scheduler, TaskController, TaskStore -from .virtual_tools import VirtualToolController, VirtualToolStore +from ..actions import Dispatcher +from ..agent import Agent +from ..agentic import AgenticController, LessonStore +from ..avatar import AvatarController +from ..behavior import BehaviorController +from ..chunks import ChunkStore +from ..client import ClientController +from ..config import Settings +from ..cost import CostTracker +from ..cost.tracker import Pricing +from ..http_client import PlatformClient +from ..llm import LLMClient +from ..registry import CATALOG +from ..tasks import Scheduler, TaskController, TaskStore +from ..virtual_tools import VirtualToolController, VirtualToolStore +from ._helpers import _format_offset, _now_iso, _repair_history +from .prompts import ( + BEHAVIOR_HEADER, + DOCS_GREETING, + DOCS_SYSTEM_PROMPT, + DOCS_TOOLS, + LOGIN_REQUEST, + TOPIC_CLASSIFIER_PROMPT, + _parse_topic, + _system_prompt_for, +) logger = logging.getLogger("devii.session") @@ -37,37 +48,6 @@ INTERNAL_PREFIXES = ( "Protocol violation:", ) BUFFERED_TYPES = ("task", "error", "reply", "status") -BEHAVIOR_HEADER = "# TRUTH RULES AND BEHAVIOR" - - -def _repair_history(messages: list[dict[str, Any]]) -> None: - last_idx = None - for i in range(len(messages) - 1, -1, -1): - message = messages[i] - if message.get("role") == "assistant" and message.get("tool_calls"): - last_idx = i - break - if last_idx is None: - while messages and messages[-1].get("role") == "tool": - messages.pop() - return - declared = { - call.get("id") - for call in (messages[last_idx].get("tool_calls") or []) - if call.get("id") - } - answered = { - messages[j].get("tool_call_id") - for j in range(last_idx + 1, len(messages)) - if messages[j].get("role") == "tool" - } - if not declared or not declared.issubset(answered): - del messages[last_idx:] - -LOGIN_REQUEST = ( - "Welcome to Devii. Ask me anything - I can generate clients and bots, search the docs, " - "fetch pages, and more. Sign in to DevPlace whenever you want me to work on your own account." -) class DeviiSession: @@ -849,114 +829,3 @@ class DeviiSession: self.owner_kind, self.owner_id, ) - - -NON_ADMIN_COST_RULE = ( - "\n\nThe current user is NOT an administrator. Never reveal any monetary cost, dollar " - "amount, pricing, or spend figure to them under any circumstance. When asked about " - "usage or resources, call usage_quota and report only the percentage of their 24h " - "quota used and the turn count." -) - -DOCS_TOOLS = frozenset({"search_docs"}) - -TOPIC_CLASSIFIER_PROMPT = ( - "You are a routing classifier for a documentation assistant. Decide whether the " - "user's NEW message continues the PRIOR conversation (a follow-up: a clarification, " - "a refinement, or another question about the same subject) or starts a NEW, " - "unrelated topic that should begin from a clean slate.\n" - "Respond with ONLY a JSON object and nothing else: " - '{"decision": "follow_up" or "new", "reason": ""}.' -) - -DOCS_GREETING = ( - "Hi, I am Docii, the DevPlace documentation assistant. Ask me anything about " - "DevPlace and I will search the documentation to find your answer." -) - -DOCS_SYSTEM_PROMPT = ( - "You are Docii, the documentation assistant for DevPlace, a social network for " - "software developers. You answer questions about DevPlace using ONLY its official " - "documentation.\n\n" - "YOU HAVE EXACTLY ONE TOOL: search_docs(query, max_results). It performs a keyword " - "search over the documentation and returns the matching pages. Each result is an object " - 'with "title", "url" (e.g. /docs/authentication.html), "score" (BM25 relevance, higher ' - 'means more relevant), and "content" (the page text).\n\n' - "HARD RULES - follow every one, every turn:\n" - "1. Ground every CLAIM ABOUT DEVPLACE (routes, endpoints, parameters, auth methods, " - "settings, behavior, limits) in content returned by search_docs in THIS turn. Never state " - "a platform fact from memory or assumption.\n" - "2. For EVERY question, your FIRST action is to call search_docs with focused keywords " - "drawn from the question (feature names, endpoint paths, nouns).\n" - "3. READ the content in the results. If they do not fully and confidently answer the " - "question, call search_docs AGAIN with different or more specific keywords (synonyms, " - "related terms, the exact feature or route). Keep searching recursively - visiting the " - "results, refining the query, searching again - until you have gathered enough grounded " - "information to answer. Prefer several targeted searches over one broad search.\n" - "4. Do not invent routes, endpoints, parameters, settings, or behavior that the docs did " - "not state. The platform-specific facts in your answer must come from the docs.\n" - "5. If, after several distinct searches, the documentation genuinely does not cover a " - "needed FACT, say which part is undocumented - but still help as far as the docs allow " - "(for example, write the code using the endpoints you DID find).\n" - "6. Your only tool is search_docs and you perform no platform actions (no posting, " - "account changes, file edits, or web browsing). Writing code and examples in your reply " - "is allowed and encouraged.\n\n" - "WRITING CODE - when the user asks for a script, code example, API client, bot, or " - "snippet in ANY language, WRITE complete, runnable example code. Do NOT refuse merely " - "because the literal source is not published in the docs - synthesize it from the " - "documented API: take the base URL, endpoints, HTTP methods, parameters and " - "authentication from your search_docs results, and write all the ordinary programming " - "scaffolding (imports, language syntax, error handling, a main loop, comments) yourself. " - "Put it in a fenced code block with a language tag. Never invent endpoints or parameters " - "the docs do not describe; if a detail you need is missing, search for it first.\n\n" - "LINKING - this is mandatory:\n" - "- ACTIVE INLINE LINKING: whenever you mention a documented page, feature, endpoint, or " - "concept in your prose, write it as a markdown link to that page's `url`, for example " - "[authentication](/docs/authentication.html). Use the exact `url` from the search " - "results. Never paste a bare URL - always a markdown link with descriptive text.\n" - "- ALWAYS end every answer with a '## References' section: a markdown bullet list of the " - "documentation pages you actually used to answer, each as a clickable markdown link " - "followed by its relevance score, highest score first, e.g. " - "`- [Authentication](/docs/authentication.html) - score 12.3`. Include only pages you " - "used; do not invent links or scores.\n\n" - "STYLE: concise, practical, technical. Use markdown." -) - - -def _parse_topic(raw: str) -> tuple[str, str]: - import json - import re - - match = re.search(r"\{.*\}", raw or "", re.S) - if not match: - return "follow_up", "" - try: - data = json.loads(match.group()) - except (ValueError, TypeError): - return "follow_up", "" - decision = ( - "new" if str(data.get("decision", "")).strip().lower() == "new" else "follow_up" - ) - reason = str(data.get("reason", "")).strip()[:200] - return decision, reason - - -def _system_prompt_for(is_admin: bool) -> str: - if is_admin: - return SYSTEM_PROMPT - return SYSTEM_PROMPT + NON_ADMIN_COST_RULE - - -def _now_iso() -> str: - from datetime import datetime, timezone - - return datetime.now(timezone.utc).isoformat() - - -def _format_offset(offset: Any) -> str: - if offset is None: - return "+00:00" - total_minutes = int(offset.total_seconds() // 60) - sign = "+" if total_minutes >= 0 else "-" - total_minutes = abs(total_minutes) - return f"{sign}{total_minutes // 60:02d}:{total_minutes % 60:02d}" diff --git a/devplacepy/services/devii/session/prompts.py b/devplacepy/services/devii/session/prompts.py new file mode 100644 index 00000000..8ded5276 --- /dev/null +++ b/devplacepy/services/devii/session/prompts.py @@ -0,0 +1,107 @@ +# retoor + +from __future__ import annotations + +from ..agent import SYSTEM_PROMPT + +BEHAVIOR_HEADER = "# TRUTH RULES AND BEHAVIOR" + +LOGIN_REQUEST = ( + "Welcome to Devii. Ask me anything - I can generate clients and bots, search the docs, " + "fetch pages, and more. Sign in to DevPlace whenever you want me to work on your own account." +) + +NON_ADMIN_COST_RULE = ( + "\n\nThe current user is NOT an administrator. Never reveal any monetary cost, dollar " + "amount, pricing, or spend figure to them under any circumstance. When asked about " + "usage or resources, call usage_quota and report only the percentage of their 24h " + "quota used and the turn count." +) + +DOCS_TOOLS = frozenset({"search_docs"}) + +TOPIC_CLASSIFIER_PROMPT = ( + "You are a routing classifier for a documentation assistant. Decide whether the " + "user's NEW message continues the PRIOR conversation (a follow-up: a clarification, " + "a refinement, or another question about the same subject) or starts a NEW, " + "unrelated topic that should begin from a clean slate.\n" + "Respond with ONLY a JSON object and nothing else: " + '{"decision": "follow_up" or "new", "reason": ""}.' +) + +DOCS_GREETING = ( + "Hi, I am Docii, the DevPlace documentation assistant. Ask me anything about " + "DevPlace and I will search the documentation to find your answer." +) + +DOCS_SYSTEM_PROMPT = ( + "You are Docii, the documentation assistant for DevPlace, a social network for " + "software developers. You answer questions about DevPlace using ONLY its official " + "documentation.\n\n" + "YOU HAVE EXACTLY ONE TOOL: search_docs(query, max_results). It performs a keyword " + "search over the documentation and returns the matching pages. Each result is an object " + 'with "title", "url" (e.g. /docs/authentication.html), "score" (BM25 relevance, higher ' + 'means more relevant), and "content" (the page text).\n\n' + "HARD RULES - follow every one, every turn:\n" + "1. Ground every CLAIM ABOUT DEVPLACE (routes, endpoints, parameters, auth methods, " + "settings, behavior, limits) in content returned by search_docs in THIS turn. Never state " + "a platform fact from memory or assumption.\n" + "2. For EVERY question, your FIRST action is to call search_docs with focused keywords " + "drawn from the question (feature names, endpoint paths, nouns).\n" + "3. READ the content in the results. If they do not fully and confidently answer the " + "question, call search_docs AGAIN with different or more specific keywords (synonyms, " + "related terms, the exact feature or route). Keep searching recursively - visiting the " + "results, refining the query, searching again - until you have gathered enough grounded " + "information to answer. Prefer several targeted searches over one broad search.\n" + "4. Do not invent routes, endpoints, parameters, settings, or behavior that the docs did " + "not state. The platform-specific facts in your answer must come from the docs.\n" + "5. If, after several distinct searches, the documentation genuinely does not cover a " + "needed FACT, say which part is undocumented - but still help as far as the docs allow " + "(for example, write the code using the endpoints you DID find).\n" + "6. Your only tool is search_docs and you perform no platform actions (no posting, " + "account changes, file edits, or web browsing). Writing code and examples in your reply " + "is allowed and encouraged.\n\n" + "WRITING CODE - when the user asks for a script, code example, API client, bot, or " + "snippet in ANY language, WRITE complete, runnable example code. Do NOT refuse merely " + "because the literal source is not published in the docs - synthesize it from the " + "documented API: take the base URL, endpoints, HTTP methods, parameters and " + "authentication from your search_docs results, and write all the ordinary programming " + "scaffolding (imports, language syntax, error handling, a main loop, comments) yourself. " + "Put it in a fenced code block with a language tag. Never invent endpoints or parameters " + "the docs do not describe; if a detail you need is missing, search for it first.\n\n" + "LINKING - this is mandatory:\n" + "- ACTIVE INLINE LINKING: whenever you mention a documented page, feature, endpoint, or " + "concept in your prose, write it as a markdown link to that page's `url`, for example " + "[authentication](/docs/authentication.html). Use the exact `url` from the search " + "results. Never paste a bare URL - always a markdown link with descriptive text.\n" + "- ALWAYS end every answer with a '## References' section: a markdown bullet list of the " + "documentation pages you actually used to answer, each as a clickable markdown link " + "followed by its relevance score, highest score first, e.g. " + "`- [Authentication](/docs/authentication.html) - score 12.3`. Include only pages you " + "used; do not invent links or scores.\n\n" + "STYLE: concise, practical, technical. Use markdown." +) + + +def _parse_topic(raw: str) -> tuple[str, str]: + import json + import re + + match = re.search(r"\{.*\}", raw or "", re.S) + if not match: + return "follow_up", "" + try: + data = json.loads(match.group()) + except (ValueError, TypeError): + return "follow_up", "" + decision = ( + "new" if str(data.get("decision", "")).strip().lower() == "new" else "follow_up" + ) + reason = str(data.get("reason", "")).strip()[:200] + return decision, reason + + +def _system_prompt_for(is_admin: bool) -> str: + if is_admin: + return SYSTEM_PROMPT + return SYSTEM_PROMPT + NON_ADMIN_COST_RULE