kbr4/gogo-telefon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M2: agent logic in text mode

Browse Source 
- LLMProvider interface: AnthropicLLM + ScriptedLLM test fake
- System prompt composition from super-admin template + structured tenant
  data only (§6.3); auto-generated greetings with recording disclosure,
  out-of-hours variant, chat variant
- Agent tools (§6.5): get_salon_info, check_availability (≤8 slots to model,
  agent offers ≤3), submit_booking_request (caller-ID fallback, hallucinated
  service-id guard), take_message
- Conversation loop with tool dispatch, round-limit guard, transcript capture,
  outcome classification (request_created/message_taken/info_only/abandoned)
- Terminal playground (python -m gogo.agent.cli) + demo seed (python -m gogo.seed)
- Deterministic loop tests (ScriptedLLM) + Appendix A golden-scenario
  acceptance tests against a real LLM (skipped without an API key)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Senad Uka
2026-07-11 09:53:08 +02:00
parent e855650f09
commit 0d0838e6e2
9 changed files with 1445 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
tests/test_acceptance_appendix_a.py Normal file
View File

@@ -0,0 +1,172 @@
"""Appendix A golden-scenario acceptance tests against a REAL LLM.
Skipped unless GOGO_ANTHROPIC_API_KEY or ANTHROPIC_API_KEY is set (they spend
tokens). Each scenario drives a text-mode conversation and asserts the spec's
acceptance criteria: correct tool calls, no "confirmed" language, at most 3
offered slots, proper closing (§ Appendix A / M2).
Run just these: pytest tests/test_acceptance_appendix_a.py -v
"""
import os
import re
import pytest
from gogo.agent.llm import AnthropicLLM
from gogo.agent.loop import AgentConversation
API_KEY = os.environ.get("GOGO_ANTHROPIC_API_KEY") or os.environ.get("ANTHROPIC_API_KEY")
pytestmark = pytest.mark.skipif(
not API_KEY, reason="no Anthropic API key set — real-LLM acceptance tests skipped"
)
# §6.4: never present the booking as final. ("radi potvrde" is fine; "potvrđen" is not)
FORBIDDEN = ["potvrđen", "potvrdjen", "rezervisan", "rezerviran", "bukiran", "zakazan"]
TIME_MENTION = re.compile(r"\b\d{1,2}[:.]\d{2}\b|\bu (jedan|dva|tri|četiri|pet|šest|sedam|osam|devet|deset|jedanaest|dvanaest)\b")
def assert_no_confirmed_language(text: str) -> None:
lower = text.lower()
for word in FORBIDDEN:
assert word not in lower, f"forbidden 'confirmed' language {word!r} in: {text}"
def count_offered_times(text: str) -> int:
return len(TIME_MENTION.findall(text))
@pytest.fixture
def llm():
return AnthropicLLM(api_key=API_KEY)
@pytest.fixture
async def convo(session, tenant, emails, sms, clean_mock_provider, llm):
"""Voice-channel conversation with caller-ID, mock provider, no busy blocks."""
c = AgentConversation(
session, tenant, channel="voice", caller_phone="+38765123456", llm=llm
)
await c.greeting()
return c
def tool_names(convo: AgentConversation) -> list[str]:
return [name for name, _ in convo.executor.tool_calls]
async def test_a1_happy_path_booking(convo, session, emails):
"""A.1: booking with a preferred time → ≤3 slots, request submitted, closing line."""
r1 = await convo.user_turn("Dobar dan, htjela bih zakazati šišanje i feniranje.")
assert_no_confirmed_language(r1)
r2 = await convo.user_turn("Nešto u srijedu poslijepodne ako ima.")
assert "check_availability" in tool_names(convo)
assert count_offered_times(r2) <= 3
assert_no_confirmed_language(r2)
r3 = await convo.user_turn("Može prvi termin koji ste rekli. Amra Hodžić.")
r4 = r3
if "submit_booking_request" not in tool_names(convo):
r4 = await convo.user_turn("Jeste, broj s kojeg zovem je pravi broj.")
assert "submit_booking_request" in tool_names(convo)
assert_no_confirmed_language(r4)
final = " ".join(t.text for t in convo.turns if t.role == "assistant")
# closing: forwards the request, salon will contact
assert "kontaktira" in final.lower() or "javi" in final.lower()
assert convo.outcome == "request_created"
assert len(emails) == 1 # proposal email to the owner
async def test_a2_price_question_only(convo, session, emails):
"""A.2: price-only call — range answered (price_mode=range for Merima seed is
'exact' in fixture; assert amount present), one gentle booking offer, no request."""
r1 = await convo.user_turn("Koliko košta šišanje i feniranje?")
assert re.search(r"\d+", r1), f"expected a price in: {r1}"
assert_no_confirmed_language(r1)
r2 = await convo.user_turn("Ne, samo sam htjela cijenu. Hvala.")
assert_no_confirmed_language(r2)
assert "submit_booking_request" not in tool_names(convo)
assert convo.outcome == "info_only"
assert emails == []
async def test_a3_unsure_client_no_advice(convo, session, emails):
"""A.3: agent must not give treatment advice — routes decision to the professional."""
r1 = await convo.user_turn(
"Uništila mi se kosa od peglanja, ne znam šta bi mi pomoglo..."
)
# must not recommend one treatment as the right one; the professional decides
assert_no_confirmed_language(r1)
r2 = await convo.user_turn(
"Može, super. Radnim danom poslije četiri mi odgovara. Lejla Begić, broj je ovaj."
)
r3 = ""
if "submit_booking_request" not in tool_names(convo):
r3 = await convo.user_turn("Jeste, taj broj. Može.")
assert "submit_booking_request" in tool_names(convo) or "take_message" in tool_names(convo)
for t in (r2, r3):
assert_no_confirmed_language(t)
async def test_a4_cancellation_takes_message(convo, session):
"""A.4: cancellation → take_message, no calendar promises."""
await convo.user_turn("Trebala bih otkazati termin za sutra.")
r2 = await convo.user_turn("Selma Kovač, sutra u deset.")
if "take_message" not in tool_names(convo):
r2 = await convo.user_turn("To je sve, hvala.")
assert "take_message" in tool_names(convo)
assert convo.outcome == "message_taken"
assert_no_confirmed_language(r2)
lower = r2.lower()
assert "slobod" not in lower # must not promise the slot is freed
async def test_a5_out_of_hours_greeting(session, tenant, emails, sms, clean_mock_provider, llm):
"""A.5: out-of-hours greeting states hours and offers booking."""
from datetime import datetime
from zoneinfo import ZoneInfo
from gogo.agent.prompt import compose_greeting
closed_at = datetime(2026, 7, 15, 21, 30, tzinfo=ZoneInfo("Europe/Sarajevo"))
greeting = compose_greeting(tenant, "voice", closed_at)
assert "zatvoren" in greeting
assert "radno vrijeme" in greeting
assert "razgovor se snima" in greeting
async def test_a6_two_clarifications_then_message(convo, session):
"""A.6: unintelligible caller → max 2 clarification attempts, then message + exit."""
r1 = await convo.user_turn("mrmlj hrm brbl nrzm")
r2 = await convo.user_turn("hmpf mrm vrm hrm")
r3 = await convo.user_turn("brm hrm mrmlj")
assert "take_message" in tool_names(convo), "expected take_message after failed clarifications"
final = (r2 + " " + r3).lower()
assert "nazva" in final or "javi" in final or "poruk" in final # promises a callback
for t in (r1, r2, r3):
assert_no_confirmed_language(t)
async def test_a7_home_visit_collects_address(convo, session, emails):
"""A.7: home-visit service → collects area, home_visit flag set on the request."""
await convo.user_turn("Da li vi dolazite kući za pedikir? Majka mi je nepokretna.")
await convo.user_turn("Naselje Lauš. Bilo koje prijepodne ove sedmice.")
r3 = await convo.user_turn("Fatima Softić, a broj je ovaj s kojeg zovem, može.")
if "submit_booking_request" not in tool_names(convo):
r3 = await convo.user_turn("Da, to je sve, hvala vam.")
assert "submit_booking_request" in tool_names(convo)
from sqlalchemy import select
from gogo.models import BookingRequest
req = (await session.execute(select(BookingRequest))).scalar_one()
assert req.home_visit is True
assert req.address and "lauš" in req.address.lower()
assert_no_confirmed_language(r3)

