o2o-site-AEO/solution/backend/services/agent/runtime.py

"""에이전트 런타임 — 발화 → 도구 선택 → 실행 → 응답.

★★ **채널을 모른다.** 빌더 화면에서 왔는지 카카오톡에서 왔는지 알 필요가 없다.
   이걸 웹훅 핸들러 안에 짜면 빌더에서 같은 걸 못 쓰고, 카카오 심사가 끝나야
   무엇 하나 검증되지 않는다(docs/AGENT.md).

★ 확인이 필요한지는 **레지스트리의 등급**이 정한다. 모델이 정하게 두면 프롬프트에
  끼어든 한 줄이 확인 절차를 건너뛴다.

★ 실행 결과 문구는 도구가 만든다(tools.py). LLM 문장은 '되묻기' 에만 쓴다 —
  모델이 결과를 쓰면 하지 않은 일을 했다고 말할 수 있다.
"""

import uuid

import httpx

from common.category_schema.loader import get_schema
from common.enums import DBWRType, ErrorType, PlaceCategory
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import places
from common.models.gmodel import UserInfo
from config.server_configs import external_api_config
from crud.fact_crud import FactCRUD
from crud.place_crud import PlaceCRUD
from services.agent import tools as registry
from services.agent.tools import ToolContext, ToolGrade, ToolRejected
from services.fact_service import FactService
from services.llm import provider
from services.llm.errors import LlmError
from services.prompts import agent as prompt
from common.logger import LOG

# 발화 길이 상한. 프롬프트 비용은 입력 토큰에 비례하고, 사장님이 한 번에 치는 말은 길지 않다.
MAX_MESSAGE = 500
# 도구 선택은 짧은 프롬프트라 빠르다. 카카오 웹훅의 5초 벽 안에 들어가야 한다(docs/AGENT.md).
REQUEST_TIMEOUT = httpx.Timeout(20.0, connect=5.0)


class AgentError(RuntimeError):
    """라우터가 HTTP 로 옮길 도메인 예외. 코드 문자열만 담는다(social 과 같은 규약)."""


def is_configured() -> bool:
    return provider.active().is_configured()


async def _load_place(user: UserInfo, place_id: str):
    """★ 소유자 범위. 없는 것과 남의 것을 똑같이 PLACE_NOT_FOUND 로 답한다(레포 관례).

    에이전트가 이 관례를 벗어나면 대화창이 소유자 스코프를 우회하는 유일한 입구가 된다."""
    err, place = await DB_SESSION_MNG.execute_lambda(
        places.DBType(),
        DBWRType.DB_READ.value,
        lambda s: PlaceCRUD().get_place(s, uuid.UUID(user.user_id), uuid.UUID(place_id)),
    )
    if err != ErrorType.SUCCESS or place is None:
        raise AgentError("PLACE_NOT_FOUND")
    return place


async def _context_facts(user: UserInfo, place_id: str, place) -> list[dict]:
    """모델에게 줄 '지금 값'. 이게 없으면 "3시로 바꿔줘" 가 무엇을 바꾸는지 모델이 모른다."""
    res = await FactService(FactCRUD(), PlaceCRUD()).list_facts(user, place_id, publishable_only=True)
    schema = get_schema(PlaceCategory(place.category))
    out = []
    for f in (res.facts or [])[:60]:
        spec = schema.get(f.key)
        if spec and spec.scope == "place" and (f.value or "").strip():
            out.append({"key": f.key, "label": spec.label, "value": f.value})
    return out


async def _choose(place, fields, facts, site_line, message) -> dict:
    """LLM 한 번. 고른 도구 이름과 인자만 받는다."""
    active = provider.active()
    async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
        result = await active.generate(
            client,
            external_api_config.gemini_text_model if active.__name__.endswith("gemini") else external_api_config.openai_text_model,
            prompt=prompt.build_prompt(
                place_name=place.name,
                tools=registry.describe(),
                fields=fields,
                facts=facts,
                site={"요약": site_line},
                message=message,
            ),
            response_schema=prompt.RESPONSE_SCHEMA,
            temperature=0.0,
        )
    return result.json or {}


async def chat(user: UserInfo, place_id: str, message: str, confirm: dict | None = None) -> dict:
    """대화 한 번.

    confirm 이 오면 LLM 을 부르지 않는다 — 사장님이 직전에 본 확인 문구에 '네' 를 누른 것이고,
    그 문장이 가리키는 도구를 그대로 실행한다. **인자는 다시 검증한다** — 화면에서 온 값을
    믿고 실행하면, 확인 절차가 오히려 검증을 건너뛰는 구멍이 된다.
    """
    message = (message or "").strip()
    if confirm is None and not message:
        raise AgentError("AGENT_EMPTY_MESSAGE")
    if len(message) > MAX_MESSAGE:
        raise AgentError("AGENT_MESSAGE_TOO_LONG")

    place = await _load_place(user, place_id)
    ctx = ToolContext(user=user, place_id=place_id, place=place)

    if confirm is not None:
        tool = registry.REGISTRY.get(confirm.get("tool") or "")
        if tool is None or tool.grade == ToolGrade.READ:
            raise AgentError("AGENT_UNKNOWN_TOOL")
        return await _execute(ctx, tool, confirm.get("args") or {})

    if not is_configured():
        raise AgentError("AGENT_NOT_CONFIGURED")

    fields = registry.fields_of(place)
    facts = await _context_facts(user, place_id, place)
    site_line = await registry.REGISTRY["get_site_status"].run(ctx, {})

    try:
        choice = await _choose(place, fields, facts, site_line, message)
    except LlmError as ex:
        LOG.w(f"[agent] 도구 선택 실패: {type(ex).__name__}")
        raise AgentError("AGENT_CALL_FAILED") from ex

    name = (choice.get("tool") or "").strip()
    tool = registry.REGISTRY.get(name)
    if tool is None:
        # ★ 모르는 이름을 지어냈거나 모델이 되묻기를 골랐다. 둘 다 '실행하지 않는다' 로 같다.
        return {
            "reply": (choice.get("message") or "").strip() or "무엇을 도와드릴까요?",
            "tool": None,
            "needs_confirm": False,
        }

    args = choice.get("args") or {}
    if tool.grade == ToolGrade.SEMI:
        # 실행하지 않는다. 사장님이 한 번 더 눌러야 한다.
        return {"reply": tool.confirm, "tool": tool.name, "args": args, "needs_confirm": True}

    return await _execute(ctx, tool, args)


async def _execute(ctx: ToolContext, tool, args: dict) -> dict:
    try:
        reply = await tool.run(ctx, args)
    except ToolRejected as ex:
        # 도구가 거절한 이유는 사장님께 그대로 보여 준다 — 실패를 숨기면 다시 시도한다.
        return {"reply": str(ex), "tool": tool.name, "needs_confirm": False, "rejected": True}
    return {"reply": reply, "tool": tool.name, "needs_confirm": False, "done": tool.grade != ToolGrade.READ}