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 import agent_config as config
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
# ★ 한 발화로 실행할 도구 수 상한. 무한정 허용하면 "다 지워줘" 한 마디에 연쇄로 실행된다.
MAX_ACTIONS = 5
# 값을 바꾼 뒤 한 번만 붙이는 안내. fact 는 바뀌어도 사이트는 안 바뀐다 —
# 이 줄이 빠지면 사장님은 반영된 줄 알고 확인하러 갔다가 옛 값을 보고 "고장났네" 가 된다.
REPUBLISH_NOTICE = "사이트에 반영하려면 다시 발행해야 해요 — 지금 할까요?"
# 도구 선택은 짧은 프롬프트라 빠르다. 카카오 웹훅의 5초 벽 안에 들어가야 한다(docs/AGENT.md).
REQUEST_TIMEOUT = httpx.Timeout(20.0, connect=5.0)


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


def _args(value) -> dict:
    """★ 모델은 스키마를 어길 수 있다 — args 를 배열·문자열로 보내면 도구의 `.get` 에서 죽고,
    사장님에게는 "처리할 수 없어요" 로만 보인다. 인자가 없는 것으로 치면 도구가 되묻는다."""
    return value if isinstance(value, dict) else {}


def is_configured() -> bool:
    """대화창을 열 수 있나 — 스위치와 LLM 키를 **둘 다** 본다.

    ★ 스위치(`AGENT_CHAT_ENABLED`)와 키를 **둘 다** 보는 이유: 키만 보면 "잠시 닫아 두기" 를
      키를 지워서 해야 하는데 그러면 소개문·사진분류까지 같이 꺼진다. 스위치만 보면
      키 없는 환경에서 **눌러도 안 되는 입구**가 생긴다.
      실제로 2026-09-21 에 카카오 채널 보류로 한 번 닫았고, 채널 인증이 끝나 다시 열었다."""
    return config.chat_enabled() and 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 []):
        spec = schema.get(f.key)
        if spec and spec.scope == "place" and (f.value or "").strip():
            # ★ label 은 싣지 않는다 — 아래 '항목 목록' 에 이미 key↔label 이 있다.
            #   같은 표를 두 번 보내면 프롬프트만 커지고 모델이 얻는 것은 없다.
            out.append({f.key: f.value})
    # ★ 상한을 둔다. 실측(2026-09-22): 필드 43 + fact 수십 개가 실린 프롬프트가 5초 벽을
    #   넘겼다. 무한정 싣지 않는다 — 대화 한 턴에 필요한 맥락은 그렇게 많지 않다.
    return out[:30]


async def _choose(place, fields, facts, sections, photos, 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,
                sections=sections,
                photos=photos,
                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, _args(confirm.get("args")))

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

    fields = registry.fields_of(place)
    facts = await _context_facts(user, place_id, place)
    # ★ 섹션은 이름으로 지목한다("후기 빼줘"). 목록을 안 실으면 모델이 이름을 지어낸다.
    sections, _theme = await registry.sections_of(ctx)
    # ★ 사진도 이름으로 지목한다. 목록을 안 실으면 모델이 라벨을 지어낸다.
    photos = await registry.photo_names(ctx)
    # ★ 사이트 상태는 프롬프트에 싣지 않는다. 그 한 줄 때문에 매 턴 사이트 조회 + 슬러그 계산이
    #   돌았고, 정작 모델이 필요할 때는 `get_site_status` 도구를 부르면 된다.

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

    actions = [a for a in (choice.get("actions") or []) if isinstance(a, dict)]
    skipped = _skipped(choice.get("skipped"))
    if not actions:
        # ★ 모델이 되묻기를 골랐다. '실행하지 않는다' 로 끝난다. 할 수 없는 요청뿐이었으면
        #   그렇다고 말한다 — 빈 되묻기로 끝나면 사장님은 같은 말을 되풀이한다.
        asked = (choice.get("message") or "").strip()
        body = "\n".join(line for line in [asked, _skipped_line(skipped)] if line)
        return {
            "reply": body or "무엇을 도와드릴까요?",
            "tool": None,
            "needs_confirm": False,
        }
    return await _run_actions(ctx, actions, skipped)


