| """ |
| Paper Research Agent - Learn from arXiv, Semantic Scholar, and OpenAlex. |
| """ |
| import httpx |
| from typing import Any, Dict, List, Optional |
| from xml.etree import ElementTree as ET |
|
|
| from .base import BaseAgent, AgentContext |
| from .registry import AgentRegistry |
| from src.memory.store import associative_store |
| from src.memory.models import MemoryItem |
| from src.config import settings |
|
|
|
|
| @AgentRegistry.register("paper_researcher") |
| class PaperResearchAgent(BaseAgent): |
| """Search and learn from academic papers.""" |
|
|
| ARXIV_URL = "http://export.arxiv.org/api/query" |
| SEMANTIC_SCHOLAR_URL = "https://api.semanticscholar.org/graph/v1" |
| OPENALEX_URL = "https://api.openalex.org" |
| TIMEOUT = 30.0 |
|
|
| async def execute( |
| self, |
| context: AgentContext, |
| query: str, |
| source: str = "all", |
| max_results: int = 5, |
| store_in_memory: bool = True, |
| **kwargs |
| ) -> Dict[str, Any]: |
| """Search papers and store in memory.""" |
| results: List[Dict] = [] |
|
|
| if source in ["arxiv", "all"]: |
| results.extend(await self._search_arxiv(query, max_results)) |
| if source in ["semantic", "all"]: |
| results.extend(await self._search_semantic_scholar(query, max_results)) |
| if source in ["openalex", "all"]: |
| results.extend(await self._search_openalex(query, max_results)) |
|
|
| unique_results = self._deduplicate(results) |
|
|
| if store_in_memory: |
| for paper in unique_results: |
| self._store_paper(paper) |
|
|
| return { |
| "query": query, |
| "source": source, |
| "results": unique_results, |
| "count": len(unique_results) |
| } |
|
|
| async def _search_arxiv(self, query: str, max_results: int) -> List[Dict]: |
| """Search arXiv - no API key needed.""" |
| params = { |
| "search_query": query, |
| "start": 0, |
| "max_results": max_results, |
| "sortBy": "relevance", |
| "sortOrder": "descending" |
| } |
| async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: |
| try: |
| response = await client.get(self.ARXIV_URL, params=params) |
| response.raise_for_status() |
| return self._parse_arxiv(response.text) |
| except Exception as e: |
| return [{"error": f"arXiv error: {e}", "source": "arxiv"}] |
|
|
| def _parse_arxiv(self, xml_text: str) -> List[Dict]: |
| """Parse arXiv Atom feed.""" |
| try: |
| root = ET.fromstring(xml_text) |
| except ET.ParseError: |
| return [] |
| namespace = {"atom": "http://www.w3.org/2005/Atom"} |
| papers = [] |
| for entry in root.findall("atom:entry", namespace): |
| try: |
| arxiv_id = entry.find("atom:id", namespace).text |
| if arxiv_id: |
| arxiv_id = arxiv_id.replace("http://arxiv.org/abs/", "") |
| title_elem = entry.find("atom:title", namespace) |
| title = title_elem.text if title_elem is not None else "No title" |
| summary_elem = entry.find("atom:summary", namespace) |
| summary = summary_elem.text if summary_elem is not None else "" |
| published_elem = entry.find("atom:published", namespace) |
| published = published_elem.text if published_elem is not None else "" |
| authors = [] |
| for author in entry.findall("atom:author", namespace): |
| name_elem = author.find("atom:name", namespace) |
| if name_elem is not None and name_elem.text: |
| authors.append(name_elem.text) |
| papers.append({ |
| "source": "arxiv", |
| "id": f"arxiv:{arxiv_id}", |
| "arxiv_id": arxiv_id, |
| "title": title, |
| "abstract": summary, |
| "authors": authors, |
| "published": published, |
| "url": f"https://arxiv.org/abs/{arxiv_id}", |
| "pdf_url": f"https://arxiv.org/pdf/{arxiv_id}.pdf" |
| }) |
| except Exception: |
| continue |
| return papers |
|
|
| async def _search_semantic_scholar(self, query: str, max_results: int) -> List[Dict]: |
| """Search Semantic Scholar.""" |
| headers = {} |
| if settings.semantic_scholar_api_key: |
| headers["x-api-key"] = settings.semantic_scholar_api_key |
| params = { |
| "query": query, |
| "limit": max_results, |
| "fields": "paperId,title,abstract,authors,year,venue,doi,url" |
| } |
| async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: |
| try: |
| response = await client.get( |
| f"{self.SEMANTIC_SCHOLAR_URL}/paper/search", |
| params=params, |
| headers=headers |
| ) |
| response.raise_for_status() |
| data = response.json() |
| return [self._format_semantic(p) for p in data.get("data", [])] |
| except Exception as e: |
| return [{"error": f"Semantic Scholar error: {e}", "source": "semantic"}] |
|
|
| def _format_semantic(self, paper: Dict) -> Dict: |
| """Format Semantic Scholar paper.""" |
| authors = [a.get("name") for a in paper.get("authors", []) if a and a.get("name")] |
| return { |
| "source": "semantic_scholar", |
| "id": f"semantic:{paper.get('paperId')}", |
| "semantic_id": paper.get("paperId"), |
| "title": paper.get("title", "No title"), |
| "abstract": paper.get("abstract", ""), |
| "authors": authors, |
| "year": paper.get("year"), |
| "venue": paper.get("venue"), |
| "doi": paper.get("doi"), |
| "url": paper.get("url") |
| } |
|
|
| async def _search_openalex(self, query: str, max_results: int) -> List[Dict]: |
| """Search OpenAlex - no API key needed.""" |
| params = { |
| "search": query, |
| "per-page": max_results, |
| "filter": "type:work", |
| "select": "id,title,abstract,authorships,publication_year,doi,url" |
| } |
| async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: |
| try: |
| response = await client.get(f"{self.OPENALEX_URL}/works", params=params) |
| response.raise_for_status() |
| data = response.json() |
| return [self._format_openalex(p) for p in data.get("results", [])] |
| except Exception as e: |
| return [{"error": f"OpenAlex error: {e}", "source": "openalex"}] |
|
|
| def _format_openalex(self, paper: Dict) -> Dict: |
| """Format OpenAlex paper.""" |
| authors = [] |
| for authorship in paper.get("authorships", []): |
| author = authorship.get("author", {}) |
| display_name = author.get("display_name") |
| if display_name: |
| authors.append(display_name) |
| return { |
| "source": "openalex", |
| "id": f"openalex:{paper.get('id')}", |
| "openalex_id": paper.get("id"), |
| "title": paper.get("title", "No title"), |
| "abstract": paper.get("abstract", ""), |
| "authors": authors, |
| "year": paper.get("publication_year"), |
| "doi": paper.get("doi"), |
| "url": paper.get("url") |
| } |
|
|
| def _deduplicate(self, papers: List[Dict]) -> List[Dict]: |
| """Remove duplicates by DOI or ID.""" |
| seen: set = set() |
| unique: List[Dict] = [] |
| for p in papers: |
| key = p.get("doi") or p.get("arxiv_id") or p.get("semantic_id") or p.get("openalex_id") or p.get("id") |
| if key and key not in seen: |
| seen.add(key) |
| unique.append(p) |
| return unique |
|
|
| def _store_paper(self, paper: Dict) -> None: |
| """Store paper in associative memory with tier 10 (highest authority).""" |
| try: |
| tags = ["research", "academic", paper.get("source", "paper")] |
| if paper.get("authors"): |
| tags.extend(paper["authors"][:3]) |
| if paper.get("year"): |
| tags.append(str(paper["year"])) |
|
|
| content = paper.get("abstract") or "" |
| if paper.get("url"): |
| content += "\n\nURL: " + paper.get("url") |
|
|
| item = MemoryItem( |
| title=paper.get("title", "Untitled Paper"), |
| content=content, |
| tags=tags, |
| tier=10, |
| metadata={ |
| "source": paper.get("source"), |
| "paper_id": paper.get("id"), |
| "authors": paper.get("authors"), |
| "year": paper.get("year"), |
| "url": paper.get("url"), |
| "doi": paper.get("doi") |
| } |
| ) |
| associative_store.add(item) |
| except Exception: |
| pass |