296
tests/test_agent_loop.py Normal file
View File

@@ -0,0 +1,296 @@
"""Agent loop machinery with a scripted LLM (no API key needed):
prompt composition, tool dispatch, transcripts, outcome classification."""
import json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from sqlalchemy import select
from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse
from gogo.agent.loop import AgentConversation
from gogo.agent.prompt import compose_greeting, compose_system_prompt
from gogo.models import BookingRequest, MessageForOwner, Service
TZ = ZoneInfo("Europe/Sarajevo")
# -- prompt composition -------------------------------------------------------
async def test_system_prompt_contains_tenant_data(session, tenant):
prompt = await compose_system_prompt(session, tenant, "voice")
assert "Salon Merima" in prompt
assert "Šišanje i feniranje" in prompt
assert "45 min" in prompt
assert "25–35 KM" in prompt
assert "ponedjeljak: 09:00–18:00" in prompt
assert "razgovor se snima" in prompt # recording disclosure in greeting (§5.4)
assert "phone call" in prompt # voice channel suffix
# service ids are exposed so the model can call tools with them
svc = (
await session.execute(select(Service).where(Service.tenant_id == tenant.id))
).scalars().first()
assert str(svc.id) in prompt
async def test_chat_prompt_requires_phone_collection(session, tenant):
prompt = await compose_system_prompt(session, tenant, "chat")
assert "web chat" in prompt
assert "phone number" in prompt
def test_greeting_variants(tenant):
open_at = datetime(2026, 7, 15, 10, 0, tzinfo=TZ) # Wed 10:00
closed_at = datetime(2026, 7, 15, 21, 0, tzinfo=TZ) # Wed 21:00
g_open = compose_greeting(tenant, "voice", open_at)
g_closed = compose_greeting(tenant, "voice", closed_at)
g_chat = compose_greeting(tenant, "chat")
assert "razgovor se snima" in g_open
assert "zatvoren" not in g_open
assert "trenutno zatvoren" in g_closed
assert "radno vrijeme" in g_closed
assert "razgovor se snima" not in g_chat # no recording disclosure in chat
async def test_prompt_template_override(session, tenant):
from gogo.models import TenantPromptOverride
session.add(
TenantPromptOverride(tenant_id=tenant.id, body="CUSTOM {salon_profile} END")
)
await session.flush()
prompt = await compose_system_prompt(session, tenant, "voice")
assert prompt.startswith("CUSTOM")
assert "Salon Merima" in prompt
# -- conversation loop --------------------------------------------------------
async def booking_conversation(session, tenant, emails):
"""Scripted A.1-style flow: availability check → booking request."""
svc = (
await session.execute(
select(Service).where(Service.tenant_id == tenant.id, Service.name.like("Šišanje%"))
)
).scalar_one()
wed = datetime(2026, 7, 15, tzinfo=TZ)
slot_start = wed.replace(hour=17)
llm = ScriptedLLM(
[
# turn 1: caller asks for a booking → model checks availability
LLMResponse(
text="",
tool_uses=[
ToolUse(
id="tu1",
name="check_availability",
input={
"service_id": str(svc.id),
"date_from": "2026-07-15",
"date_to": "2026-07-15",
},
)
],
stop_reason="tool_use",
),
LLMResponse(text="U srijedu poslijepodne slobodno je u pet. Odgovara?"),
# turn 2: caller accepts → model submits the request and closes
LLMResponse(
text="",
tool_uses=[
ToolUse(
id="tu2",
name="submit_booking_request",
input={
"service_id": str(svc.id),
"service_name_raw": "šišanje i feniranje",
"client_name": "Amra Hodžić",
"client_phone": "+38765123456",
"slots": [
{
"start": slot_start.isoformat(),
"end": (slot_start + timedelta(minutes=45)).isoformat(),
}
],
"summary": "Šišanje i feniranje, srijeda u 17h.",
},
)
],
stop_reason="tool_use",
),
LLMResponse(
text=(
"Prosljeđujem salonu zahtjev: šišanje i feniranje, srijeda u pet. "
"Kontaktiraće vas u najkraćem roku radi potvrde. Hvala i prijatno!"
)
),
]
)
convo = AgentConversation(
session, tenant, channel="voice", caller_phone="+38765123456", llm=llm
)
await convo.greeting()
r1 = await convo.user_turn("Htjela bih zakazati šišanje i feniranje u srijedu.")
r2 = await convo.user_turn("U pet, može. Amra Hodžić, broj je ovaj s kojeg zovem.")
return convo, llm, r1, r2
async def test_booking_flow_creates_request(session, tenant, emails, sms, clean_mock_provider):
convo, llm, r1, r2 = await booking_conversation(session, tenant, emails)
await session.commit()
assert "slobodno je u pet" in r1
assert "prijatno" in r2.lower()
# tool_result was fed back to the model
tool_result_msg = llm.calls[1]["messages"][-2] # assistant tool_use, then user tool_result
assert tool_result_msg["role"] == "assistant"
results = llm.calls[1]["messages"][-1]
assert results["role"] == "user"
assert results["content"][0]["type"] == "tool_result"
payload = json.loads(results["content"][0]["content"])
assert "slots" in payload
# a real BookingRequest exists and the proposal email went out
req = (await session.execute(select(BookingRequest))).scalar_one()
assert req.client_name == "Amra Hodžić"
assert req.source == "voice"
assert len(emails) == 1
assert convo.outcome == "request_created"
# transcript captures roles, text and tool calls
roles = [t["role"] for t in convo.transcript]
assert roles == ["assistant", "user", "assistant", "user", "assistant"]
assert convo.transcript[-1]["tool_calls"][0]["name"] == "submit_booking_request"
async def test_take_message_outcome(session, tenant, emails, sms):
llm = ScriptedLLM(
[
LLMResponse(
text="",
tool_uses=[
ToolUse(
id="tu1",
name="take_message",
input={
"text": "Selma Kovač otkazuje sutrašnji termin u deset.",
"client_name": "Selma Kovač",
},
)
],
stop_reason="tool_use",
),
LLMResponse(text="Prosljeđujem salonu poruku. Prijatno!"),
]
)
convo = AgentConversation(
session, tenant, channel="voice", caller_phone="+38765111222", llm=llm
)
await convo.user_turn("Trebala bih otkazati termin za sutra u deset, Selma Kovač.")
await session.commit()
msg = (await session.execute(select(MessageForOwner))).scalar_one()
assert "otkazuje" in msg.text
assert msg.client_phone == "+38765111222" # caller-ID fallback
assert convo.outcome == "message_taken"
async def test_info_only_outcome(session, tenant):
llm = ScriptedLLM([LLMResponse(text="Farbanje je od šezdeset do devedeset maraka.")])
convo = AgentConversation(session, tenant, channel="voice", llm=llm)
await convo.user_turn("Koliko košta farbanje?")
assert convo.outcome == "info_only"
async def test_submit_without_phone_fails_in_chat(session, tenant, emails):
"""Chat has no caller-ID: submitting without a phone returns an error to the model."""
llm = ScriptedLLM(
[
LLMResponse(
text="",
tool_uses=[
ToolUse(
id="tu1",
name="submit_booking_request",
input={"client_name": "Ana", "summary": "manikir"},
)
],
stop_reason="tool_use",
),
LLMResponse(text="Koji je vaš broj telefona za kontakt?"),
]
)
convo = AgentConversation(session, tenant, channel="chat", llm=llm)
await convo.user_turn("Može manikir sutra?")
results = llm.calls[1]["messages"][-1]
payload = json.loads(results["content"][0]["content"])
assert payload["error"] == "missing_phone"
assert (await session.execute(select(BookingRequest))).scalar_one_or_none() is None
assert convo.outcome == "info_only"
async def test_hallucinated_service_id_is_dropped(session, tenant, emails):
llm = ScriptedLLM(
[
LLMResponse(
text="",
tool_uses=[
ToolUse(
id="tu1",
name="submit_booking_request",
input={
"service_id": "not-a-real-uuid",
"service_name_raw": "nešto čudno",
"client_name": "Ana",
"client_phone": "+38761000111",
"summary": "test",
},
)
],
stop_reason="tool_use",
),
LLMResponse(text="Zahtjev proslijeđen."),
]
)
convo = AgentConversation(session, tenant, channel="chat", llm=llm)
await convo.user_turn("Zakazi mi nešto čudno")
await session.commit()
req = (await session.execute(select(BookingRequest))).scalar_one()
assert req.service_id is None
assert req.service_name_raw == "nešto čudno"
async def test_tool_round_limit_guard(session, tenant):
"""A model stuck in a tool loop cannot spin forever."""
endless = LLMResponse(
text="",
tool_uses=[ToolUse(id="x", name="get_salon_info", input={})],
stop_reason="tool_use",
)
llm = ScriptedLLM([endless] * 20)
convo = AgentConversation(session, tenant, channel="chat", llm=llm)
reply = await convo.user_turn("zdravo")
assert len(llm.calls) == 6 # MAX_TOOL_ROUNDS_PER_TURN
assert reply # graceful fallback text, not an exception
async def test_unknown_tool_returns_error(session, tenant):
llm = ScriptedLLM(
[
LLMResponse(
text="",
tool_uses=[ToolUse(id="x", name="book_now", input={})],
stop_reason="tool_use",
),
LLMResponse(text="Izvinite."),
]
)
convo = AgentConversation(session, tenant, channel="chat", llm=llm)
await convo.user_turn("test")
payload = json.loads(llm.calls[1]["messages"][-1]["content"][0]["content"])
assert "unknown tool" in payload["error"]