# 모델이 "할 수 없는 요청" 으로 돌려준 이름. 문장이 아니라 이름이라 짧게 끊는다 —
# 틀에 끼워 코드가 문장을 만든다(_skipped_line).
_SKIPPED_MAX = 5
_SKIPPED_LEN = 30


def _skipped(value) -> list[str]:
    if not isinstance(value, list):
        return []
    names = [str(v).strip()[:_SKIPPED_LEN] for v in value if str(v or "").strip()]
    return names[:_SKIPPED_MAX]


def _skipped_line(skipped: list[str]) -> str:
    """★ 되는 것만 하고 입을 다물면 사장님은 전부 된 줄 안다(2026-09-29).
    모델은 '무엇' 만 주고 문장은 코드가 만든다 — "했다" 고 말할 자리를 주지 않는다."""
    if not skipped:
        return ""
    return ", ".join(f"'{name}'" for name in skipped) + " 은(는) 대화로는 아직 할 수 없어요."


def _target(action: dict) -> tuple:
    """같은 요청인지 가를 열쇠. 도구가 겨누는 인자(Tool.target)가 같으면 같은 요청이다."""
    name = (action.get("tool") or "").strip()
    args = _args(action.get("args"))
    tool = registry.REGISTRY.get(name)
    if tool is None or tool.target is None:
        return (name, repr(sorted((str(k), str(v)) for k, v in args.items())))
    return (name,) + tuple(str(args.get(key) or "").strip() for key in tool.target)


def _merge(actions: list) -> list:
    """같은 대상을 두 번 시키면 **마지막 하나**만 남긴다 — "체크인 3시… 아니 4시로".

    ★ 둘 다 실행하면 값은 같아도 문구에 "15:00 로" "16:00 로" 가 함께 서서, 사장님은
      어느 쪽이 남았는지 되묻게 된다. 자리는 마지막 것의 자리다 — 고쳐 말한 그 순간이다.
    ★ 상한(MAX_ACTIONS)을 세기 전에 합친다. 고쳐 말한 것까지 세면 할 수 있는 일이 잘린다."""
    last = {_target(a): i for i, a in enumerate(actions)}
    return [a for i, a in enumerate(actions) if last[_target(a)] == i]


def _left_lines(ctx: ToolContext, left: list, unknown: int) -> list[str]:
    """멈춘 뒤 남은 요청과, 알아듣지 못해 건너뛴 요청.

    ★ 말없이 버리면 사장님은 그것도 된 줄 안다. 이름은 도구가 만든다(tools.describe_action)."""
    names = []
    for action in left:
        tool = registry.REGISTRY.get((action.get("tool") or "").strip())
        if tool is None:
            unknown += 1
        else:
            names.append(registry.describe_action(ctx, tool, _args(action.get("args"))))
    lines = []
    if names:
        lines.append(f"{', '.join(names)} 은(는) 아직 하지 않았어요. 다시 말씀해 주세요.")
    if unknown:
        lines.append(f"알아듣지 못한 요청 {unknown}가지는 하지 않았어요. 다시 말씀해 주세요.")
    return lines


