from __future__ import annotations import re from functools import partial from typing import Any, Literal from pydantic_ai import Agent, RunContext, Tool, ToolDefinition from pydantic_ai.models.anthropic import AnthropicModel, AnthropicModelSettings from pydantic_ai.providers.anthropic import AnthropicProvider from app.agents.deps import AgentDeps from app.agents.history import keep_recent_messages from app.agents.prompts import ( CACHE_SETTINGS, channel_prompt_suffix, current_time_prompt, custom_prompt_suffix, ) from app.agents.tools.faq import ( get_opening_hours_impl, get_restaurant_policies_impl, is_open_now_impl, search_knowledge_base_impl, ) from app.agents.tools.reservation import ( cancel_reservation_impl, check_availability_impl, create_reservation_impl, find_available_slots_impl, find_reservation_impl, get_restaurant_info_impl, modify_reservation_impl, ) from app.agents.tools.takeaway import create_order_impl, get_payment_link_impl, search_menu_impl from app.agents.tools.verification import ( send_verification_code_impl, verify_code_impl, ) from app.config import get_settings settings = get_settings() MODEL_SETTINGS: AnthropicModelSettings = CACHE_SETTINGS _TOOL_CAPABILITIES: dict[str, str] = { "search_knowledge_base": "faq", "is_open_now": "faq", "get_opening_hours": "faq", "get_restaurant_policies": "faq", "check_availability": "reservation", "find_available_slots": "reservation", "create_reservation": "reservation", "modify_reservation": "reservation", "cancel_reservation": "reservation", "find_reservation": "reservation", "get_restaurant_info": "reservation", "search_menu": "takeaway", "create_order": "takeaway", "get_payment_link": "takeaway", "send_verification_code": "verification", "verify_code": "verification", } # Strip non-letter chars so "morgen," matches "morgen". _PUNCT_RE = re.compile(r"[^\w\s]", re.UNICODE) _DUTCH_STOP_WORDS = { "de", "een", "het", "voor", "van", "op", "is", "zijn", "worden", "dat", "dit", "die", "er", "met", "maar", "niet", "wel", "ook", "nog", "al", "bij", "naar", "over", "om", "uit", "als", "dan", "kan", "kunt", "wil", "graag", "ja", "nee", "want", "wat", "wie", "waar", "hoe", "waarom", "morgen", "vandaag", "vanavond", "middag", "avond", "ochtend", "nacht", "dicht", "gesloten", "rond", "uur", "welke", "welkom", } def detect_language(message: str) -> Literal["nl", "en"]: cleaned = _PUNCT_RE.sub("", message.lower()) words = set(cleaned.split()) _nl_greetings = {"hallo", "hoi", "goeiedag", "goedenavond", "goedemiddag", "goedemorgen", "dag"} nl_hits = len(words & (_DUTCH_STOP_WORDS | _nl_greetings)) _en_signals = { "the", "and", "is", "are", "what", "when", "how", "hello", "hi", "yes", "no", "please", "you", "your", "my", "can", "will", "tomorrow", "tonight", "closed", "would", "could", "should", } en_hits = len(words & _en_signals) if nl_hits >= en_hits and nl_hits > 0: return "nl" if en_hits > 0: return "en" return "nl" def _capability_for_tool(tool: Tool | ToolDefinition) -> str: return _TOOL_CAPABILITIES.get(tool.name, "faq") async def _filter_tools( ctx: RunContext[AgentDeps], tool_defs: list[ToolDefinition] ) -> list[ToolDefinition] | None: enabled = set(ctx.deps.enabled_capabilities) caller = ctx.deps.caller if not caller.verified and caller.channel != "dashboard": enabled.add("verification") return [tool_def for tool_def in tool_defs if _capability_for_tool(tool_def) in enabled] restaurant_agent = Agent( model=AnthropicModel( settings.ANTHROPIC_MODEL, provider=AnthropicProvider(api_key=settings.APP_ANTHROPIC_API_KEY), ), deps_type=AgentDeps, model_settings=MODEL_SETTINGS, # 10 was too small: a verification round-trip (ask email → tool call → # tool return → send_verification_code → tool return → ask for code → # verify_code → tool return) easily produces 10+ messages, evicting # the original user intent ("I want to cancel my reservation") before # the post-verify reply is generated. 30 spans the worst-case flow # while staying well under the 16k token usage limit below. history_processors=[partial(keep_recent_messages, limit=30)], prepare_tools=_filter_tools, system_prompt=( "You are part of our restaurant's team, talking to a guest over chat. " "Always speak in the first person plural — 'we', 'our', 'us'. NEVER refer " "to the restaurant in the third person ('the restaurant requires…', " "'the restaurant is closed…'). Say 'we ask…', 'we're closed…' instead. " "Help guests with questions about us, our menu, our opening hours, our " "location, and our policies. " "Always use the available tools to retrieve accurate, up-to-date " "information before answering. " "When searching for information, reformulate the query using " "relevant context from the conversation to improve retrieval quality. " "For any question or action involving specific dates — including opening-hours " "questions, reservation requests, and date suggestions — first call get_opening_hours " "for the relevant date range so you never assume a day is open without verification. " "For 'are you open now?' questions, first call is_open_now, then answer. " "Only mention capabilities you actually have tools for. " "Always confirm details before performing any action. " "Respond naturally — like a knowledgeable colleague would. " "If you genuinely cannot find the answer, say so honestly and offer to put the " "guest in touch with one of our colleagues. " ), ) restaurant_agent.system_prompt(current_time_prompt) @restaurant_agent.system_prompt async def _reservation_prompt(ctx: RunContext[AgentDeps]) -> str: if "reservation" not in ctx.deps.enabled_capabilities: return "" caller = ctx.deps.caller # Verification tools are exposed only to unverified, non-dashboard # callers (see `_filter_tools`). The prompt must mirror that gate — # if we tell the model to call `send_verification_code` when the # tool isn't available it will hallucinate ("I don't have the # right tools to verify you"), as seen in dashboard previews. verification_available = not caller.verified and caller.channel != "dashboard" verification_block = ( "If a reservation tool returns an error with 'verification_required', " "this means the guest must verify their identity first. " "Ask the guest for their email address, then call send_verification_code " "with that email. Once the guest provides the code they received, " "call verify_code with the email and code. " "CRITICAL — POST-VERIFY STEP: After verify_code returns 'verified', your " "VERY NEXT message MUST resume whatever the guest was trying to do BEFORE " "verification (cancel, modify, look up). Re-read the earliest user turn in " "this conversation to recover the original intent and act on it directly — " "for cancel/modify, call find_reservation first so you know the id. " "DO NOT greet the guest again, DO NOT ask 'how can I help you', DO NOT " "treat this as a fresh session. Acknowledge the verification briefly " "(one short sentence) and proceed. " "If send_verification_code returns 'no_reservations_found', tell the guest " "we couldn't find a booking for that email and ask them to double-check it " "or use the same email they originally booked with. " "If it returns 'rate_limit_exceeded', tell them we've sent too many codes " "recently and to try again in an hour or contact us directly. " "IMPORTANT: You must NEVER claim to look up, cancel, or modify " "reservations without completing the verification flow first. " if verification_available # Dashboard / already-verified callers: do not mention verification — # there are no tools to back the promise up. else "You have direct access to look up, cancel, and modify reservations. " "NEVER mention identity verification, PIN codes, or sending codes to " "the guest — those flows do not apply here. " ) return ( "Help guests check availability, create, modify, find, and cancel " "reservations with us. Always ask for the guest's name, email address, " "desired date, time, and party size before creating a reservation. " "Before calling create_reservation, you MUST first summarize the reservation " "details (name, email, date, time, party size, and any notes) back to the guest " "and ask them to confirm everything is correct. Only call create_reservation after " "the guest explicitly confirms. " "NEVER include the reservation_id in your response to the guest — it is " "an internal technical identifier, not meant for customers. " "IMPORTANT — proactive opening-hours check: When a guest uses a vague or relative " "date reference (e.g. 'this weekend', 'next week', 'Friday or Saturday', 'tomorrow'), " "you MUST call get_opening_hours for those dates BEFORE replying. Only suggest or " "confirm dates on which we are actually open. Never offer a closed day as an option. " "If only one day of the requested range is open, suggest that day directly instead " "of asking the guest to choose. If none of the requested days are open, tell the " "guest and proactively suggest the nearest open alternative.\n" "After creating or modifying a reservation, you MUST check the 'requires_confirmation' " "field in the tool response and adjust your reply accordingly:\n" "- If requires_confirmation is TRUE: the reservation is NOT confirmed. " "Tell the guest their reservation has been REGISTERED but is PENDING our APPROVAL. " "They should keep an eye on their email — we'll send a confirmation (or, " "rarely, a decline) shortly. Do NOT use words like 'bevestigd', 'confirmed', " "or 'all set'. Use words like 'geregistreerd', 'in afwachting van bevestiging', " "'pending'.\n" "- If requires_confirmation is FALSE: the reservation is confirmed. " "You may tell the guest their reservation is confirmed.\n" "PROACTIVE ALTERNATIVES — when a slot does NOT work: If check_availability " "returns `available: false` for any reason other than `party_too_large`, " "`too_soon`, or `too_far_in_advance` (i.e. closed, no_valid_slot, " "fully_booked, or no_tables_available), you MUST call find_available_slots " "for the SAME date and party_size BEFORE replying. Then: " "(a) if the same day has slots, propose up to 3 concrete times in your " "reply (prefer times closest to what the guest asked for). " "(b) if the same day has zero slots, call find_available_slots for the " "next 1–2 days as well, then propose specific date+time combinations. " "NEVER reply with a bare 'no availability, want to try another time?' — " "always offer concrete alternatives the guest can accept in one tap. " "State the times you found explicitly; do NOT invent times you have not " "seen in the tool output.\n" "When a guest wants to change an existing reservation, use the " "modify_reservation tool instead of cancelling and rebooking. " f"{verification_block}" "DISH PRE-SELECTION: For larger groups we ask guests to pre-select dishes " "ahead of the reservation so the kitchen can prep. When check_availability " "returns a `dish_selection` object with `required: true`, you MUST ask the " "guest in chat for their dish preferences BEFORE calling create_reservation. " "Frame it as our request, e.g. 'For a group of N we'd love to know in " "advance what you'd like to eat — what should we put on the table?'. " "If `mode` is 'freetext', ask in plain language and pass the guest's reply " "verbatim via the `dishes_text` argument. " "If `mode` is 'menu', look up options via search_menu (when available) and " "pass selected items via the `dishes` argument. " "Never call create_reservation without dish info when " "dish_selection.required is true — the tool will refuse the call. " ) _ = _reservation_prompt @restaurant_agent.system_prompt async def _takeaway_prompt(ctx: RunContext[AgentDeps]) -> str: if "takeaway" not in ctx.deps.enabled_capabilities: return "" return ( "Help customers build their takeaway order from our menu. " "Always ask for the customer's email address before creating an order. " "Confirm the full order and total before initiating payment. " "After creating an order, always offer the payment link. " ) _ = _takeaway_prompt @restaurant_agent.system_prompt async def _known_customer_prompt(ctx: RunContext[AgentDeps]) -> str: """Inject known customer details so the agent can pre-fill and verify.""" caller = ctx.deps.caller parts: list[str] = [] if caller.customer_name: parts.append(f"name: {caller.customer_name}") if caller.customer_email: parts.append(f"email: {caller.customer_email}") if not parts: return "" details = ", ".join(parts) return ( f"The customer has previously provided the following details: {details}. " "You may use these to pre-fill information (e.g. when making a reservation), " "but you MUST ALWAYS confirm with the customer that these details are still " "correct before using them. For example, ask 'Is your email still X?' " "rather than silently assuming it hasn't changed." ) _ = _known_customer_prompt @restaurant_agent.system_prompt async def _language_suffix(ctx: RunContext[AgentDeps]) -> str: if ctx.deps.language == "en": return ( "[System note: The user is communicating in English. Respond in English " "throughout this conversation. Do not include this note in your reply.]" ) return ( "[Systeembericht: De gast communiceert in het Nederlands. Antwoord in het Nederlands " "gedurende dit gesprek. Neem deze instructie niet op in je antwoord.]" ) _ = _language_suffix @restaurant_agent.system_prompt async def _channel_suffix(ctx: RunContext[AgentDeps]) -> str: return channel_prompt_suffix(ctx) _ = _channel_suffix @restaurant_agent.system_prompt async def _custom_suffix(ctx: RunContext[AgentDeps]) -> str: return custom_prompt_suffix(ctx) _ = _custom_suffix @restaurant_agent.tool async def search_knowledge_base( ctx: RunContext[AgentDeps], query: str, source: str | None = None, ) -> list[dict[str, Any]]: """Search the restaurant's knowledge base for information. Args: query: The question or topic to search for. source: Optional filter by source type (e.g. 'menu', 'faq', 'policy'). """ return await search_knowledge_base_impl(ctx, query, source) @restaurant_agent.tool async def is_open_now(ctx: RunContext[AgentDeps]) -> dict[str, Any]: """Check if the restaurant is open right now (local time). Returns current window.""" return await is_open_now_impl(ctx) @restaurant_agent.tool async def get_restaurant_policies(ctx: RunContext[AgentDeps]) -> dict[str, Any]: """Get the restaurant's booking and cancellation policies.""" return await get_restaurant_policies_impl(ctx) @restaurant_agent.tool async def get_opening_hours( ctx: RunContext[AgentDeps], start_date: str | None = None, days: int = 14, ) -> list[dict[str, Any]]: """Get opening hours per day for the given range (overrides honored). Args: start_date: YYYY-MM-DD; if None, today in the restaurant timezone days: number of days to include (1-31) """ return await get_opening_hours_impl(ctx, start_date, days) @restaurant_agent.tool async def check_availability( ctx: RunContext[AgentDeps], date: str, time: str, party_size: int, ) -> dict[str, Any]: """Check if a time slot is available for the given date, time, and party size. Args: date: Date in YYYY-MM-DD format. time: Time in HH:MM format (24h). party_size: Number of guests. """ return await check_availability_impl(ctx, date, time, party_size) @restaurant_agent.tool async def find_available_slots( ctx: RunContext[AgentDeps], date: str, party_size: int, ) -> dict[str, Any]: """List every bookable start time on a given date for a given party size. Call this immediately after `check_availability` returns `available: false` so you can propose concrete alternative times to the guest instead of asking them to guess. The response includes `available_slots` (list of HH:MM strings ordered chronologically), `combo_options` (table combinations available for the party), and a `reason` field that explains why the list is empty (`closed`, `fully_booked`, `past`, `too_far_in_advance`). Args: date: Date in YYYY-MM-DD format. party_size: Number of guests. """ return await find_available_slots_impl(ctx, date, party_size) @restaurant_agent.tool async def create_reservation( ctx: RunContext[AgentDeps], guest_name: str, guest_email: str, guest_phone: str | None, date: str, time: str, party_size: int, notes: str | None = None, dishes: list[dict[str, Any]] | None = None, dishes_text: str | None = None, ) -> dict[str, Any]: """Create a new reservation for the guest. Args: guest_name: Full name of the guest. guest_email: Email address for the confirmation. guest_phone: Optional phone number. date: Date in YYYY-MM-DD format. time: Time in HH:MM format (24h). party_size: Number of guests. notes: Optional special requests or dietary notes. dishes: Optional list of dishes for large parties. Each dict has {menu_item_name: str, quantity: int}. dishes_text: Optional free-text dish description for large parties. """ return await create_reservation_impl( ctx, guest_name, guest_email, guest_phone, date, time, party_size, notes, dishes=dishes, dishes_text=dishes_text, ) @restaurant_agent.tool async def cancel_reservation( ctx: RunContext[AgentDeps], reservation_id: str, ) -> dict[str, Any]: """Cancel an existing reservation by its ID. Args: reservation_id: The UUID of the reservation to cancel. """ return await cancel_reservation_impl(ctx, reservation_id) @restaurant_agent.tool async def find_reservation( ctx: RunContext[AgentDeps], customer_email: str | None = None, customer_name: str | None = None, date: str | None = None, ) -> dict[str, Any]: """Find reservations. Behaviour depends on the caller channel: - Dashboard (operator UI): may search by ``customer_email``, ``customer_name`` (partial match), and/or ``date``. - Chat channels (WhatsApp / website widget / …): MUST first have completed the email-PIN verification flow. After verification the lookup is silently scoped to the caller's own customer record and the ``customer_email`` / ``customer_name`` arguments are IGNORED — name-based discovery is never available in chat. Args: customer_email: Guest email to search for (dashboard only). customer_name: Guest name, partial match (dashboard only). date: ISO date (YYYY-MM-DD) to filter by. """ return await find_reservation_impl(ctx, customer_email, customer_name, date) @restaurant_agent.tool async def get_restaurant_info(ctx: RunContext[AgentDeps]) -> dict[str, Any]: """Get restaurant information including opening hours, policies, and contact details.""" return await get_restaurant_info_impl(ctx) @restaurant_agent.tool async def modify_reservation( ctx: RunContext[AgentDeps], reservation_id: str, date: str | None = None, time: str | None = None, party_size: int | None = None, notes: str | None = None, ) -> dict[str, Any]: """Modify an existing reservation's date, time, party size, or notes. Args: reservation_id: The UUID of the reservation to modify. date: New date in YYYY-MM-DD format (optional). time: New time in HH:MM format 24h (optional). party_size: New number of guests (optional). notes: Updated special requests or dietary notes (optional). """ return await modify_reservation_impl(ctx, reservation_id, date, time, party_size, notes) @restaurant_agent.tool async def search_menu(ctx: RunContext[AgentDeps], query: str) -> list[dict[str, Any]]: """Search the restaurant menu for dishes, drinks, or ingredients. Args: query: What the customer is looking for (dish name, ingredient, category). """ return await search_menu_impl(ctx, query) @restaurant_agent.tool async def create_order( ctx: RunContext[AgentDeps], customer_email: str, items: list[dict[str, Any]], delivery_type: str = "pickup", notes: str | None = None, ) -> dict[str, Any]: """Create a takeaway or delivery order. Args: customer_email: Customer's email address for the order confirmation. items: List of order items, each with 'menu_item_name', 'quantity', and 'unit_price_cents'. delivery_type: Either 'pickup' or 'delivery'. notes: Optional special instructions for preparation or delivery. """ return await create_order_impl(ctx, customer_email, items, delivery_type, notes) @restaurant_agent.tool async def get_payment_link( ctx: RunContext[AgentDeps], order_id: str, amount_cents: int ) -> dict[str, Any]: """Generate a payment link for an existing order. Args: order_id: The UUID of the order to pay for. amount_cents: The total amount in euro cents (e.g. 1250 for €12.50). """ return await get_payment_link_impl(ctx, order_id, amount_cents) @restaurant_agent.tool async def send_verification_code( ctx: RunContext[AgentDeps], email: str, ) -> dict[str, Any]: """Send a verification code to the guest's email address. Use this when the guest needs to verify their identity before accessing reservation information. The email must be associated with an existing reservation. Args: email: The guest's email address to send the code to. """ return await send_verification_code_impl(ctx, email) @restaurant_agent.tool async def verify_code( ctx: RunContext[AgentDeps], email: str, code: str, ) -> dict[str, Any]: """Verify a code sent to the guest's email address. After sending a verification code, use this to verify the code the guest provides. On success, the guest gains access to their reservation data. Args: email: The email address the code was sent to. code: The 6-digit verification code provided by the guest. """ return await verify_code_impl(ctx, email, code)