36b818eda3
"맨 위 · 맨 아래 · X 다음으로" 만 되고 "X 앞으로" "한 칸 위로" "세 번째로" "자리 바꿔줘" 는
거절됐다. 발행본(HomePage.tsx)은 히어로를 늘 맨 위, SNS 를 늘 맨 아래에 그리는데 대화는
그 둘을 옮기고 "옮겼습니다" 라고 답했다 — 화면은 그대로였다.
- tools: move_section 에 where(맨 위·맨 아래·앞·뒤·위로·아래로·번째·바꾸기)·count —
where 가 비면 예전 to 표기로 읽는다. 한 칸·N번째는 보이는 순서로 센다(꺼진 부분과 자리만
바꾸는 헛이동 방지). 없는 순번·꺼진 부분의 칸 이동은 거절, 이미 그 자리면 Unchanged
- tools: PINNED(히어로 맨 위 · SNS 맨 아래) — 옮기기와 그 둘을 기준으로 한 앞·뒤·바꾸기를 막는다.
히어로 다음은 맨 위, SNS 앞은 맨 아래로 읽는다
- tools: toggle_section 이 쉼표로 여럿을 받는다 — 하나라도 못 찾거나 잠겼으면 아무것도 안 바꾼다
- tools: list_sections 는 보이는 순서에 번호(= N번째 기준)를 붙이고 꺼진 것을 모은다, only=꺼진
- prompts/runtime: 스키마에 where·count·only, 섹션 줄에 [항상 맨 위]·[항상 맨 아래] —
목록은 tools.PINNED 하나를 runtime 이 넘긴다(prompts 는 services 를 import 하지 않는다)
- docs/AGENT.md: 옮기기·숨기기 표와 근거. 동작하지 않던 예시("후기 빼줘") 교체 —
이용 후기는 섹션 목록에 없고 발행본이 늘 그린다
테스트 21건 추가, 에이전트·카카오 174 passed. pyflakes 새 경고 없음
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
309 lines
15 KiB
Python
309 lines
15 KiB
Python
"""에이전트 런타임 — 발화 → 도구 선택 → 실행 → 응답.
|
|
|
|
★★ **채널을 모른다.** 빌더 화면에서 왔는지 카카오톡에서 왔는지 알 필요가 없다.
|
|
이걸 웹훅 핸들러 안에 짜면 빌더에서 같은 걸 못 쓰고, 카카오 심사가 끝나야
|
|
무엇 하나 검증되지 않는다(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,
|
|
pinned=registry.PINNED,
|
|
),
|
|
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}
|