Panniantong
74c3df5c3d
BREAKING: Complete architectural rewrite.
Before: Copied x-reader's fetcher code into readers/ (1205 lines of borrowed code)
After: Pluggable channel system where each channel is a thin wrapper (~50 lines)
around the best external tool for that platform. Zero copied code.
Architecture:
- channels/base.py — Universal Channel interface (read, search, check)
- channels/web.py — Jina Reader API (swappable)
- channels/github.py — GitHub API (swappable)
- channels/twitter.py — birdx + Jina fallback (swappable)
- channels/youtube.py — yt-dlp (swappable)
- channels/reddit.py — Reddit JSON API + proxy (swappable)
- channels/rss.py — feedparser (swappable)
- channels/bilibili.py — Bilibili API (swappable)
- channels/exa_search.py — Exa semantic search (swappable)
Key design: every backend can be swapped by changing ONE file.
YouTube dies? Change youtube.py. Exa sucks? Swap exa_search.py for Tavily.
Nothing else changes.
Removed: reader.py, schema.py, readers/, search/, utils/ (all x-reader code)
Tests: 36/36 passing
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Exa semantic search — the search backbone for Agent Eyes.
|
|
|
|
Backend: Exa API (https://exa.ai) — free 1000 searches/month
|
|
Swap to: Tavily, SerpAPI, or any search API
|
|
"""
|
|
|
|
import os
|
|
import requests
|
|
from .base import Channel, SearchResult
|
|
from typing import List
|
|
|
|
|
|
class ExaSearchChannel(Channel):
|
|
name = "exa_search"
|
|
description = "Semantic web search (powers Reddit/Twitter search too)"
|
|
backends = ["Exa API"]
|
|
requires_config = ["exa_api_key"]
|
|
tier = 1
|
|
|
|
API_URL = "https://api.exa.ai/search"
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
return False # Search-only channel, doesn't read URLs
|
|
|
|
async def read(self, url: str, config=None) -> None:
|
|
raise NotImplementedError("Exa is a search engine, not a reader")
|
|
|
|
def _get_key(self, config=None) -> str:
|
|
if config:
|
|
key = config.get("exa_api_key")
|
|
if key:
|
|
return key
|
|
key = os.environ.get("EXA_API_KEY")
|
|
if key:
|
|
return key
|
|
raise ValueError(
|
|
"Exa API key not configured.\n"
|
|
"Get a free key at https://exa.ai (1000 searches/month free)\n"
|
|
"Then run: agent-eyes setup"
|
|
)
|
|
|
|
async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
|
|
api_key = self._get_key(config)
|
|
limit = kwargs.get("limit", 5)
|
|
|
|
resp = requests.post(
|
|
self.API_URL,
|
|
headers={"Content-Type": "application/json", "x-api-key": api_key},
|
|
json={
|
|
"query": query,
|
|
"numResults": min(limit, 10),
|
|
"type": "auto",
|
|
"contents": {"text": {"maxCharacters": 500}},
|
|
},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
|
|
results = []
|
|
for item in resp.json().get("results", []):
|
|
results.append(SearchResult(
|
|
title=item.get("title", ""),
|
|
url=item.get("url", ""),
|
|
snippet=item.get("text", ""),
|
|
date=item.get("publishedDate", ""),
|
|
score=item.get("score", 0),
|
|
))
|
|
return results
|