|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from datetime import datetime |
| 4 | +from typing import Annotated, Any, List |
| 5 | + |
| 6 | +from agents import Agent, RunContextWrapper, StopAtTools, function_tool |
| 7 | +from chatkit.agents import AgentContext |
| 8 | +from chatkit.types import ( |
| 9 | + AssistantMessageContent, |
| 10 | + AssistantMessageItem, |
| 11 | + ProgressUpdateEvent, |
| 12 | + ThreadItemDoneEvent, |
| 13 | +) |
| 14 | +from pydantic import BaseModel, ConfigDict, Field |
| 15 | + |
| 16 | +from ..data.event_store import EventRecord, EventStore |
| 17 | +from ..memory_store import MemoryStore |
| 18 | +from ..request_context import RequestContext |
| 19 | +from ..widgets.event_list_widget import build_event_list_widget |
| 20 | + |
| 21 | +INSTRUCTIONS = """ |
| 22 | + You help Foxhollow residents discover local happenings. When a reader asks for events, |
| 23 | + search the curated calendar, call out dates and notable details, and keep recommendations brief. |
| 24 | +
|
| 25 | + Use the available tools deliberately: |
| 26 | + - Call `list_available_event_keywords` to get the full set of event keywords and categories, |
| 27 | + fuzzy match the reader's phrasing to the closest options (case-insensitive, partial matches are ok), |
| 28 | + then feed those terms into a keyword search instead of relying on hard-coded synonyms. |
| 29 | + - If they mention a specific date (YYYY-MM-DD), start with `search_events_by_date`. |
| 30 | + - If they reference a day of the week, try `search_events_by_day_of_week`. |
| 31 | + - For general vibes (e.g., “family friendly night markets”), use `search_events_by_keyword` |
| 32 | + so the search spans titles, categories, locations, and curated keywords. |
| 33 | +
|
| 34 | + Whenever a search tool returns more than one event immediately call `show_event_list_widget` |
| 35 | + with those results before sending your final text, along with a 1-sentence message explaining why these events were selected. |
| 36 | + This ensures every response ships with the timeline widget. |
| 37 | + Cite event titles in bold, mention the date, and highlight one delightful detail when replying. |
| 38 | +
|
| 39 | + When the user explicitly asks for more details on the events, you MUST describe the events in natural language |
| 40 | + without using the `show_event_list_widget` tool. |
| 41 | +""" |
| 42 | + |
| 43 | +MODEL = "gpt-4.1-mini" |
| 44 | + |
| 45 | + |
| 46 | +class EventFinderContext(AgentContext): |
| 47 | + model_config = ConfigDict(arbitrary_types_allowed=True) |
| 48 | + store: Annotated[MemoryStore, Field(exclude=True)] |
| 49 | + events: Annotated[EventStore, Field(exclude=True)] |
| 50 | + request_context: Annotated[RequestContext, Field(exclude=True, default_factory=RequestContext)] |
| 51 | + |
| 52 | + |
| 53 | +class EventKeywords(BaseModel): |
| 54 | + keywords: List[str] |
| 55 | + |
| 56 | + |
| 57 | +@function_tool( |
| 58 | + description_override="Find scheduled events happening on a specific date (YYYY-MM-DD)." |
| 59 | +) |
| 60 | +async def search_events_by_date( |
| 61 | + ctx: RunContextWrapper[EventFinderContext], |
| 62 | + date: str, |
| 63 | +) -> dict[str, Any]: |
| 64 | + print("[TOOL CALL] search_events_by_date", date) |
| 65 | + if not date: |
| 66 | + raise ValueError("Provide a valid date in YYYY-MM-DD format.") |
| 67 | + await ctx.context.stream(ProgressUpdateEvent(text=f"Looking up events on {date}")) |
| 68 | + records = ctx.context.events.search_by_date(date) |
| 69 | + return {"events": _events_to_json(records)} |
| 70 | + |
| 71 | + |
| 72 | +@function_tool(description_override="List events occurring on a given day of the week.") |
| 73 | +async def search_events_by_day_of_week( |
| 74 | + ctx: RunContextWrapper[EventFinderContext], |
| 75 | + day: str, |
| 76 | +) -> dict[str, Any]: |
| 77 | + print("[TOOL CALL] search_events_by_day_of_week", day) |
| 78 | + if not day: |
| 79 | + raise ValueError("Provide a day of the week to search for (e.g., Saturday).") |
| 80 | + await ctx.context.stream(ProgressUpdateEvent(text=f"Checking {day} events")) |
| 81 | + records = ctx.context.events.search_by_day_of_week(day) |
| 82 | + return {"events": _events_to_json(records)} |
| 83 | + |
| 84 | + |
| 85 | +@function_tool( |
| 86 | + description_override="Search events with general keywords (title, category, location, or details)." |
| 87 | +) |
| 88 | +async def search_events_by_keyword( |
| 89 | + ctx: RunContextWrapper[EventFinderContext], |
| 90 | + keywords: List[str], |
| 91 | +) -> dict[str, Any]: |
| 92 | + print("[TOOL CALL] search_events_by_keyword", keywords) |
| 93 | + tokens = [keyword.strip() for keyword in keywords if keyword and keyword.strip()] |
| 94 | + if not tokens: |
| 95 | + raise ValueError("Provide at least one keyword to search for.") |
| 96 | + label = ", ".join(tokens) |
| 97 | + await ctx.context.stream(ProgressUpdateEvent(text=f"Searching for: {label}")) |
| 98 | + records = ctx.context.events.search_by_keyword(tokens) |
| 99 | + return {"events": _events_to_json(records)} |
| 100 | + |
| 101 | + |
| 102 | +@function_tool(description_override="List all unique event keywords and categories.") |
| 103 | +async def list_available_event_keywords( |
| 104 | + ctx: RunContextWrapper[EventFinderContext], |
| 105 | +) -> EventKeywords: |
| 106 | + print("[TOOL CALL] list_available_event_keywords") |
| 107 | + await ctx.context.stream(ProgressUpdateEvent(text="Referencing available event keywords...")) |
| 108 | + return EventKeywords(keywords=ctx.context.events.list_available_keywords()) |
| 109 | + |
| 110 | + |
| 111 | +@function_tool(description_override="Show a timeline-styled widget for a provided set of events.") |
| 112 | +async def show_event_list_widget( |
| 113 | + ctx: RunContextWrapper[EventFinderContext], |
| 114 | + events: List[EventRecord], |
| 115 | + message: str | None = None, |
| 116 | +): |
| 117 | + print("[TOOL CALL] show_event_list_widget", events) |
| 118 | + records: List[EventRecord] = [event for event in events if event] |
| 119 | + |
| 120 | + # Gracefully handle case where agent mistakenly calls this tool with no events. |
| 121 | + # Otherewise, since the agent is configured to stop running after this tool call, the user |
| 122 | + # will never see further output. |
| 123 | + if not records: |
| 124 | + fallback = message or "I couldn't find any events that match that search." |
| 125 | + await ctx.context.stream( |
| 126 | + ThreadItemDoneEvent( |
| 127 | + item=AssistantMessageItem( |
| 128 | + thread_id=ctx.context.thread.id, |
| 129 | + id=ctx.context.generate_id("message"), |
| 130 | + created_at=datetime.now(), |
| 131 | + content=[AssistantMessageContent(text=fallback)], |
| 132 | + ), |
| 133 | + ) |
| 134 | + ) |
| 135 | + |
| 136 | + widget = build_event_list_widget(records) |
| 137 | + copy_text = ", ".join(filter(None, (event.title for event in records))) |
| 138 | + await ctx.context.stream_widget(widget, copy_text=copy_text or "Local events") |
| 139 | + |
| 140 | + summary = message or "Here are the events that match your request." |
| 141 | + await ctx.context.stream( |
| 142 | + ThreadItemDoneEvent( |
| 143 | + item=AssistantMessageItem( |
| 144 | + thread_id=ctx.context.thread.id, |
| 145 | + id=ctx.context.generate_id("message"), |
| 146 | + created_at=datetime.now(), |
| 147 | + content=[AssistantMessageContent(text=summary)], |
| 148 | + ), |
| 149 | + ) |
| 150 | + ) |
| 151 | + |
| 152 | + |
| 153 | +def _events_to_json(events: List[EventRecord]) -> List[dict[str, Any]]: |
| 154 | + """Convert EventRecord models to JSON-safe dicts for tool responses.""" |
| 155 | + return [event.model_dump(mode="json", by_alias=True) for event in events] |
| 156 | + |
| 157 | + |
| 158 | +class EventSummaryContext(AgentContext): |
| 159 | + model_config = ConfigDict(arbitrary_types_allowed=True) |
| 160 | + store: Annotated[MemoryStore, Field(exclude=True)] |
| 161 | + events: Annotated[EventStore, Field(exclude=True)] |
| 162 | + request_context: Annotated[RequestContext, Field(exclude=True, default_factory=RequestContext)] |
| 163 | + |
| 164 | + |
| 165 | +event_finder_agent = Agent[EventFinderContext]( |
| 166 | + model=MODEL, |
| 167 | + name="Foxhollow Event Finder", |
| 168 | + instructions=INSTRUCTIONS, |
| 169 | + tools=[ |
| 170 | + search_events_by_date, |
| 171 | + search_events_by_day_of_week, |
| 172 | + search_events_by_keyword, |
| 173 | + list_available_event_keywords, |
| 174 | + show_event_list_widget, |
| 175 | + ], |
| 176 | + # Stop inference after showing the event list widget to prevent content from being repeated in a continued response. |
| 177 | + tool_use_behavior=StopAtTools(stop_at_tool_names=[show_event_list_widget.name]), |
| 178 | +) |
0 commit comments