retoor/devplacepy
1
0
Fork 1
Code Issues 44 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: add multi-channel Devii sessions with docs search mode and channel-aware conversation persistence

Browse Source
This commit is contained in:
retoor
2026-06-13 21:37:13 +00:00
parent 873186b274
commit 61c1ae8c5d
25 changed files with 1250 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
devplacepy/services/devii/actions/dispatcher.py
+1 -1
View File
@@ -248,7 +248,7 @@ class Dispatcher:
self._owner_kind = owner_kind
self._owner_id = owner_id
self._fetch = FetchController(settings)
self._docs = DocsController(settings)
self._docs = DocsController(settings, is_admin=is_admin)
self._cost = CostController(quota_provider=quota_provider)
self._chunks = ChunkController(settings)
self._rsearch = RsearchController(settings)
devplacepy/services/devii/docs/controller.py
+13 -81
View File
@@ -2,81 +2,32 @@
from __future__ import annotations
import asyncio
import json
import logging
import re
from typing import Any
import httpx
from ..agentic.lessons import tokenize
from ..config import Settings
from ..errors import NetworkError, ToolInputError
from ..fetch.controller import STEALTH_HEADERS
from ..text import truncate
from ..errors import ToolInputError
logger = logging.getLogger("devii.docs")
DOCS_PATH = "/docs/download.md"
HEADING = re.compile(r"^(#{1,3})\s+(.*)$")
DEFAULT_MAX_RESULTS = 5
SECTION_CHARS = 1800
HEADING_WEIGHT = 3
CONTENT_CHARS = 2400
class DocsController:
def __init__(self, settings: Settings) -> None:
def __init__(self, settings: Settings, is_admin: bool = False) -> None:
self._settings = settings
self._sections: list[tuple[str, str]] | None = None
self._lock = asyncio.Lock()
self._is_admin = is_admin
async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
if name != "search_docs":
raise ToolInputError(f"Unknown docs tool: {name}")
return await self._search(arguments)
async def _load(self) -> list[tuple[str, str]]:
async with self._lock:
if self._sections is not None:
return self._sections
url = self._settings.base_url + DOCS_PATH
try:
async with httpx.AsyncClient(
headers=STEALTH_HEADERS,
follow_redirects=True,
timeout=self._settings.fetch_timeout_seconds,
) as client:
response = await client.get(url)
response.raise_for_status()
markdown = response.text
except httpx.HTTPError as exc:
raise NetworkError(
f"Could not load documentation: {exc}", url=url
) from exc
self._sections = self._split(markdown)
logger.info("Loaded %d documentation sections", len(self._sections))
return self._sections
@staticmethod
def _split(markdown: str) -> list[tuple[str, str]]:
sections: list[tuple[str, str]] = []
heading = "Overview"
body: list[str] = []
for line in markdown.splitlines():
match = HEADING.match(line)
if match:
if body:
sections.append((heading, "\n".join(body).strip()))
body = []
heading = match.group(2).strip()
else:
body.append(line)
if body:
sections.append((heading, "\n".join(body).strip()))
return [(title, text) for title, text in sections if text]
async def _search(self, arguments: dict[str, Any]) -> str:
from devplacepy import docs_search
query = str(arguments.get("query", "")).strip()
if not query:
raise ToolInputError("search_docs requires a query.")
@@ -85,32 +36,13 @@ class DocsController:
)
max_results = max(1, min(max_results, 10))
sections = await self._load()
terms = tokenize(query)
scored: list[tuple[float, str, str]] = []
for heading, text in sections:
heading_tokens = set(tokenize(heading))
body_tokens = tokenize(text)
body_counts: dict[str, int] = {}
for token in body_tokens:
body_counts[token] = body_counts.get(token, 0) + 1
score = 0.0
for term in terms:
score += body_counts.get(term, 0)
if term in heading_tokens:
score += HEADING_WEIGHT
if score > 0:
scored.append((score, heading, text))
scored.sort(key=lambda item: item[0], reverse=True)
results = [
{
"title": heading,
"score": round(score, 2),
"content": truncate(text, SECTION_CHARS),
}
for score, heading, text in scored[:max_results]
]
results = docs_search.search_pages(
query,
user=None,
is_admin=self._is_admin,
limit=max_results,
content_chars=CONTENT_CHARS,
)
return json.dumps(
{
"status": "success",
devplacepy/services/devii/hub.py
+16 -7
View File
@@ -30,7 +30,7 @@ class DeviiHub:
"ledger": UsageLedger(),
"turns": TurnAudit(),
}
self._sessions: dict[tuple[str, str], DeviiSession] = {}
self._sessions: dict[tuple[str, str, str], DeviiSession] = {}
@property
def ledger(self) -> UsageLedger:
@@ -44,15 +44,20 @@ class DeviiHub:
api_key: str,
base_url: str,
is_admin: bool = False,
channel: str = "main",
) -> DeviiSession:
key = (owner_kind, owner_id)
key = (owner_kind, owner_id, channel)
session = self._sessions.get(key)
if session is not None:
return session
settings, pricing = self._build(api_key, base_url, owner_kind, is_admin)
llm = LLMClient(settings)
# Persistent, owner-isolated stores for signed-in users; ephemeral in-memory for guests.
owned_db = db if owner_kind == "user" else memory_db()
# The `docs` channel (Docii) is a self-contained documentation assistant: it gets its own
# ephemeral stores so Devii's tasks, lessons, behavior and virtual tools never leak into it
# (and a Docii reflection never pollutes the user's Devii memory). Only the conversation
# thread persists, keyed per channel in the shared ConversationStore.
owned_db = db if (owner_kind == "user" and channel == "main") else memory_db()
task_store = TaskStore(owned_db, owner_kind, owner_id)
lessons = LessonStore(owned_db, owner_kind, owner_id)
virtual_tool_store = VirtualToolStore(owned_db, owner_kind, owner_id)
@@ -70,22 +75,26 @@ class DeviiHub:
behavior_store,
self._stores,
is_admin=is_admin,
channel=channel,
)
if owner_kind == "user":
saved = self._stores["conversations"].load(owner_kind, owner_id)
saved = self._stores["conversations"].load(owner_kind, owner_id, channel)
if saved:
session.restore_history(saved)
self._sessions[key] = session
logger.info(
"Created session %s/%s (total %d)",
"Created session %s/%s [%s] (total %d)",
owner_kind,
owner_id,
channel,
len(self._sessions),
)
return session
def find(self, owner_kind: str, owner_id: str) -> DeviiSession | None:
return self._sessions.get((owner_kind, owner_id))
def find(
self, owner_kind: str, owner_id: str, channel: str = "main"
) -> DeviiSession | None:
return self._sessions.get((owner_kind, owner_id, channel))
def active_sessions(self) -> int:
return len(self._sessions)
devplacepy/services/devii/llm.py
+24
View File
@@ -71,6 +71,30 @@ class LLMClient:
)
return message
async def complete_text(
self, messages: list[dict[str, Any]], temperature: float = 0.0
) -> str:
payload = {
"model": self._settings.ai_model,
"messages": messages,
"temperature": temperature,
}
try:
response = await self._client.post(self._settings.ai_url, json=payload)
except httpx.HTTPError as exc:
raise LLMError(f"Could not reach the model endpoint: {exc}") from exc
if response.status_code >= 400:
raise LLMError(
f"Model endpoint returned {response.status_code}: {self._reason(response)}"
)
try:
data = response.json()
content = data["choices"][0]["message"]["content"] or ""
except (ValueError, KeyError, IndexError) as exc:
raise LLMError("Model endpoint returned an unexpected response.") from exc
record_usage(data.get("usage"))
return content
async def summarize(self, text: str) -> str:
payload = {
"model": self._settings.ai_model,
devplacepy/services/devii/session.py
+170 -8
View File
@@ -85,9 +85,11 @@ class DeviiSession:
behavior_store: Any,
stores: dict[str, Any],
is_admin: bool = False,
channel: str = "main",
) -> None:
self.owner_kind = owner_kind
self.owner_id = owner_id
self.channel = channel
self.username = username
self.settings = settings
self.is_admin = is_admin
@@ -127,8 +129,10 @@ class DeviiSession:
virtual_tools=self.virtual_tools,
behavior=self.behavior,
)
self.tools = CATALOG.tool_schemas_for(self.client.authenticated, is_admin)
self._system_prompt = _system_prompt_for(is_admin)
self.tools = self._builtin_tools()
self._system_prompt = (
DOCS_SYSTEM_PROMPT if channel == "docs" else _system_prompt_for(is_admin)
)
self.agentic.bind(
llm=llm,
dispatcher=self.dispatcher,
@@ -224,7 +228,7 @@ class DeviiSession:
self._disconnected.clear()
self.avatar.bind(self._avatar_request)
self.browser.bind(self._client_request)
if not self._started:
if not self._started and self.channel == "main":
self.scheduler.start()
self._started = True
if self._buffer:
@@ -282,6 +286,8 @@ class DeviiSession:
await self._llm.aclose()
async def bootstrap_greeting(self) -> str:
if self.channel == "docs":
return DOCS_GREETING
if self.client.authenticated:
return f"Signed in as {self.username}. Devii is operating your DevPlace account. How can I help?"
return LOGIN_REQUEST
@@ -307,7 +313,7 @@ class DeviiSession:
self._buffer = []
if self.persist_conversation:
try:
self._conv.clear(self.owner_kind, self.owner_id)
self._conv.clear(self.owner_kind, self.owner_id, self.channel)
except Exception: # noqa: BLE001 - clearing storage must not break the socket
logger.exception(
"Failed to clear conversation for %s/%s",
@@ -354,7 +360,10 @@ class DeviiSession:
if self.persist_conversation:
try:
self._conv.save(
self.owner_kind, self.owner_id, self.agent._messages
self.owner_kind,
self.owner_id,
self.agent._messages,
self.channel,
)
except Exception: # noqa: BLE001 - persistence must not break the socket
logger.exception(
@@ -376,6 +385,8 @@ class DeviiSession:
async with self._lock:
self._refresh_tools()
self._refresh_system_prompt()
if self.channel == "docs":
await self._docs_topic_gate(text)
reply = await self.agent.respond(text)
if epoch == self._turn_epoch:
await self._emit({"type": "reply", "text": reply}, buffer=True)
@@ -391,12 +402,26 @@ class DeviiSession:
if not cancelled and epoch == self._turn_epoch:
self._record_turn(turn_id, started_at, text, reply, error, before)
def _builtin_tools(self) -> list[dict[str, Any]]:
schemas = CATALOG.tool_schemas_for(self.client.authenticated, self.is_admin)
if self.channel == "docs":
return [
s
for s in schemas
if s.get("function", {}).get("name") in DOCS_TOOLS
]
return schemas
def _refresh_tools(self) -> None:
builtin = CATALOG.tool_schemas_for(self.client.authenticated, self.is_admin)
if self.channel == "docs":
self.tools[:] = self._builtin_tools()
return
virtual = self._virtual_tool_store.tool_schemas()
self.tools[:] = builtin + virtual
self.tools[:] = self._builtin_tools() + virtual
def _compose_system_prompt(self) -> str:
if self.channel == "docs":
return self._system_prompt
body = self._behavior_store.text().strip()
section = BEHAVIOR_HEADER if not body else f"{BEHAVIOR_HEADER}\n{body}"
return f"{self._system_prompt}\n\n{section}"
@@ -406,6 +431,56 @@ class DeviiSession:
if messages and messages[0].get("role") == "system":
messages[0]["content"] = self._compose_system_prompt()
async def _docs_topic_gate(self, text: str) -> None:
prior = [
m
for m in self.agent._messages
if m.get("role") in ("user", "assistant") and m.get("content")
]
if not prior:
return
decision, reason = await self._classify_topic(prior, text)
if decision == "new":
messages = self.agent._messages
if messages and messages[0].get("role") == "system":
system = messages[0]
else:
system = {"role": "system", "content": self._compose_system_prompt()}
self.agent._messages = [system]
self._buffer = []
if self.persist_conversation:
try:
self._conv.clear(self.owner_kind, self.owner_id, self.channel)
except Exception: # noqa: BLE001 - clearing storage must not break the turn
logger.exception(
"Failed to clear docs conversation for %s/%s",
self.owner_kind,
self.owner_id,
)
await self._emit(
{"type": "topic", "decision": decision, "reason": reason}, buffer=False
)
async def _classify_topic(
self, prior: list[dict[str, Any]], text: str
) -> tuple[str, str]:
recent = prior[-6:]
convo = "\n".join(
f"{m['role']}: {str(m['content'])[:300]}" for m in recent
)
messages = [
{"role": "system", "content": TOPIC_CLASSIFIER_PROMPT},
{
"role": "user",
"content": f"Prior conversation:\n{convo}\n\nNew message:\n{text}\n\nClassify.",
},
]
try:
raw = await self._llm.complete_text(messages)
except Exception: # noqa: BLE001 - on any failure, keep context (treat as follow-up)
return "follow_up", ""
return _parse_topic(raw)
def _quota_snapshot(self) -> dict[str, Any]:
spent = self._ledger.spent_24h(self.owner_kind, self.owner_id)
turns = self._ledger.turns_24h(self.owner_kind, self.owner_id)
@@ -440,7 +515,12 @@ class DeviiSession:
cost_delta = round(after["cost_usd"] - before["cost_usd"], 8)
try:
if self.persist_conversation:
self._conv.save(self.owner_kind, self.owner_id, self.agent._messages)
self._conv.save(
self.owner_kind,
self.owner_id,
self.agent._messages,
self.channel,
)
self._ledger.record(
self.owner_kind,
self.owner_id,
@@ -615,6 +695,88 @@ NON_ADMIN_COST_RULE = (
"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": "<one short sentence explaining why>"}.'
)
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:
devplacepy/services/devii/store.py
+15 -6
View File
@@ -28,11 +28,13 @@ def _iso(moment: datetime) -> str:
class ConversationStore:
def load(self, owner_kind: str, owner_id: str) -> list[dict[str, Any]] | None:
def load(
self, owner_kind: str, owner_id: str, channel: str = "main"
) -> list[dict[str, Any]] | None:
if CONVERSATIONS not in db.tables:
return None
row = get_table(CONVERSATIONS).find_one(
owner_kind=owner_kind, owner_id=owner_id
owner_kind=owner_kind, owner_id=owner_id, channel=channel
)
if not row or not row.get("messages"):
return None
@@ -43,18 +45,23 @@ class ConversationStore:
return None
def save(
self, owner_kind: str, owner_id: str, messages: list[dict[str, Any]]
self,
owner_kind: str,
owner_id: str,
messages: list[dict[str, Any]],
channel: str = "main",
) -> None:
now = _iso(_now())
record = {
"owner_kind": owner_kind,
"owner_id": owner_id,
"channel": channel,
"messages": json.dumps(messages),
"updated_at": now,
}
table = get_table(CONVERSATIONS)
existing = (
table.find_one(owner_kind=owner_kind, owner_id=owner_id)
table.find_one(owner_kind=owner_kind, owner_id=owner_id, channel=channel)
if CONVERSATIONS in db.tables
else None
)
@@ -64,9 +71,11 @@ class ConversationStore:
record["created_at"] = now
table.insert(record)
def clear(self, owner_kind: str, owner_id: str) -> None:
def clear(self, owner_kind: str, owner_id: str, channel: str = "main") -> None:
if CONVERSATIONS in db.tables:
get_table(CONVERSATIONS).delete(owner_kind=owner_kind, owner_id=owner_id)
get_table(CONVERSATIONS).delete(
owner_kind=owner_kind, owner_id=owner_id, channel=channel
)
class UsageLedger: