Turn strava.py chat into a tool-using agent with runlog search

pyproject.toml

@@ -6,6 +6,7 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "requests>=2.32.5",
+    "scikit-learn",
 ]
 
 [project.optional-dependencies]

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())
 
 

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.
 """

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)

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."