Skills › Scientific Agent Skills › research-lookup › research_lookup.py
research_lookup.py
1253 lines · skills/research-lookup/scripts/research_lookup.py at 5f2bc03
1#!/usr/bin/env python3
2"""Parallel-first research retrieval for manuscript evidence compilation.
4The public ``ResearchLookup`` class and CLI remain backward compatible while
5ordinary queries use Parallel Search. Parallel Chat and Research are explicit,
6and Perplexity remains an optional explicit/failure fallback.
7"""
9from __future__ import annotations
11import argparse
12import json
13import os
14import re
15import shutil
16import subprocess
17import sys
18import tempfile
19import time
20from datetime import datetime, timezone
21from pathlib import Path
22from typing import Any
24from manuscript_packet import (
25 build_manuscript_packet,
26 canonicalize_url,
27 deduplicate_sources,
28 normalize_reference,
29 packet_markdown,
30 reference_score,
31 save_packet,
32)
35DEFAULT_TARGET_REFERENCES = 60
36DEFAULT_MAX_RESULTS = 20
37DEFAULT_EXTRACT_BATCH_SIZE = 10
39ACADEMIC_DOMAINS = (
40 "pubmed.ncbi.nlm.nih.gov",
41 "pmc.ncbi.nlm.nih.gov",
42 "europepmc.org",
43 "crossref.org",
44 "openalex.org",
45 "semanticscholar.org",
46 "arxiv.org",
47 "biorxiv.org",
48 "medrxiv.org",
49 "ncbi.nlm.nih.gov",
50 "nature.com",
51 "science.org",
52 "cell.com",
53 "pnas.org",
54 "nejm.org",
55 "thelancet.com",
56 "jamanetwork.com",
57 "bmj.com",
58 "springer.com",
59 "wiley.com",
60 "sciencedirect.com",
61 "ieee.org",
62 "acm.org",
63 "nih.gov",
64 "who.int",
65)
67ACADEMIC_KEYWORDS = (
68 "find papers",
69 "find paper",
70 "find articles",
71 "find article",
72 "cite ",
73 "citation",
74 "doi ",
75 "doi:",
76 "pubmed",
77 "pmid",
78 "journal article",
79 "peer-reviewed",
80 "systematic review",
81 "meta-analysis",
82 "literature search",
83 "literature review",
84 "academic papers",
85 "research papers",
86 "published studies",
87 "scholarly",
88 "arxiv",
89 "preprint",
90 "foundational papers",
91 "seminal papers",
92 "landmark papers",
93 "highly cited",
94 "manuscript",
95 "evidence base",
96)
98ACADEMIC_FACETS = (
99 (
100 "recent-primary",
101 "Find recent peer-reviewed primary studies directly addressing the topic. "
102 "Prioritize complete bibliographic metadata, DOI or PMID, study design, "
103 "sample size, methods, outcomes, quantitative findings, and limitations.",
104 ("primary study", "peer reviewed", "recent evidence"),
105 ),
106 (
107 "reviews",
108 "Find systematic reviews, meta-analyses, evidence syntheses, and major "
109 "consensus statements directly addressing the topic.",
110 ("systematic review", "meta-analysis", "evidence synthesis"),
111 ),
112 (
113 "seminal",
114 "Find seminal, foundational, landmark, and field-defining publications on "
115 "the topic. Prefer sources with stable identifiers and authoritative records.",
116 ("seminal paper", "foundational study", "landmark research"),
117 ),
118 (
119 "methods",
120 "Find methods, protocols, measurement validation, benchmark, and mechanistic "
121 "studies that can justify manuscript methods and interpretation.",
122 ("methods", "protocol", "validation", "mechanism"),
123 ),
124 (
125 "contradictory",
126 "Find conflicting, contradictory, null, negative, replication, and limitation "
127 "evidence on the topic. Do not assume the dominant conclusion is correct.",
128 ("conflicting evidence", "null results", "limitations", "replication"),
129 ),
130)
133class ResearchLookup:
134 """Research lookup with Parallel Search as the stable default backend."""
136 PARALLEL_SYSTEM_PROMPT = (
137 "Compile a rigorous, citation-rich research report for manuscript preparation. "
138 "Prioritize primary and peer-reviewed literature, distinguish preprints, include "
139 "quantitative evidence and limitations, surface contradictory findings, and do "
140 "not invent citations."
141 )
143 def __init__(
144 self,
145 force_backend: str | None = None,
146 *,
147 academic: bool | None = None,
148 target_references: int = DEFAULT_TARGET_REFERENCES,
149 search_mode: str = "basic",
150 max_results: int = DEFAULT_MAX_RESULTS,
151 include_domains: list[str] | None = None,
152 after_date: str | None = None,
153 extract_limit: int | None = None,
154 extract_batch_size: int = DEFAULT_EXTRACT_BATCH_SIZE,
155 processor: str = "pro-fast",
156 chat_model: str = "core",
157 previous_interaction_id: str | None = None,
158 allow_perplexity_fallback: bool = False,
159 manuscript_context: dict[str, Any] | None = None,
160 cli_timeout: int = 300,
161 research_timeout: int = 3600,
162 ):
163 """Initialize routing and retrieval options.
165 ``parallel`` remains a compatibility alias for the explicit ``research``
166 backend. A bare query always selects ``search``.
167 """
168 backend_aliases = {"parallel": "research"}
169 normalized_backend = backend_aliases.get(force_backend or "", force_backend)
170 if normalized_backend not in {
171 None,
172 "search",
173 "research",
174 "chat",
175 "perplexity",
176 }:
177 raise ValueError(
178 "force_backend must be one of: search, research, parallel, chat, "
179 "perplexity"
180 )
181 if chat_model not in {"speed", "lite", "base", "core"}:
182 raise ValueError("chat_model must be one of: speed, lite, base, core")
183 if target_references < 1:
184 raise ValueError("target_references must be at least 1")
185 if max_results < 1:
186 raise ValueError("max_results must be at least 1")
187 if not 1 <= extract_batch_size <= 20:
188 raise ValueError("extract_batch_size must be between 1 and 20")
189 if search_mode not in {"turbo", "fast", "basic", "advanced"}:
190 raise ValueError("search_mode must be one of: turbo, fast, basic, advanced")
192 self.force_backend = normalized_backend
193 self.requested_backend = force_backend
194 self.academic = academic
195 self.target_references = target_references
196 self.search_mode = search_mode
197 self.max_results = max_results
198 self.include_domains = include_domains or []
199 self.after_date = after_date
200 self.extract_limit = (
201 target_references if extract_limit is None else max(0, extract_limit)
202 )
203 self.extract_batch_size = extract_batch_size
204 self.processor = processor
205 self.chat_model = chat_model
206 self.previous_interaction_id = previous_interaction_id
207 self.allow_perplexity_fallback = allow_perplexity_fallback
208 self.manuscript_context = manuscript_context or {}
209 self.cli_timeout = cli_timeout
210 self.research_timeout = research_timeout
212 self.parallel_available = shutil.which("parallel-cli") is not None
213 self.chat_available = bool(os.getenv("PARALLEL_API_KEY"))
214 self.perplexity_available = bool(os.getenv("OPENROUTER_API_KEY"))
216 if self.force_backend in {"search", "research"} and not self.parallel_available:
217 raise ValueError(
218 "parallel-cli is required for the selected Parallel backend. "
219 "Install the pinned CLI version documented in SKILL.md."
220 )
221 if self.force_backend == "chat" and not self.chat_available:
222 raise ValueError(
223 "PARALLEL_API_KEY is required when forcing the Parallel Chat backend."
224 )
225 if self.force_backend == "perplexity" and not self.perplexity_available:
226 raise ValueError(
227 "OPENROUTER_API_KEY is required when forcing the Perplexity backend."
228 )
229 if (
230 not self.parallel_available
231 and not self.chat_available
232 and not self.perplexity_available
233 ):
234 raise ValueError(
235 "No backend is available. Install/authenticate parallel-cli or set "
236 "PARALLEL_API_KEY for explicit Chat use, or OPENROUTER_API_KEY for "
237 "the optional Perplexity backend."
238 )
240 def _select_backend(self, query: str) -> str:
241 """Select Search by default; provider changes require explicit intent."""
242 del query
243 if self.force_backend:
244 return self.force_backend
245 # A stored OpenRouter key is not consent to change the destination.
246 # Missing CLI errors follow the same opt-in fallback path as API failures.
247 return "search"
249 def _is_academic_query(self, query: str) -> bool:
250 if self.academic is not None:
251 return self.academic
252 lowered = query.lower()
253 return any(keyword in lowered for keyword in ACADEMIC_KEYWORDS)
255 def _query_with_context(self, query: str) -> str:
256 if not self.manuscript_context:
257 return query
258 context_lines = [
259 f"{key.replace('_', ' ').title()}: {value}"
260 for key, value in self.manuscript_context.items()
261 if value not in (None, "", [], {})
262 ]
263 if not context_lines:
264 return query
265 return f"{query}\n\nManuscript context:\n" + "\n".join(context_lines)
267 def _run_parallel_cli(
268 self, args: list[str], *, timeout: int | None = None
269 ) -> dict[str, Any]:
270 """Run the pinned CLI without shell interpolation and parse JSON."""
271 command = ["parallel-cli", *args]
272 try:
273 completed = subprocess.run(
274 command,
275 check=False,
276 capture_output=True,
277 text=True,
278 timeout=timeout or self.cli_timeout,
279 )
280 except FileNotFoundError as exc:
281 raise RuntimeError("parallel-cli was not found on PATH.") from exc
282 except subprocess.TimeoutExpired as exc:
283 raise RuntimeError(
284 f"parallel-cli timed out after {timeout or self.cli_timeout} seconds."
285 ) from exc
287 if completed.returncode != 0:
288 detail = completed.stderr.strip() or completed.stdout.strip()
289 raise RuntimeError(
290 f"parallel-cli exited with status {completed.returncode}: {detail}"
291 )
292 try:
293 payload = json.loads(completed.stdout)
294 except json.JSONDecodeError as exc:
295 raise RuntimeError(
296 "parallel-cli returned non-JSON output despite --json."
297 ) from exc
298 if not isinstance(payload, dict):
299 raise RuntimeError("parallel-cli returned a non-object JSON response.")
300 if payload.get("type") == "error" or payload.get("error"):
301 raise RuntimeError(f"parallel-cli returned an error: {payload.get('error')}")
302 return payload
304 @staticmethod
305 def _usage_counts(payloads: list[dict[str, Any]]) -> list[dict[str, Any]]:
306 counts: dict[str, int] = {}
307 for payload in payloads:
308 for item in payload.get("usage") or []:
309 name = str(item.get("name") or "unknown")
310 counts[name] = counts.get(name, 0) + int(item.get("count") or 0)
311 return [{"name": name, "count": count} for name, count in sorted(counts.items())]
313 def _search_once(
314 self,
315 *,
316 objective: str,
317 keyword_queries: tuple[str, ...],
318 facet: str,
319 academic_domains: bool,
320 mode: str,
321 session_id: str | None = None,
322 apply_after_date: bool = True,
323 ) -> tuple[dict[str, Any], dict[str, Any], list[dict[str, Any]]]:
324 args = [
325 "search",
326 objective,
327 "--mode",
328 mode,
329 "--max-results",
330 str(self.max_results),
331 "--excerpt-max-chars-total",
332 str(max(27000, self.max_results * 2500)),
333 "--json",
334 ]
335 for query in keyword_queries:
336 args.extend(["-q", query])
337 domains = list(self.include_domains)
338 if academic_domains:
339 domains.extend(ACADEMIC_DOMAINS)
340 domains = list(dict.fromkeys(domains))
341 if domains:
342 args.extend(["--include-domains", ",".join(domains)])
343 if self.after_date and apply_after_date:
344 args.extend(["--after-date", self.after_date])
345 if session_id:
346 args.extend(["--session-id", session_id])
348 started = datetime.now(timezone.utc).isoformat()
349 payload = self._run_parallel_cli(args)
350 raw_results = payload.get("results") or []
351 results: list[dict[str, Any]] = []
352 for raw_result in raw_results:
353 source = dict(raw_result)
354 source["facets"] = sorted(
355 set(source.get("facets") or []) | {facet}
356 )
357 source["url"] = canonicalize_url(str(source.get("url") or ""))
358 results.append(source)
359 ledger = {
360 "capability": "search",
361 "facet": facet,
362 "objective": objective,
363 "keyword_queries": list(keyword_queries),
364 "mode": mode,
365 "domains": domains,
366 "after_date": self.after_date if apply_after_date else None,
367 "timestamp": started,
368 "result_count": len(results),
369 "search_id": payload.get("search_id"),
370 "session_id": payload.get("session_id"),
371 # V1 Search has no top-level status field.
372 "status": "ok",
373 "warnings": payload.get("warnings") or [],
374 }
375 return payload, ledger, results
377 def _rank_sources_for_extraction(
378 self, sources: list[dict[str, Any]]
379 ) -> list[dict[str, Any]]:
380 ranked: list[tuple[tuple[int, int, int, int], dict[str, Any]]] = []
381 for index, source in enumerate(sources, start=1):
382 reference = normalize_reference(source, index)
383 ranked.append((reference_score(reference), source))
384 ranked.sort(key=lambda item: item[0], reverse=True)
385 return [source for _, source in ranked]
387 def _extract_sources(
388 self,
389 sources: list[dict[str, Any]],
390 search_ledger: list[dict[str, Any]],
391 session_id: str | None = None,
392 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
393 if self.extract_limit == 0:
394 return sources, []
395 candidates = [
396 source
397 for source in self._rank_sources_for_extraction(sources)
398 if source.get("url")
399 ][: self.extract_limit]
400 extraction_payloads: list[dict[str, Any]] = []
401 extracted_sources: list[dict[str, Any]] = []
402 objective = (
403 "Extract only source-supported bibliographic and study evidence: authors, "
404 "year, journal or venue, DOI/PMID, publication type, study design, population "
405 "or system, sample size, methods, intervention/exposure, comparator, outcomes, "
406 "quantitative findings, uncertainty, limitations, correction/retraction status, "
407 "and conclusions. Preserve exact wording for evidence excerpts."
408 )
409 for offset in range(0, len(candidates), self.extract_batch_size):
410 batch = candidates[offset : offset + self.extract_batch_size]
411 urls = [str(source["url"]) for source in batch]
412 args = [
413 "extract",
414 *urls,
415 "--objective",
416 objective,
417 "--excerpt-max-chars-per-result",
418 "6000",
419 "--excerpt-max-chars-total",
420 str(max(12000, len(urls) * 6000)),
421 "--json",
422 ]
423 if session_id:
424 args.extend(["--session-id", session_id])
425 timestamp = datetime.now(timezone.utc).isoformat()
426 try:
427 payload = self._run_parallel_cli(args)
428 except RuntimeError as exc:
429 search_ledger.append(
430 {
431 "capability": "extract",
432 "timestamp": timestamp,
433 "urls": urls,
434 "status": "error",
435 "error": str(exc),
436 }
437 )
438 continue
439 extraction_payloads.append(payload)
440 session_id = session_id or payload.get("session_id")
441 batch_results = payload.get("results") or []
442 retrieved_urls: set[str] = set()
443 for raw_result in batch_results:
444 source = dict(raw_result)
445 source["url"] = canonicalize_url(str(source.get("url") or ""))
446 excerpts = [
447 excerpt
448 for excerpt in source.get("excerpts") or []
449 if isinstance(excerpt, str) and excerpt.strip()
450 ]
451 source["extraction_excerpts"] = excerpts
452 source["extracted"] = bool(source["url"] and excerpts)
453 if source["extracted"]:
454 retrieved_urls.add(source["url"])
455 source["facets"] = ["extracted-evidence"]
456 extracted_sources.append(source)
457 unresolved_urls = [
458 url for url in urls if canonicalize_url(url) not in retrieved_urls
459 ]
460 search_ledger.append(
461 {
462 "capability": "extract",
463 "timestamp": timestamp,
464 "urls": urls,
465 "result_count": len(batch_results),
466 "extract_id": payload.get("extract_id"),
467 "session_id": payload.get("session_id"),
468 "status": (
469 "partial" if unresolved_urls or payload.get("errors") else "ok"
470 ),
471 "errors": payload.get("errors") or [],
472 "unresolved_urls": unresolved_urls,
473 "warnings": payload.get("warnings") or [],
474 }
475 )
476 return deduplicate_sources([*sources, *extracted_sources]), extraction_payloads
478 def _parallel_search(self, query: str) -> dict[str, Any]:
479 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
480 academic = self._is_academic_query(query)
481 contextual_query = self._query_with_context(query)
482 search_payloads: list[dict[str, Any]] = []
483 ledger: list[dict[str, Any]] = []
484 sources: list[dict[str, Any]] = []
485 session_id: str | None = None
486 errors: list[str] = []
488 if academic:
489 mode = "advanced"
490 for facet, instruction, keywords in ACADEMIC_FACETS:
491 objective = f"{instruction}\n\nTopic: {contextual_query}"
492 keyword_queries = tuple(f"{query} {keyword}" for keyword in keywords)
493 try:
494 payload, entry, facet_sources = self._search_once(
495 objective=objective,
496 keyword_queries=keyword_queries,
497 facet=facet,
498 academic_domains=True,
499 mode=mode,
500 session_id=session_id,
501 apply_after_date=facet != "seminal",
502 )
503 except RuntimeError as exc:
504 errors.append(f"{facet}: {exc}")
505 ledger.append(
506 {
507 "capability": "search",
508 "facet": facet,
509 "objective": objective,
510 "timestamp": datetime.now(timezone.utc).isoformat(),
511 "status": "error",
512 "error": str(exc),
513 }
514 )
515 continue
516 search_payloads.append(payload)
517 ledger.append(entry)
518 sources.extend(facet_sources)
519 session_id = session_id or payload.get("session_id")
521 if len(deduplicate_sources(sources)) < self.target_references:
522 objective = (
523 "Find additional authoritative sources that directly address this "
524 "manuscript topic, including important evidence missed by scholarly "
525 "domain filters. Prefer sources with stable links and bibliographic "
526 f"metadata.\n\nTopic: {contextual_query}"
527 )
528 try:
529 payload, entry, general_sources = self._search_once(
530 objective=objective,
531 keyword_queries=(query,),
532 facet="general-companion",
533 academic_domains=False,
534 mode=mode,
535 session_id=session_id,
536 )
537 search_payloads.append(payload)
538 ledger.append(entry)
539 sources.extend(general_sources)
540 except RuntimeError as exc:
541 errors.append(f"general-companion: {exc}")
542 ledger.append(
543 {
544 "capability": "search",
545 "facet": "general-companion",
546 "objective": objective,
547 "timestamp": datetime.now(timezone.utc).isoformat(),
548 "status": "error",
549 "error": str(exc),
550 }
551 )
552 else:
553 objective = (
554 "Find current, authoritative, directly relevant information for this "
555 f"research question:\n\n{contextual_query}"
556 )
557 payload, entry, sources = self._search_once(
558 objective=objective,
559 keyword_queries=(query,),
560 facet="general",
561 academic_domains=False,
562 mode=self.search_mode,
563 )
564 search_payloads.append(payload)
565 ledger.append(entry)
567 sources = deduplicate_sources(sources)
568 if not sources:
569 detail = "; ".join(errors) if errors else "No results returned."
570 raise RuntimeError(f"Parallel Search produced no usable sources. {detail}")
572 extraction_payloads: list[dict[str, Any]] = []
573 if academic:
574 sources, extraction_payloads = self._extract_sources(sources, ledger, session_id)
576 packet = build_manuscript_packet(
577 query=query,
578 sources=sources,
579 search_ledger=ledger,
580 target_references=self.target_references if academic else len(sources),
581 manuscript_context=self.manuscript_context,
582 )
583 packet["raw_service_responses"] = {
584 "search": search_payloads,
585 "extract": extraction_payloads,
586 }
587 if errors:
588 packet["warnings"].append(
589 "Some bounded search passes failed: " + "; ".join(errors)
590 )
591 if any(
592 entry.get("status") in {"partial", "error"}
593 for entry in ledger
594 if entry.get("capability") == "extract"
595 ):
596 packet["warnings"].append(
597 "Some Extract URLs were unavailable or returned no excerpts; inspect "
598 "errors and unresolved_urls in the search ledger. Redirected URLs may "
599 "also require manual reconciliation."
600 )
601 response = packet_markdown(packet)
602 references = packet["references"]
603 citations = [
604 {
605 "type": "source",
606 "title": reference["title"],
607 "url": reference["url"],
608 "date": reference["year"],
609 "doi": reference["doi"],
610 "pmid": reference["pmid"],
611 }
612 for reference in references
613 if reference.get("url")
614 ]
615 normalized_sources = [
616 {
617 "title": reference["title"],
618 "url": reference["url"],
619 "publish_date": reference["year"],
620 "excerpts": reference["supporting_excerpts"],
621 }
622 for reference in references
623 ]
624 usage = self._usage_counts([*search_payloads, *extraction_payloads])
625 return {
626 "success": True,
627 "query": query,
628 "response": response,
629 "citations": citations,
630 "sources": normalized_sources,
631 "timestamp": timestamp,
632 "backend": "search",
633 "model": f"parallel-search/{'advanced' if academic else self.search_mode}",
634 "usage": usage,
635 "academic": academic,
636 "references": references,
637 "search_ledger": ledger,
638 "packet": packet,
639 }
641 @staticmethod
642 def _find_report_text(payload: Any) -> str:
643 """Read the CLI text-output contract without mistaking basis for a report."""
644 output = payload.get("output", {}) if isinstance(payload, dict) else {}
645 content = output.get("content") if isinstance(output, dict) else None
646 return content if isinstance(content, str) else ""
648 @staticmethod
649 def _sources_from_payload(payload: Any) -> list[dict[str, str]]:
650 sources: list[dict[str, str]] = []
652 def walk(value: Any) -> None:
653 if isinstance(value, dict):
654 url = value.get("url")
655 if isinstance(url, str) and url.startswith(("http://", "https://")):
656 sources.append(
657 {
658 "type": "source",
659 "url": canonicalize_url(url),
660 "title": str(value.get("title") or ""),
661 }
662 )
663 for nested in value.values():
664 walk(nested)
665 elif isinstance(value, list):
666 for nested in value:
667 walk(nested)
669 walk(payload)
670 unique: dict[str, dict[str, str]] = {}
671 for source in sources:
672 unique.setdefault(source["url"], source)
673 return list(unique.values())
675 def _parallel_research(self, query: str) -> dict[str, Any]:
676 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
677 research_query = (
678 f"{self.PARALLEL_SYSTEM_PROMPT}\n\nResearch topic:\n"
679 f"{self._query_with_context(query)}"
680 )
681 if len(research_query) > 15000:
682 raise ValueError(
683 "Research query plus context and instructions exceeds the CLI's "
684 "15,000-character limit; shorten it to avoid silent truncation."
685 )
686 with tempfile.TemporaryDirectory(prefix="research-lookup-") as temp_dir:
687 output_base = Path(temp_dir) / "report"
688 args = [
689 "research",
690 "run",
691 research_query,
692 "--processor",
693 self.processor,
694 "--text",
695 "--text-description",
696 (
697 "Produce a manuscript-research report with complete inline citations, "
698 "quantitative evidence, methods, limitations, conflicts, and gaps."
699 ),
700 "--timeout",
701 str(self.research_timeout),
702 "--json",
703 "-o",
704 str(output_base),
705 ]
706 if self.previous_interaction_id:
707 args.extend(
708 ["--previous-interaction-id", self.previous_interaction_id]
709 )
710 payload = self._run_parallel_cli(args, timeout=self.research_timeout + 60)
711 if payload.get("status") != "completed":
712 raise RuntimeError(
713 f"Parallel Research is not completed: {payload.get('status')} "
714 f"(run_id={payload.get('run_id')})."
715 )
716 markdown_path = output_base.with_suffix(".md")
717 content = (
718 markdown_path.read_text(encoding="utf-8")
719 if markdown_path.exists()
720 else self._find_report_text(payload)
721 )
722 if not content.strip():
723 raise RuntimeError("Parallel Research returned no text report.")
724 sources = self._sources_from_payload(
725 (payload.get("output") or {}).get("basis") or []
726 )
727 text_citations = self._extract_citations_from_text(content)
728 return {
729 "success": True,
730 "query": query,
731 "response": content,
732 "citations": [*sources, *text_citations],
733 "sources": sources,
734 "timestamp": timestamp,
735 "backend": "research",
736 "model": f"parallel-research/{self.processor}",
737 "usage": payload.get("usage") or [],
738 "run_id": payload.get("run_id") or payload.get("id"),
739 "interaction_id": payload.get("interaction_id"),
740 "raw_response": payload,
741 }
743 def _parallel_chat(self, query: str) -> dict[str, Any]:
744 """Run the explicit OpenAI-compatible Parallel Chat backend."""
745 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
746 api_key = os.getenv("PARALLEL_API_KEY")
747 if not api_key:
748 raise RuntimeError(
749 "PARALLEL_API_KEY is required for the explicit Chat backend."
750 )
751 try:
752 import requests
753 except ImportError as exc:
754 raise ImportError(
755 "The optional Parallel Chat backend requires requests."
756 ) from exc
758 payload = {
759 "model": self.chat_model,
760 "messages": [
761 {"role": "system", "content": self.PARALLEL_SYSTEM_PROMPT},
762 {"role": "user", "content": self._query_with_context(query)},
763 ],
764 "stream": False,
765 }
766 response = requests.post(
767 "https://api.parallel.ai/v1beta/chat/completions",
768 headers={
769 "x-api-key": api_key,
770 "Content-Type": "application/json",
771 },
772 json=payload,
773 timeout=self.research_timeout,
774 )
775 response.raise_for_status()
776 response_payload = response.json()
777 choices = response_payload.get("choices") or []
778 if not choices:
779 raise RuntimeError("No response choices received from Parallel Chat.")
780 content = str(choices[0].get("message", {}).get("content") or "")
781 basis_sources = self._sources_from_payload(
782 response_payload.get("basis") or []
783 )
784 text_citations = self._extract_citations_from_text(content)
785 return {
786 "success": True,
787 "query": query,
788 "response": content,
789 "citations": [*basis_sources, *text_citations],
790 "sources": basis_sources,
791 "timestamp": timestamp,
792 "backend": "chat",
793 "model": f"parallel-chat/{self.chat_model}",
794 "usage": response_payload.get("usage") or {},
795 "basis": response_payload.get("basis"),
796 "raw_response": response_payload,
797 }
799 def _perplexity_lookup(self, query: str) -> dict[str, Any]:
800 """Run the preserved optional academic backend through OpenRouter."""
801 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
802 api_key = os.getenv("OPENROUTER_API_KEY")
803 if not api_key:
804 raise RuntimeError("OPENROUTER_API_KEY is not set.")
805 try:
806 import requests
807 except ImportError as exc:
808 raise ImportError(
809 "The optional Perplexity backend requires requests."
810 ) from exc
812 model = "perplexity/sonar-pro-search"
813 current_year = datetime.now().year
814 prompt = (
815 "Find high-quality academic evidence for manuscript preparation. Return "
816 "complete citations with DOI/PMID where available, quantitative findings, "
817 "methods, limitations, contradictory evidence, and research gaps. Prioritize "
818 f"peer-reviewed literature and clearly label preprints. Current year: "
819 f"{current_year}.\n\nQuery: {self._query_with_context(query)}"
820 )
821 data = {
822 "model": model,
823 "messages": [
824 {
825 "role": "system",
826 "content": (
827 "You are an academic research assistant. Never invent references "
828 "or bibliographic metadata."
829 ),
830 },
831 {"role": "user", "content": prompt},
832 ],
833 "max_tokens": 8000,
834 "temperature": 0.1,
835 "web_search_options": {"search_context_size": "high"},
836 }
837 response = requests.post(
838 "https://openrouter.ai/api/v1/chat/completions",
839 headers={
840 "Authorization": f"Bearer {api_key}",
841 "Content-Type": "application/json",
842 },
843 json=data,
844 timeout=90,
845 )
846 response.raise_for_status()
847 payload = response.json()
848 choices = payload.get("choices") or []
849 if not choices:
850 raise RuntimeError("No response choices received from Perplexity.")
851 content = str(choices[0].get("message", {}).get("content") or "")
852 api_citations = self._extract_api_citations(payload, choices[0])
853 text_citations = self._extract_citations_from_text(content)
854 return {
855 "success": True,
856 "query": query,
857 "response": content,
858 "citations": [*api_citations, *text_citations],
859 "sources": api_citations,
860 "timestamp": timestamp,
861 "backend": "perplexity",
862 "model": model,
863 "usage": payload.get("usage") or {},
864 "raw_response": payload,
865 }
867 @staticmethod
868 def _extract_api_citations(
869 response: dict[str, Any], choice: dict[str, Any]
870 ) -> list[dict[str, str]]:
871 citations: list[dict[str, str]] = []
872 for annotation in choice.get("message", {}).get("annotations") or []:
873 if annotation.get("type") != "url_citation":
874 continue
875 item = annotation.get("url_citation") or {}
876 citations.append(
877 {
878 "type": "source",
879 "url": canonicalize_url(str(item.get("url") or "")),
880 "title": str(item.get("title") or ""),
881 "snippet": str(item.get("content") or ""),
882 "date": "",
883 }
884 )
885 search_results = (
886 response.get("search_results")
887 or choice.get("search_results")
888 or choice.get("message", {}).get("search_results")
889 or []
890 )
891 for result in search_results:
892 citations.append(
893 {
894 "type": "source",
895 "title": str(result.get("title") or ""),
896 "url": canonicalize_url(str(result.get("url") or "")),
897 "date": str(result.get("date") or ""),
898 "snippet": str(result.get("snippet") or ""),
899 }
900 )
901 legacy = (
902 response.get("citations")
903 or choice.get("citations")
904 or choice.get("message", {}).get("citations")
905 or []
906 )
907 for citation in legacy:
908 if isinstance(citation, str):
909 citations.append(
910 {
911 "type": "source",
912 "url": canonicalize_url(citation),
913 "title": "",
914 "date": "",
915 }
916 )
917 elif isinstance(citation, dict):
918 citations.append(
919 {
920 "type": "source",
921 "url": canonicalize_url(str(citation.get("url") or "")),
922 "title": str(citation.get("title") or ""),
923 "date": str(citation.get("date") or ""),
924 }
925 )
926 unique: dict[str, dict[str, str]] = {}
927 for citation in citations:
928 if citation["url"]:
929 unique.setdefault(citation["url"], citation)
930 return list(unique.values())
932 @staticmethod
933 def _extract_citations_from_text(text: str) -> list[dict[str, str]]:
934 citations: list[dict[str, str]] = []
935 seen: set[str] = set()
936 doi_pattern = re.compile(
937 r"(?:doi[:\s]*|https?://(?:dx\.)?doi\.org/)"
938 r"(10\.\d{4,9}/[-._;()/:A-Z0-9]+)",
939 re.IGNORECASE,
940 )
941 for doi in doi_pattern.findall(text or ""):
942 clean_doi = doi.rstrip(".,;:)]}").lower()
943 url = f"https://doi.org/{clean_doi}"
944 if url not in seen:
945 seen.add(url)
946 citations.append(
947 {"type": "doi", "doi": clean_doi, "url": url, "title": ""}
948 )
949 url_pattern = re.compile(r"https?://[^\s)\]>,\"']+", re.IGNORECASE)
950 for url in url_pattern.findall(text or ""):
951 clean_url = canonicalize_url(url.rstrip(".,;:"))
952 if clean_url and clean_url not in seen:
953 seen.add(clean_url)
954 citations.append(
955 {"type": "url", "url": clean_url, "title": ""}
956 )
957 return citations
959 @staticmethod
960 def _failure_result(query: str, backend: str, exc: Exception) -> dict[str, Any]:
961 return {
962 "success": False,
963 "query": query,
964 "error": str(exc),
965 "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
966 "backend": backend,
967 "model": "unknown",
968 }
970 def lookup(self, query: str) -> dict[str, Any]:
971 """Perform one lookup while isolating errors in the result envelope."""
972 backend = self._select_backend(query)
973 print(
974 f"[Research] Backend: {backend} | Query: {query[:80]}...",
975 file=sys.stderr,
976 )
977 try:
978 if backend == "search":
979 return self._parallel_search(query)
980 if backend == "research":
981 return self._parallel_research(query)
982 if backend == "chat":
983 return self._parallel_chat(query)
984 return self._perplexity_lookup(query)
985 except Exception as exc:
986 if (
987 backend in {"search", "research", "chat"}
988 and self.allow_perplexity_fallback
989 and self.perplexity_available
990 ):
991 print(
992 f"[Research] Parallel failed; explicit Perplexity fallback: {exc}",
993 file=sys.stderr,
994 )
995 try:
996 result = self._perplexity_lookup(query)
997 result["fallback_from"] = backend
998 result["fallback_reason"] = str(exc)
999 return result
1000 except Exception as fallback_exc:
1001 return self._failure_result(
1002 query,
1003 "perplexity",
1004 RuntimeError(
1005 f"Parallel failed: {exc}; fallback failed: {fallback_exc}"
1006 ),
1007 )
1008 return self._failure_result(query, backend, exc)
1010 def batch_lookup(
1011 self, queries: list[str], delay: float = 1.0
1012 ) -> list[dict[str, Any]]:
1013 """Perform multiple lookups with preserved per-query error isolation."""
1014 results: list[dict[str, Any]] = []
1015 for index, query in enumerate(queries):
1016 if index and delay > 0:
1017 time.sleep(delay)
1018 result = self.lookup(query)
1019 results.append(result)
1020 print(
1021 f"[Research] Completed query {index + 1}/{len(queries)}: "
1022 f"{query[:50]}...",
1023 file=sys.stderr,
1024 )
1025 return results
1028def _load_context(path: str | None) -> dict[str, Any]:
1029 if not path:
1030 return {}
1031 payload = json.loads(Path(path).read_text(encoding="utf-8"))
1032 if not isinstance(payload, dict):
1033 raise ValueError("The manuscript context file must contain a JSON object.")
1034 return payload
1037def _split_domains(values: list[str] | None) -> list[str]:
1038 domains: list[str] = []
1039 for value in values or []:
1040 domains.extend(part.strip() for part in value.split(",") if part.strip())
1041 return list(dict.fromkeys(domains))
1044def _slug(text: str) -> str:
1045 return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")[:60] or "research"
1048def _render_human_result(result: dict[str, Any], index: int) -> str:
1049 if not result.get("success"):
1050 return f"\nError in query {index}: {result.get('error', 'Unknown error')}"
1051 lines = [
1052 "",
1053 "=" * 80,
1054 f"Query {index}: {result['query']}",
1055 f"Timestamp: {result['timestamp']}",
1056 f"Backend: {result.get('backend', 'unknown')} | "
1057 f"Model: {result.get('model', 'unknown')}",
1058 "=" * 80,
1059 str(result.get("response") or ""),
1060 ]
1061 sources = result.get("sources") or []
1062 if sources and result.get("backend") != "search":
1063 lines.append(f"\nSources ({len(sources)}):")
1064 for source_index, source in enumerate(sources, start=1):
1065 title = source.get("title") or "Untitled"
1066 url = source.get("url") or ""
1067 lines.append(f" [{source_index}] {title}")
1068 if url:
1069 lines.append(f" {url}")
1070 if result.get("usage"):
1071 lines.append(f"\nUsage: {result['usage']}")
1072 if result.get("artifacts"):
1073 lines.append(f"\nArtifacts: {result['artifacts']}")
1074 return "\n".join(lines)
1077def build_parser() -> argparse.ArgumentParser:
1078 parser = argparse.ArgumentParser(
1079 description=(
1080 "Parallel-first research lookup and manuscript evidence compiler"
1081 ),
1082 formatter_class=argparse.RawDescriptionHelpFormatter,
1083 epilog="""
1084Examples:
1085 # Academic manuscript packet (60 verified references by default)
1086 python research_lookup.py "CRISPR off-target effects" --academic \
1087 --packet-dir sources/crispr
1089 # Fast bounded web lookup
1090 python research_lookup.py "latest NIST AI guidance" --no-academic
1092 # Explicit deep research (legacy 'parallel' alias remains accepted)
1093 python research_lookup.py "comprehensive quantum error correction review" \
1094 --force-backend research --processor pro
1096 # Explicit OpenAI-compatible Parallel Chat (never selected by default)
1097 python research_lookup.py "synthesize the evidence" \
1098 --force-backend chat --chat-model core
1100 # Optional explicit Perplexity fallback
1101 python research_lookup.py "find papers on topic" --force-backend perplexity
1102 """,
1103 )
1104 parser.add_argument("query", nargs="?", help="Research query to look up")
1105 parser.add_argument("--batch", nargs="+", help="Run multiple queries")
1106 parser.add_argument(
1107 "--force-backend",
1108 choices=["search", "research", "parallel", "chat", "perplexity"],
1109 help=(
1110 "'parallel' is retained as a compatibility alias for 'research'; "
1111 "'chat' is explicit and never selected by default"
1112 ),
1113 )
1114 parser.add_argument(
1115 "--academic",
1116 action=argparse.BooleanOptionalAction,
1117 default=None,
1118 help="Force or disable the multi-pass academic retrieval strategy",
1119 )
1120 parser.add_argument(
1121 "--target-references",
1122 type=int,
1123 default=DEFAULT_TARGET_REFERENCES,
1124 help="Target verified references for academic retrieval (default: 60)",
1125 )
1126 parser.add_argument(
1127 "--search-mode",
1128 choices=["turbo", "fast", "basic", "advanced"],
1129 default="basic",
1130 help="Mode for non-academic Search calls (academic calls use advanced)",
1131 )
1132 parser.add_argument(
1133 "--max-results",
1134 type=int,
1135 default=DEFAULT_MAX_RESULTS,
1136 help="Maximum results requested per bounded Search pass",
1137 )
1138 parser.add_argument(
1139 "--include-domains",
1140 action="append",
1141 help="Additional comma-separated domains; may be repeated",
1142 )
1143 parser.add_argument("--after-date", help="Search results after YYYY-MM-DD")
1144 parser.add_argument(
1145 "--extract-limit",
1146 type=int,
1147 help="Maximum academic sources to verify with Extract (default: target)",
1148 )
1149 parser.add_argument(
1150 "--no-extract",
1151 action="store_true",
1152 help="Skip Extract verification (reference shortfalls will be reported)",
1153 )
1154 parser.add_argument(
1155 "--processor",
1156 default="pro-fast",
1157 help="Parallel Research processor for explicit deep research",
1158 )
1159 parser.add_argument(
1160 "--chat-model",
1161 choices=["speed", "lite", "base", "core"],
1162 default="core",
1163 help="Parallel Chat model for the explicit Chat backend (default: core)",
1164 )
1165 parser.add_argument(
1166 "--previous-interaction-id",
1167 help="Continue a related Parallel Research interaction",
1168 )
1169 parser.add_argument(
1170 "--fallback-perplexity",
1171 action="store_true",
1172 help="Allow Perplexity only if a Parallel call fails",
1173 )
1174 parser.add_argument(
1175 "--context-file",
1176 help="JSON object with manuscript question, study type, PICO, field, and journal",
1177 )
1178 parser.add_argument(
1179 "--packet-dir",
1180 help="Write manuscript packet artifacts to this directory",
1181 )
1182 parser.add_argument(
1183 "--batch-delay",
1184 type=float,
1185 default=1.0,
1186 help="Delay between batch queries in seconds",
1187 )
1188 parser.add_argument("-o", "--output", help="Write primary output to a file")
1189 parser.add_argument("--json", action="store_true", help="Output result JSON")
1190 return parser
1193def main() -> int:
1194 parser = build_parser()
1195 args = parser.parse_args()
1196 if not args.query and not args.batch:
1197 parser.print_help()
1198 return 1
1199 try:
1200 context = _load_context(args.context_file)
1201 research = ResearchLookup(
1202 force_backend=args.force_backend,
1203 academic=args.academic,
1204 target_references=args.target_references,
1205 search_mode=args.search_mode,
1206 max_results=args.max_results,
1207 include_domains=_split_domains(args.include_domains),
1208 after_date=args.after_date,
1209 extract_limit=0 if args.no_extract else args.extract_limit,
1210 processor=args.processor,
1211 chat_model=args.chat_model,
1212 previous_interaction_id=args.previous_interaction_id,
1213 allow_perplexity_fallback=args.fallback_perplexity,
1214 manuscript_context=context,
1215 )
1216 except Exception as exc:
1217 print(f"Error: {exc}", file=sys.stderr)
1218 return 1
1220 queries = args.batch or [args.query]
1221 print(f"Running research for {len(queries)} query(s)...", file=sys.stderr)
1222 results = research.batch_lookup(queries, delay=args.batch_delay)
1224 if args.packet_dir:
1225 for index, result in enumerate(results):
1226 packet = result.get("packet")
1227 if not packet:
1228 continue
1229 destination = Path(args.packet_dir)
1230 if len(results) > 1:
1231 destination = destination / f"{index + 1:02d}-{_slug(result['query'])}"
1232 result["artifacts"] = save_packet(packet, destination)
1234 if args.json:
1235 rendered = json.dumps(results, indent=2, ensure_ascii=False, default=str) + "\n"
1236 else:
1237 rendered = "\n".join(
1238 _render_human_result(result, index)
1239 for index, result in enumerate(results, start=1)
1240 )
1241 rendered += "\n"
1243 if args.output:
1244 output_path = Path(args.output)
1245 output_path.parent.mkdir(parents=True, exist_ok=True)
1246 output_path.write_text(rendered, encoding="utf-8")
1247 else:
1248 print(rendered, end="")
1249 return 0 if all(result.get("success") for result in results) else 1
1252if __name__ == "__main__":
1253 sys.exit(main())