feat(chat): implement web chat interface with accessibility
Core implementation: - chat_app.py: Chainlit entry point with @cl.on_chat_start, @cl.on_message (streaming via llm.stream_complete), @cl.on_chat_end Reuses Config, KnowledgeBase, ConversationStore, AdminNotifier from src/ Handles registration completion, post-completion updates, new-child flow - src/llm.py: add stream_complete() generator (litellm stream=True) alongside existing complete(); tests added in tests/test_llm.py - src/agent/response_parser.py: extract parse_llm_response(), apply_updates(), fallback_message() from EmailAgent into shared module EmailAgent now delegates to these functions (no logic change) Chainlit configuration: - chainlit.toml: telemetry off, German default, custom CSS + JS paths - chainlit.md: German welcome page with playgroup info Accessibility (WCAG 2.1 AA): - public/custom.css: contrast overrides (≥4.5:1), prefers-reduced-motion (static "…" replaces animated dots), skip link styles, 100dvh fix - public/accessibility.js: MutationObserver injects aria-live="polite" on message list, focus management after agent replies, skip link element Other: - .gitignore: add .chainlit/ (Chainlit runtime, auto-generated) - openspec/config.yaml: populate context field with tech stack - openspec/changes/implement-web-chat/tasks.md: mark completed tasks 95 tests pass. https://claude.ai/code/session_01SUWzMzFvSfWiHXA2p6rPg9
This commit is contained in:
Claude
+5
-75
@@ -1,17 +1,16 @@
|
||||
"""EmailAgent — the channel-agnostic conversation orchestrator."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from ..models.conversation import ConversationState, ChatMessage
|
||||
from ..models.registration import BookingDay, RegistrationData
|
||||
from ..models.registration import RegistrationData
|
||||
from .. import llm
|
||||
from ..knowledge_base.loader import KnowledgeBase
|
||||
from ..storage.json_store import ConversationStore, normalize_email, _diff_registrations
|
||||
from ..notifications.notifier import AdminNotifier
|
||||
from .prompts import build_system_prompt
|
||||
from .response_parser import apply_updates, fallback_message, parse_llm_response
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -192,79 +191,10 @@ class EmailAgent:
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _parse_llm_response(self, content: str) -> dict:
|
||||
"""Extract the JSON payload from the LLM's raw output."""
|
||||
text = content.strip()
|
||||
|
||||
fence_match = re.match(r"^```(?:json)?\s*\n(.*?)\n```\s*$", text, re.DOTALL)
|
||||
if fence_match:
|
||||
text = fence_match.group(1).strip()
|
||||
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
brace_match = re.search(r"\{.*\}", text, re.DOTALL)
|
||||
if brace_match:
|
||||
try:
|
||||
return json.loads(brace_match.group())
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
logger.warning("Could not parse LLM response as JSON — using raw text as reply.")
|
||||
return {
|
||||
"reply": content,
|
||||
"intent": "question",
|
||||
"updates": {},
|
||||
"next_step": "greeting",
|
||||
"registration_complete": False,
|
||||
"language": "de",
|
||||
}
|
||||
return parse_llm_response(content)
|
||||
|
||||
def _fallback_message(self, state: ConversationState) -> str:
|
||||
if state.language == "en":
|
||||
return (
|
||||
"I'm sorry, I'm having a technical issue right now. "
|
||||
"Please try again in a moment or contact us directly."
|
||||
)
|
||||
return (
|
||||
"Entschuldigung, ich habe gerade ein technisches Problem. "
|
||||
"Bitte versuche es gleich nochmal oder kontaktiere uns direkt."
|
||||
)
|
||||
return fallback_message(state.language)
|
||||
|
||||
def _apply_updates(self, state: ConversationState, updates: dict) -> None:
|
||||
"""Write extracted field values into the RegistrationData object."""
|
||||
reg = state.registration
|
||||
|
||||
field_map = {
|
||||
"child.fullName": lambda v: setattr(reg.child, "full_name", v),
|
||||
"child.dateOfBirth": lambda v: setattr(reg.child, "date_of_birth", v),
|
||||
"child.specialNeeds": lambda v: setattr(reg.child, "special_needs", v),
|
||||
"parentGuardian.fullName": lambda v: (
|
||||
setattr(reg.parent_guardian, "full_name", v),
|
||||
setattr(state, "parent_name", v),
|
||||
),
|
||||
"parentGuardian.streetAddress": lambda v: setattr(reg.parent_guardian, "street_address", v),
|
||||
"parentGuardian.postalCode": lambda v: setattr(reg.parent_guardian, "postal_code", str(v)),
|
||||
"parentGuardian.city": lambda v: setattr(reg.parent_guardian, "city", v),
|
||||
"parentGuardian.phone": lambda v: setattr(reg.parent_guardian, "phone", v),
|
||||
"parentGuardian.email": lambda v: setattr(reg.parent_guardian, "email", v),
|
||||
"emergencyContact.fullName": lambda v: setattr(reg.emergency_contact, "full_name", v),
|
||||
"emergencyContact.phone": lambda v: setattr(reg.emergency_contact, "phone", v),
|
||||
}
|
||||
|
||||
for key, value in updates.items():
|
||||
if value is None:
|
||||
continue
|
||||
if key in field_map:
|
||||
field_map[key](value)
|
||||
elif key == "booking.playgroupTypes" and isinstance(value, list):
|
||||
reg.booking.playgroup_types = value
|
||||
elif key == "booking.selectedDays" and isinstance(value, list):
|
||||
reg.booking.selected_days = [
|
||||
BookingDay(day=d["day"], type=d["type"])
|
||||
for d in value
|
||||
if isinstance(d, dict) and "day" in d and "type" in d
|
||||
]
|
||||
else:
|
||||
logger.debug("Unknown update key ignored: %s", key)
|
||||
apply_updates(state, updates)
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Parse and apply LLM JSON responses — shared between email and chat channels."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
from ..models.conversation import ConversationState
|
||||
from ..models.registration import BookingDay
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def parse_llm_response(content: str) -> dict:
|
||||
"""Extract the JSON payload from the LLM's raw output.
|
||||
|
||||
Tries three strategies in order:
|
||||
1. Entire content is a fenced code block (```json ... ```)
|
||||
2. Entire content is a bare JSON object
|
||||
3. JSON object embedded somewhere in the text
|
||||
|
||||
Falls back to wrapping raw text as a ``reply`` if nothing parses.
|
||||
"""
|
||||
text = content.strip()
|
||||
|
||||
fence_match = re.match(r"^```(?:json)?\s*\n(.*?)\n```\s*$", text, re.DOTALL)
|
||||
if fence_match:
|
||||
text = fence_match.group(1).strip()
|
||||
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
brace_match = re.search(r"\{.*\}", text, re.DOTALL)
|
||||
if brace_match:
|
||||
try:
|
||||
return json.loads(brace_match.group())
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
logger.warning("Could not parse LLM response as JSON — using raw text as reply.")
|
||||
return {
|
||||
"reply": content,
|
||||
"intent": "question",
|
||||
"updates": {},
|
||||
"next_step": "greeting",
|
||||
"registration_complete": False,
|
||||
"language": "de",
|
||||
}
|
||||
|
||||
|
||||
def apply_updates(state: ConversationState, updates: dict) -> None:
|
||||
"""Write extracted field values into the RegistrationData on *state*."""
|
||||
reg = state.registration
|
||||
|
||||
field_map = {
|
||||
"child.fullName": lambda v: setattr(reg.child, "full_name", v),
|
||||
"child.dateOfBirth": lambda v: setattr(reg.child, "date_of_birth", v),
|
||||
"child.specialNeeds": lambda v: setattr(reg.child, "special_needs", v),
|
||||
"parentGuardian.fullName": lambda v: (
|
||||
setattr(reg.parent_guardian, "full_name", v),
|
||||
setattr(state, "parent_name", v),
|
||||
),
|
||||
"parentGuardian.streetAddress": lambda v: setattr(reg.parent_guardian, "street_address", v),
|
||||
"parentGuardian.postalCode": lambda v: setattr(reg.parent_guardian, "postal_code", str(v)),
|
||||
"parentGuardian.city": lambda v: setattr(reg.parent_guardian, "city", v),
|
||||
"parentGuardian.phone": lambda v: setattr(reg.parent_guardian, "phone", v),
|
||||
"parentGuardian.email": lambda v: setattr(reg.parent_guardian, "email", v),
|
||||
"emergencyContact.fullName": lambda v: setattr(reg.emergency_contact, "full_name", v),
|
||||
"emergencyContact.phone": lambda v: setattr(reg.emergency_contact, "phone", v),
|
||||
}
|
||||
|
||||
for key, value in updates.items():
|
||||
if value is None:
|
||||
continue
|
||||
if key in field_map:
|
||||
field_map[key](value)
|
||||
elif key == "booking.playgroupTypes" and isinstance(value, list):
|
||||
reg.booking.playgroup_types = value
|
||||
elif key == "booking.selectedDays" and isinstance(value, list):
|
||||
reg.booking.selected_days = [
|
||||
BookingDay(day=d["day"], type=d["type"])
|
||||
for d in value
|
||||
if isinstance(d, dict) and "day" in d and "type" in d
|
||||
]
|
||||
else:
|
||||
logger.debug("Unknown update key ignored: %s", key)
|
||||
|
||||
|
||||
def fallback_message(language: str) -> str:
|
||||
"""Return a safe error message in the parent's detected language."""
|
||||
if language == "en":
|
||||
return (
|
||||
"I'm sorry, I'm having a technical issue right now. "
|
||||
"Please try again in a moment or contact us directly."
|
||||
)
|
||||
return (
|
||||
"Entschuldigung, ich habe gerade ein technisches Problem. "
|
||||
"Bitte versuche es gleich nochmal oder kontaktiere uns direkt."
|
||||
)
|
||||
+26
@@ -1,5 +1,7 @@
|
||||
"""LLM completion via litellm — supports any provider with a single call."""
|
||||
|
||||
from collections.abc import Generator
|
||||
|
||||
import litellm
|
||||
|
||||
|
||||
@@ -20,3 +22,27 @@ def complete(model: str, system: str, messages: list) -> str:
|
||||
api_messages += [{"role": m.role, "content": m.content} for m in messages]
|
||||
response = litellm.completion(model=model, messages=api_messages, max_tokens=2048)
|
||||
return response.choices[0].message.content
|
||||
|
||||
|
||||
def stream_complete(
|
||||
model: str, system: str, messages: list
|
||||
) -> Generator[str, None, None]:
|
||||
"""Call any LLM with streaming and yield text chunks as they arrive.
|
||||
|
||||
Args:
|
||||
model: litellm model string (same format as ``complete``).
|
||||
system: System prompt text.
|
||||
messages: List of objects with .role and .content attributes.
|
||||
|
||||
Yields:
|
||||
Non-empty text chunks from the model's streamed response.
|
||||
"""
|
||||
api_messages = [{"role": "system", "content": system}]
|
||||
api_messages += [{"role": m.role, "content": m.content} for m in messages]
|
||||
response = litellm.completion(
|
||||
model=model, messages=api_messages, max_tokens=2048, stream=True
|
||||
)
|
||||
for chunk in response:
|
||||
delta = chunk.choices[0].delta.content
|
||||
if delta:
|
||||
yield delta
|
||||
|
||||
Reference in New Issue
Block a user