gogo-telefon/gogo/agent/loop.py

"""Conversation loop shared by voice, chat and the playground (M2).

One AgentConversation per call/chat session. Each user turn may trigger several
LLM round-trips when the model calls tools; the loop executes them via
ToolExecutor and feeds tool_results back until the model produces text.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field

from sqlalchemy.ext.asyncio import AsyncSession

from gogo.agent.llm import LLMProvider, get_llm
from gogo.agent.prompt import compose_greeting, compose_system_prompt
from gogo.agent.tools import TOOL_DEFINITIONS, ToolExecutor
from gogo.domain import CallOutcome
from gogo.models import Tenant

log = logging.getLogger("gogo.agent.loop")

MAX_TOOL_ROUNDS_PER_TURN = 6  # guard against tool-call loops


@dataclass
class Turn:
    role: str  # "user" | "assistant"
    text: str
    tool_calls: list[tuple[str, dict]] = field(default_factory=list)


class AgentConversation:
    def __init__(
        self,
        session: AsyncSession,
        tenant: Tenant,
        *,
        channel: str,  # "voice" | "chat"
        caller_phone: str = "",
        call_id=None,
        chat_session_id=None,
        llm: LLMProvider | None = None,
    ):
        self.session = session
        self.tenant = tenant
        self.channel = channel
        self.llm = llm or get_llm()
        self.executor = ToolExecutor(
            session,
            tenant,
            source=channel,
            caller_phone=caller_phone,
            call_id=call_id,
            chat_session_id=chat_session_id,
        )
        self.messages: list[dict] = []  # Anthropic-format history
        self.turns: list[Turn] = []  # human-readable transcript
        self._system: str | None = None

    async def greeting(self) -> str:
        """Opening line (spoken by TTS / shown in the chat widget)."""
        text = compose_greeting(self.tenant, self.channel)
        # seed history so the model knows it already greeted
        self.messages.append({"role": "assistant", "content": text})
        self.turns.append(Turn("assistant", text))
        return text

    async def user_turn(self, text: str) -> str:
        """Process one user utterance; returns the assistant's reply text."""
        if self._system is None:
            self._system = await compose_system_prompt(self.session, self.tenant, self.channel)

        self.messages.append({"role": "user", "content": text})
        self.turns.append(Turn("user", text))

        reply_parts: list[str] = []
        turn_tool_calls: list[tuple[str, dict]] = []

        for _round in range(MAX_TOOL_ROUNDS_PER_TURN):
            response = await self.llm.complete(
                system=self._system,
                messages=self.messages,
                tools=TOOL_DEFINITIONS,
                model=self.tenant.llm_model or None,
            )
            if response.text:
                reply_parts.append(response.text)

            if not response.tool_uses:
                self.messages.append(
                    {"role": "assistant", "content": response.text or "…"}
                )
                break

            # record assistant blocks (text + tool_use) exactly as produced
            content: list[dict] = []
            if response.text:
                content.append({"type": "text", "text": response.text})
            for tu in response.tool_uses:
                content.append(
                    {"type": "tool_use", "id": tu.id, "name": tu.name, "input": tu.input}
                )
            self.messages.append({"role": "assistant", "content": content})

            results = []
            for tu in response.tool_uses:
                turn_tool_calls.append((tu.name, tu.input))
                result = await self.executor.execute(tu.name, tu.input)
                results.append(
                    {"type": "tool_result", "tool_use_id": tu.id, "content": result}
                )
            self.messages.append({"role": "user", "content": results})
        else:
            log.warning("tool-round limit hit (tenant %s)", self.tenant.slug)

        reply = "\n".join(p for p in reply_parts if p).strip()
        if not reply:
            reply = "Izvinite, došlo je do tehničke greške. Salon će vas nazvati u najkraćem roku."
        self.turns.append(Turn("assistant", reply, tool_calls=turn_tool_calls))
        return reply

    @property
    def outcome(self) -> str:
        """Conversation outcome for call/chat history (§10)."""
        if self.executor.created_request_id is not None:
            return CallOutcome.request_created.value
        if self.executor.took_message:
            return CallOutcome.message_taken.value
        if any(t.role == "user" for t in self.turns):
            return CallOutcome.info_only.value
        return CallOutcome.abandoned.value

    @property
    def transcript(self) -> list[dict]:
        """JSON-serializable transcript for calls.transcript / chat storage."""
        return [
            {
                "role": t.role,
                "text": t.text,
                **(
                    {"tool_calls": [{"name": n, "input": i} for n, i in t.tool_calls]}
                    if t.tool_calls
                    else {}
                ),
            }
            for t in self.turns
        ]