async def _run_actions(ctx: ToolContext, actions: list, skipped: list[str] | None = None) -> dict:
    """시킨 순서대로 실행한다.

    ★ SEMI(되돌릴 수 없는 쪽)를 만나면 **거기서 멈춘다.** 앞서 한 일을 함께 말하고 확인을
      받는다 — 확인이 필요한 행위를 다른 일에 묻어 실행하면 확인의 의미가 없다.
    ★ 하나가 실패해도 **앞의 것을 되돌리지 않는다**(2026-09-28 사장님 결정). 되돌리는 것도
      사장님이 시키지 않은 변경이다. 대신 **무엇이 됐고 무엇이 안 됐는지 그대로 말한다** —
      부분 성공을 뭉뚱그리면 사장님은 전부 된 줄 안다.
    ★ 멈춘 뒤의 요청도 이름을 대서 알린다(2026-09-29). 확인을 눌러도 발행 하나만 돈다.
    """
    lines: list[str] = []
    changed = False          # 실제로 바뀐 것이 하나라도 있었나(재발행 안내의 조건)
    republish = False
    last_tool = None
    unknown = 0
    skipped_line = _skipped_line(skipped or [])

    actions = _merge(actions)
    queue, over = actions[:MAX_ACTIONS], actions[MAX_ACTIONS:]
    for at, action in enumerate(queue):
        tool = registry.REGISTRY.get((action.get("tool") or "").strip())
        if tool is None:
            unknown += 1     # 모델이 지어낸 이름 — 실행하지 않고, 건너뛰었다고 말한다
            continue
        args = _args(action.get("args"))

        if tool.grade == ToolGrade.SEMI:
            # 실행하지 않는다. 사장님이 한 번 더 눌러야 한다. ★ 묻는 말은 맨 끝에 선다 —
            # 그 뒤에 다른 말이 붙으면 [네, 해주세요] 가 무엇에 대한 답인지 흐려진다.
            left = _left_lines(ctx, queue[at + 1:] + over, unknown)
            body = "\n".join(line for line in lines + left + [skipped_line, tool.confirm] if line)
            return {"reply": body, "tool": tool.name, "args": args,
                    "needs_confirm": True, "done": changed}

        try:
            line = await tool.run(ctx, args)
        except ToolRejected as ex:
            # ★ 거절 이유를 그대로 보여 주고 거기서 멈춘다. 뒤의 것을 마저 하면
            #   사장님이 못 본 사이에 더 바뀐다.
            lines.append(f"{ex} — 여기서 멈췄습니다." if lines else str(ex))
            left = _left_lines(ctx, queue[at + 1:] + over, unknown)
            return {"reply": _compose(lines + left + [skipped_line], republish), "tool": last_tool,
                    "needs_confirm": False, "rejected": True, "done": changed}

        lines.append(line)
        last_tool = tool.name
        # ★ "이미 켜져 있어요" 는 바뀐 것이 아니다 — 재발행을 권하면 무언가 바뀐 줄 안다.
        if tool.grade != ToolGrade.READ and not isinstance(line, registry.Unchanged):
            changed = True
            republish = republish or tool.republish

    if over:
        lines.append(f"한 번에 {MAX_ACTIONS}가지까지 해 드릴 수 있어요. 나머지는 다시 말씀해 주세요.")
    lines += _left_lines(ctx, [], unknown) + [skipped_line]
    return {"reply": _compose(lines, republish), "tool": last_tool,
            "needs_confirm": False, "done": changed}


def _compose(lines: list, republish: bool) -> str:
    """★ 재발행 안내는 **한 번만** 붙인다. 도구마다 문장에 박아 두면 셋을 고쳤을 때
    같은 말이 세 번 나온다."""
    body = "\n".join(line for line in lines if line)
    return f"{body}\n{REPUBLISH_NOTICE}" if republish else body


async def _execute(ctx: ToolContext, tool, args: dict) -> dict:
    """확인(SEMI)을 받고 돌아온 한 건. 목록 경로와 달리 이건 언제나 도구 하나다."""
    try:
        reply = await tool.run(ctx, args)
    except ToolRejected as ex:
        # 도구가 거절한 이유는 사장님께 그대로 보여 준다 — 실패를 숨기면 다시 시도한다.
        return {"reply": str(ex), "tool": tool.name, "needs_confirm": False, "rejected": True}
    changed = tool.grade != ToolGrade.READ and not isinstance(reply, registry.Unchanged)
    return {"reply": _compose([reply], tool.republish and changed), "tool": tool.name,
            "needs_confirm": False, "done": changed}