Store Support Bot in Python: Source Code

1"""The support assistant: one customer message in, one reply out, with tool calls in between."""23from __future__ import annotations45import asyncio6import json7import logging8import time9from dataclasses import dataclass, field10from typing import Any, Protocol1112import openai1314from .language import language_instruction15from .memory import ConversationMemory16from .tools import TOOL_SCHEMAS, ToolContext, ToolExecutor1718logger = logging.getLogger(__name__)1920FALLBACK_ERROR = "سمحلي، صرا مشكل تقني. عاود جرب بعد شوية.\nDésolé, un problème technique est survenu. Réessayez dans un instant."21FALLBACK_BUSY = "كاين ضغط كبير دروك. عاود ابعث رسالتك بعد دقيقة.\nLe service est très sollicité. Renvoyez votre message dans une minute."22FALLBACK_EMPTY = "سمحلي، ما فهمتش مليح. تقدر تعاود تكتب سؤالك؟\nDésolé, je n'ai pas bien compris. Pouvez-vous reformuler ?"232425class ChatCompletionsClient(Protocol):26    """The part of the OpenAI-compatible client this module uses."""2728    @property29    def chat(self) -> Any: ...303132@dataclass33class Reply:34    text: str35    tools_used: list[str] = field(default_factory=list)36    ticket_ids: list[int] = field(default_factory=list)37    failed: bool = False383940class StoreAssistant:41    def __init__(42        self,43        client: ChatCompletionsClient,44        model: str,45        executor: ToolExecutor,46        memory: ConversationMemory,47        system_prompt: str,48        *,49        max_tool_rounds: int = 4,50        max_output_tokens: int = 8192,51    ) -> None:52        self._client = client53        self._model = model54        self._executor = executor55        self._memory = memory56        self._system_prompt = system_prompt57        self._max_tool_rounds = max_tool_rounds58        self._max_output_tokens = max_output_tokens5960    def reset(self, chat_id: int) -> None:61        self._memory.clear(chat_id)6263    async def reply(self, chat_id: int, user_text: str, username: str | None = None) -> Reply:64        ctx = ToolContext(chat_id=chat_id, username=username)65        messages: list[dict[str, Any]] = [66            {"role": "system", "content": self._system_prompt},67            *(dict(turn) for turn in self._memory.history(chat_id)),68        ]69        hint = language_instruction(user_text)70        if hint:71            messages.append({"role": "system", "content": hint})72        messages.append({"role": "user", "content": user_text})73        tools_used: list[str] = []74        started = time.monotonic()7576        try:77            text = await self._run_loop(messages, ctx, tools_used)78        except openai.RateLimitError:79            logger.warning("chat=%s rate limited by the model API", chat_id)80            return Reply(FALLBACK_BUSY, tools_used, ctx.tickets, failed=True)81        except (openai.APIConnectionError, openai.APITimeoutError):82            logger.warning("chat=%s model API unreachable or timed out", chat_id)83            return Reply(FALLBACK_ERROR, tools_used, ctx.tickets, failed=True)84        except openai.APIStatusError as exc:85            logger.error("chat=%s model API returned HTTP %s", chat_id, exc.status_code)86            return Reply(FALLBACK_ERROR, tools_used, ctx.tickets, failed=True)8788        if not text:89            text = FALLBACK_EMPTY90        else:91            self._memory.add_exchange(chat_id, user_text, text)9293        logger.info(94            "chat=%s replied in %.1fs tools=%s tickets=%s",95            chat_id,96            time.monotonic() - started,97            ",".join(tools_used) or "-",98            ctx.tickets or "-",99        )100        return Reply(text, tools_used, ctx.tickets)101102    async def _run_loop(self, messages: list[dict[str, Any]], ctx: ToolContext, tools_used: list[str]) -> str:103        cache: dict[tuple[str, str], str] = {}104        for round_no in range(self._max_tool_rounds + 1):105            # On the last round the model must answer with text.106            final_round = round_no == self._max_tool_rounds107            response = await self._client.chat.completions.create(108                model=self._model,109                messages=messages,110                tools=TOOL_SCHEMAS,111                tool_choice="none" if final_round else "auto",112                max_tokens=self._max_output_tokens,113            )114            if not response.choices:115                return ""116            message = response.choices[0].message117            tool_calls = message.tool_calls or []118            if not tool_calls or final_round:119                return (message.content or "").strip()120121            if response.choices[0].finish_reason == "length":122                logger.warning("model hit the output limit while calling tools; consider raising MAX_OUTPUT_TOKENS")123124            messages.append(_assistant_turn(message))125            for call in tool_calls:126                name = call.function.name127                key = (name, _canonical_arguments(call.function.arguments))128                if key in cache:129                    # Models sometimes repeat the same call; answer it from the first result130                    # (and never open a second ticket for the same request).131                    result = cache[key]132                else:133                    tools_used.append(name)134                    result = await asyncio.to_thread(self._executor.run, name, call.function.arguments, ctx)135                    cache[key] = result136                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})137        return ""138139140def _canonical_arguments(raw: str | None) -> str:141    try:142        return json.dumps(json.loads(raw or "{}"), sort_keys=True, ensure_ascii=False)143    except json.JSONDecodeError:144        return raw or ""145146147def _assistant_turn(message: Any) -> dict[str, Any]:148    """Rebuild the assistant message that requested tools, to send it back with the results."""149    turn: dict[str, Any] = {150        "role": "assistant",151        "content": message.content or "",152        "tool_calls": [153            {154                "id": call.id,155                "type": "function",156                "function": {"name": call.function.name, "arguments": call.function.arguments or "{}"},157            }158            for call in message.tool_calls159        ],160    }161    # Reasoning models may return their reasoning next to the tool calls; some providers162    # expect it back in the same turn.163    reasoning = getattr(message, "reasoning_content", None)164    if reasoning:165        turn["reasoning_content"] = reasoning166    return turn