Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python | |
| """Source adapters for the Web Research Agent (#13) — different areas need different sources. | |
| Every adapter is keyless and returns the same shape [{title,url,snippet,source}] so the agent can | |
| fetch + cite uniformly. Areas map to a blend of adapters: | |
| web → general web (opportunity/competitive/client baseline) | |
| opportunity → web + fresh news | |
| competitive → web + fresh news | |
| technical → web + arXiv + GitHub | |
| client → web | |
| """ | |
| import urllib.parse | |
| import requests | |
| import feedparser | |
| from search import search as web_search, search_news | |
| _UA = "Mozilla/5.0 (compatible; GritAI-Research/1.0)" | |
| def arxiv(query: str, n: int = 4) -> list: | |
| """arXiv abstracts (keyless Atom API). Abstract pages are readable/fetchable.""" | |
| url = ("http://export.arxiv.org/api/query?search_query=all:" | |
| f"{urllib.parse.quote(query)}&start=0&max_results={int(n)}&sortBy=relevance") | |
| d = feedparser.parse(url) | |
| out = [] | |
| for e in d.entries[:n]: | |
| link = e.get("link", "") | |
| out.append({"title": e.get("title", "").replace("\n", " ").strip(), | |
| "url": link, "snippet": (e.get("summary", "") or "")[:300], "source": "arxiv"}) | |
| return [r for r in out if r["url"]] | |
| def github(query: str, n: int = 4) -> list: | |
| """GitHub repository search (keyless public API; low-volume friendly).""" | |
| try: | |
| r = requests.get("https://api.github.com/search/repositories", | |
| params={"q": query, "sort": "stars", "per_page": n}, | |
| headers={"User-Agent": _UA, "Accept": "application/vnd.github+json"}, timeout=15) | |
| items = r.json().get("items", []) | |
| return [{"title": f"{it['full_name']} (★{it.get('stargazers_count', 0)})", | |
| "url": it["html_url"], "snippet": (it.get("description") or "")[:300], "source": "github"} | |
| for it in items[:n]] | |
| except Exception: | |
| return [] | |
| AREAS = { | |
| "web": [("web", 1.0)], | |
| "opportunity": [("web", 0.6), ("news", 0.4)], | |
| "competitive": [("web", 0.6), ("news", 0.4)], | |
| "technical": [("web", 0.5), ("arxiv", 0.25), ("github", 0.25)], | |
| "client": [("web", 1.0)], | |
| } | |
| _ADAPTERS = {"web": web_search, "news": search_news, "arxiv": arxiv, "github": github} | |
| def search_area(query: str, area: str = "web", n: int = 6) -> list: | |
| """Blend results from the adapters mapped to `area`, deduped by URL.""" | |
| plan = AREAS.get(area, AREAS["web"]) | |
| seen, out = set(), [] | |
| for name, weight in plan: | |
| want = max(1, round(n * weight)) | |
| try: | |
| hits = _ADAPTERS[name](query, want + 1) | |
| except Exception: | |
| hits = [] | |
| for h in hits[:want]: | |
| key = (h["url"] or "").split("#")[0].rstrip("/") | |
| if key and key not in seen: | |
| seen.add(key) | |
| out.append(h) | |
| return out[:n] | |
| if __name__ == "__main__": | |
| import sys, json | |
| area = sys.argv[1] if len(sys.argv) > 1 else "technical" | |
| q = " ".join(sys.argv[2:]) or "retrieval augmented generation agents" | |
| print(f"area={area} query={q}") | |
| for r in search_area(q, area, 8): | |
| print(f" [{r['source']}] {r['title'][:64]} — {r['url'][:70]}") | |