Spaces:
Running
Running
File size: 2,885 Bytes
e6ce96e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
search_episode_data tool — BM25 + semantic search over accumulated episode response data.
Searches all request/response bodies from prior curl_exec calls in this episode.
"""
from __future__ import annotations
import json
import re
from typing import Any
def search_episode_data(query: str, episode_store: dict) -> list[dict]:
"""
Hybrid BM25 + keyword search over episode accumulated response bodies.
Args:
query: Keyword or natural language query (e.g. "Radiant Tee sku", "_csrf_token")
episode_store: Per-episode store containing bm25_corpus and bm25_metadata
Returns:
Top-5 matching JSON objects from episode history, annotated with step info
"""
corpus: list[str] = episode_store.get("bm25_corpus", [])
metadata: list[dict] = episode_store.get("bm25_metadata", [])
if not corpus:
return [{"note": "No episode data yet. Make API calls with curl_exec() first."}]
# Try BM25 ranking
try:
from rank_bm25 import BM25Okapi
tokenized_corpus = [_tokenize(doc) for doc in corpus]
tokenized_query = _tokenize(query)
bm25 = BM25Okapi(tokenized_corpus)
scores = bm25.get_scores(tokenized_query)
# Get top 5 by BM25 score
import numpy as np
top_k = min(5, len(scores))
top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
results = []
for idx in top_indices:
if scores[idx] > 0:
meta = metadata[idx]
result = {
"step": idx + 1,
"url": meta.get("url", ""),
"method": meta.get("method", ""),
"status_code": meta.get("status_code", 0),
"data": meta.get("response_body"),
}
results.append(result)
if results:
return results
except ImportError:
pass
except Exception as e:
print(f"[search_episode_data] BM25 error: {e}", flush=True)
# Fallback: keyword match
query_lower = query.lower()
query_terms = query_lower.split()
results = []
for idx, doc in enumerate(corpus):
if any(term in doc.lower() for term in query_terms):
meta = metadata[idx]
results.append({
"step": idx + 1,
"url": meta.get("url", ""),
"method": meta.get("method", ""),
"status_code": meta.get("status_code", 0),
"data": meta.get("response_body"),
})
return results[:5] if results else [{"note": f"No results found for: {query}"}]
def _tokenize(text: str) -> list[str]:
"""Simple whitespace + punctuation tokenizer for BM25."""
text = text.lower()
tokens = re.findall(r"[a-z0-9_\-\.]+", text)
return tokens if tokens else [""]
|