diff --git a/pyproject.toml b/pyproject.toml index 58fd7a7..406a50c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,7 @@ readme = "README.md" requires-python = ">=3.13" dependencies = [ "requests>=2.32.5", + "scikit-learn", ] [project.optional-dependencies] diff --git a/strava.py b/strava.py index 57dbc22..fb29415 100644 --- a/strava.py +++ b/strava.py @@ -5,6 +5,7 @@ # "stravalib==2.4", # "llm==0.31", # "python-dotenv==1.0.1", +# "scikit-learn", # ] # /// @@ -22,7 +23,12 @@ from dotenv import load_dotenv from stravalib import Client +# Imported as a module, not `from strava_tools import RUNLOG_PATH`: --runlog overrides the log +# path by reassigning strava_tools.RUNLOG_PATH, and search_runlog reads that module global at +# call time. A `from`-import would bind a separate name here that rebinding never reaches. +import strava_tools from strava_prompts import SYSTEM_PROMPT +from strava_tools import search_runlog TOKEN_PATH = Path.home() / ".strava_tokens.json" @@ -135,11 +141,21 @@ def main(): default=None, help="llm model to use (default: your configured llm default model)", ) + parser.add_argument( + "--runlog", + default=None, + help="Path to the running log the search_runlog tool reads (default: RUNLOG.txt)", + ) args = parser.parse_args() if args.end and not args.start: parser.error("--end requires --start") + # Point the search_runlog tool at a custom log file if requested (default lives in + # strava_tools.RUNLOG_PATH). Set before the tool is ever invoked in the chat loop below. + if args.runlog: + strava_tools.RUNLOG_PATH = Path(args.runlog) + # Load STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, and OPENAI_API_KEY from a local .env # (gitignored; see .env.example) so they don't have to be exported in the shell. # @@ -194,8 +210,9 @@ def main(): # so this script relies on whatever model/key the user has set up via `llm`. # - model.conversation() is a multi-turn object that retains history automatically, so # follow-up questions keep the context of earlier turns (and the activity below). + # model = llm.get_model(args.model) - conversation = model.conversation() + conversation = model.conversation(tools=[search_runlog]) # Seed the activity into context via the system prompt so it's available without a # round-trip — nothing is sent until the user asks a question. @@ -218,19 +235,20 @@ def main(): if user_input.lower() in ("exit", "quit"): break - # conversation.prompt() sends the turn to the model; the system prompt (carrying the - # activity) is passed only on the first turn, since the conversation retains history. + # conversation.chain() sends the turn and, unlike prompt(), auto-executes any + # search_runlog tool calls the model makes and re-prompts until it produces a final + # answer. The system prompt (carrying the activity) is passed only on the first turn, + # since the conversation retains history. if first_turn: - response = conversation.prompt(user_input, system=system) + response = conversation.chain(user_input, system=system) else: - response = conversation.prompt(user_input) + response = conversation.chain(user_input) first_turn = False - # response.text() resolves and returns the full reply as a single string. - # Alternative: an llm response is iterable, so you can stream each text chunk as it - # arrives with `for chunk in response: print(chunk, end="", flush=True)`. + # response.text() resolves the whole chain (including tool round-trips) and returns the + # final reply as a single string. print(response.text()) diff --git a/strava_prompts.py b/strava_prompts.py index 0e3ae04..2b13caf 100644 --- a/strava_prompts.py +++ b/strava_prompts.py @@ -12,4 +12,10 @@ Convert units to whatever is natural for the user (km or miles, min/km or min/mile) and show your reasoning briefly. Be concise and specific, cite the numbers you used, and say so plainly when the data doesn't support a conclusion. + +You also have a search_runlog tool over the athlete's past running log (prior runs and +coaching notes). Call it when history, trends, or earlier guidance would strengthen an +answer -- for example when the user asks how this run compares to past efforts, or when +recalling advice you gave before. Ground any claim about the athlete's history in what the +search returns; if it returns nothing relevant, say so rather than inventing past runs. """ diff --git a/strava_tools.py b/strava_tools.py new file mode 100644 index 0000000..10169b9 --- /dev/null +++ b/strava_tools.py @@ -0,0 +1,120 @@ +# Tools for the strava.py chat agent, exposed to the LLM via `llm`'s tool-calling +# (model.conversation(tools=[...]) + conversation.chain(...)). + +import re +from dataclasses import dataclass +from pathlib import Path + +from sklearn.feature_extraction.text import TfidfVectorizer + +# -------------------------------------------- Tool: Search the athlete's past running log + +# Default runlog location (the RUNLOG.txt next to this script). strava.py can override this +# module attribute from its --runlog flag, and tests point it at a fixture file. +RUNLOG_PATH = Path(__file__).parent / "RUNLOG.txt" + + +# `llm` builds each tool's schema from the function signature and uses its docstring +# as the description shown to the model, so search_runlog's docstring is written for the model to read. + +def search_runlog(query: str) -> str: + """Search the athlete's past running log for entries relevant to `query` + (for example: 'half marathon pacing', 'late-run fade', 'hill workouts'). + Returns the most relevant past run notes and coaching advice as text. Call this + to compare the current activity against past runs or to recall prior guidance, + and ground any claims about the athlete's history in what it returns.""" + text = Path(RUNLOG_PATH).read_text() + + # Structure mirrors the repo's fetch_*/process_*/format_* seam: the pure helpers + # (parse_runlog, search_entries, format_entries) do no I/O and are the test targets; + # search_runlog is the thin tool wrapper that reads the file and glues them together. + return format_entries(search_entries(parse_runlog(text), query)) + + +# frozen=True keeps the value semantics a parsed entry wants -- immutable, and compared by +# field so tests can assert on a whole entry -- without being a tuple: an entry won't compare +# equal to a bare ("title", "body") pair or silently unpack, and adding a field later (a parsed +# date, say, normalized in __post_init__) won't shift anyone's positional access. +@dataclass(frozen=True) +class RunlogEntry: + title: str + body: str + + +def parse_runlog(text: str) -> list[RunlogEntry]: + """Split raw runlog text into entries. + + An entry starts at a "title" line -- a markdown heading ("# ...") -- and runs until the + next heading. Everything in between (blank lines, bullets, prose) forms the body, so an + entry can hold any content as long as it doesn't open a line with "#". The leading "#" + and surrounding whitespace are stripped from the title. Text before the first heading + belongs to no entry and is dropped. + """ + # Splitting on the heading pattern hands us each title and body whole, so they go straight + # into the immutable RunlogEntry -- no line-by-line accumulator to build them up in. + # + # re.split with a capturing group keeps what it captures instead of discarding it, so the + # result interleaves captured headings with the text between them: + # [text before the first heading, title, body, title, body, ...] + # Only "(.*)" is captured, so the leading "#"s and spaces go away with the rest of the + # separator. After index 0 the list strictly alternates, hence parts[1::2] is every title + # and parts[2::2] every body, and zip pairs each title with the body that follows it. + # parts[0] is in neither slice -- that's how the pre-heading text gets dropped. + # (n headings always yield 2n+1 parts, a trailing heading getting an empty body, so the + # two slices are always the same length and zip never truncates one away.) + parts = re.split(r"(?m)^[ \t]*#+[ \t]*(.*)$", text) + return [ + RunlogEntry(title.strip(), body.strip()) + for title, body in zip(parts[1::2], parts[2::2]) + ] + + +def search_entries( + entries: list[RunlogEntry], query: str, limit: int = 3 +) -> list[RunlogEntry]: + """Rank entries by TF-IDF cosine similarity to the query and return the top `limit`. + + TF-IDF weights each term by how rare it is across the log, so a match on a distinctive + word (e.g. 'fartlek') outranks a match on a ubiquitous one (e.g. 'run') -- the thing + plain term-counting got wrong. Entries sharing no terms with the query are dropped; + ties keep the log's original (chronological) order. + """ + # Decision: scikit-learn's TfidfVectorizer instead of a hand-rolled scorer. It gives the + # simplest correct code -- it tokenizes, lowercases, and L2-normalizes for us -- at the + # cost of a heavy numpy/scipy dependency, declared in BOTH strava.py's script block and + # pyproject.toml because standalone `uv run strava.py` and `uv run pytest` are different + # environments. We chose the smaller function over the lighter dependency on purpose. + # + # Decision: title text is folded into the document but NOT up-weighted. Runlog titles are + # just a date + run type, so a title-only boost wasn't worth the extra code; if titles + # ever need to rank higher, double them here (f"{e.title} {e.title}\n{e.body}"). + # + # BM25 is the natural next step if ranking needs to improve: it keeps this TF-IDF backbone + # but adds term-frequency saturation (repeated terms give diminishing returns) and + # document-length normalization (long entries don't win just by being long). To switch, + # replace only the vectorize-and-score lines below -- keeping the guard, the `> 0` filter, + # and the sort/limit tail -- with either rank-bm25 + # (BM25Okapi(tokenized_docs).get_scores(query.split())) or, for a zero-dependency route + # more in keeping with this repo, SQLite FTS5's built-in bm25(). + if not entries: + # TfidfVectorizer().fit_transform([]) raises "empty vocabulary"; guard it. + return [] + docs = [f"{e.title}\n{e.body}" for e in entries] + vectorizer = TfidfVectorizer() + matrix = vectorizer.fit_transform(docs) + # TF-IDF rows are L2-normalized, so this dot product is exactly the cosine similarity. + # An empty or all-unknown query yields a zero vector, hence all-zero scores -> no matches. + scores = (vectorizer.transform([query]) @ matrix.T).toarray().ravel() + # (-scores).argsort is descending; kind="stable" makes earlier entries win ties. + order = (-scores).argsort(kind="stable") + # The `> 0` filter is load-bearing: it makes a no-match query return [] (and thus + # format_entries' "no matching entries" sentinel) rather than unrelated entries. + return [entries[i] for i in order if scores[i] > 0][:limit] + + +def format_entries(entries: list[RunlogEntry]) -> str: + """Render matched entries back to plain text for the model (self-contained: the returned + string is the only thing that survives into the next turn, so include titles + bodies).""" + if not entries: + return "No matching runlog entries found." + return "\n\n".join(f"{e.title}\n{e.body}".rstrip() for e in entries) \ No newline at end of file diff --git a/test_strava_tools.py b/test_strava_tools.py new file mode 100644 index 0000000..2258349 --- /dev/null +++ b/test_strava_tools.py @@ -0,0 +1,74 @@ +from strava_tools import ( + RunlogEntry, + format_entries, + parse_runlog, + search_entries, +) + +# A small multi-entry fixture in the RUNLOG.txt shape: a "# " heading, a blank line, bullets. +RUNLOG = """# April 26 Half Marathon Run + +- The fade in watts alongside the pace drop suggests your legs gave out. Add a weekly + long run finishing at goal pace to build muscular endurance. + +# May 10 Hill Repeats + +- Strong power on the climbs; turnover held late. Hills are paying off. + +# May 17 Easy Recovery Run + +- Nice and controlled, heart rate stayed in zone 2 the whole way. +""" + + +def test_parse_runlog_splits_entries(): + entries = parse_runlog(RUNLOG) + assert [e.title for e in entries] == [ + "April 26 Half Marathon Run", + "May 10 Hill Repeats", + "May 17 Easy Recovery Run", + ] + assert entries[0].body.startswith("- The fade in watts") + + +def test_parse_runlog_single_entry_and_trailing_blanks(): + entries = parse_runlog("# Solo Run\n\n- Felt good.\n\n\n") + assert len(entries) == 1 + assert entries[0] == RunlogEntry(title="Solo Run", body="- Felt good.") + + +def test_parse_runlog_ignores_text_before_first_heading(): + entries = parse_runlog("stray preamble\n\n# Solo Run\n\nFelt good.\n") + assert entries == [RunlogEntry(title="Solo Run", body="Felt good.")] + + +# TODO: Run these against the "# heading" parse_runlog -- the fixtures above were converted +# from the old bullet-delimited format and test_parse_runlog_ignores_text_before_first_heading +# is new, so neither has been executed yet. +# TODO: Add a case for a heading with no body ("# Heat Advisory" appears that way in the real +# RUNLOG.txt, as does its leading blank line). Those are the two edges of the re.split parse -- +# the dropped parts[0] and the empty trailing body -- and only the first is covered above. + + +# The search_entries tests cover only the wrapping logic we wrote around TfidfVectorizer -- +# the empty-corpus guard, the score > 0 filter (which drives the no-match sentinel), and the +# limit slice. They deliberately do NOT assert ranking order: that's scikit-learn's TF-IDF +# behavior, which we don't own and shouldn't re-test. + + +def test_search_entries_empty_corpus(): + assert search_entries([], "anything") == [] + + +def test_search_entries_no_shared_terms_returns_empty(): + entries = [RunlogEntry("A", "hill repeats"), RunlogEntry("B", "recovery jog")] + assert search_entries(entries, "cycling swimming") == [] + + +def test_search_entries_respects_limit(): + entries = [RunlogEntry(t, "tempo run") for t in ("A", "B", "C")] + assert len(search_entries(entries, "tempo", limit=2)) == 2 + + +def test_format_entries_no_match_sentinel(): + assert format_entries([]) == "No matching runlog entries found."