mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
246 lines
8.4 KiB
Python
246 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
LLM-assisted re-mapping of arXiv findings to unified equation symbols.
|
||
|
||
Supports:
|
||
- OpenAI API (GPT-4o-mini is cheap and fast)
|
||
- Anthropic API (Claude 3 Haiku)
|
||
- Local Ollama (free, no API key)
|
||
|
||
Usage:
|
||
export OPENAI_API_KEY="sk-..."
|
||
python llm_remapper.py --provider openai --input arxiv_findings_500.md --output arxiv_llm_mapped.md
|
||
|
||
# Or local Ollama (free)
|
||
python llm_remapper.py --provider ollama --model llama3.2 --input arxiv_findings_500.md
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import dataclass
|
||
from typing import Optional
|
||
|
||
import requests
|
||
|
||
|
||
@dataclass
|
||
class Entry:
|
||
number: int
|
||
title: str
|
||
url: str
|
||
year: str
|
||
summary: str
|
||
|
||
|
||
def parse_entries(text: str) -> list[Entry]:
|
||
entries = []
|
||
pattern = re.compile(
|
||
r"## (\d+)\. (.+?)\n\n"
|
||
r"\*\*Source:\*\* \[([^\]]+)\]\([^)]+\)\s+\((\d{4})\)\n\n"
|
||
r"\*\*Summary:\*\* (.+?)\n\n"
|
||
r"\| Symbol \| Mapping \|\n\|--------\|---------\|\n"
|
||
r"\| Ω \| .+? \|\n"
|
||
r"\| Ψ \| .+? \|\n"
|
||
r"\| B \| .+? \|\n"
|
||
r"\| C \| .+? \|\n"
|
||
r"\| Δ \| .+? \|\n",
|
||
re.DOTALL,
|
||
)
|
||
for m in pattern.finditer(text):
|
||
entries.append(
|
||
Entry(
|
||
number=int(m.group(1)),
|
||
title=m.group(2).strip(),
|
||
url=m.group(3).strip(),
|
||
year=m.group(4).strip(),
|
||
summary=m.group(5).strip()[:600],
|
||
)
|
||
)
|
||
return entries
|
||
|
||
|
||
PROMPT_TEMPLATE = """You are a precise scientific classifier. Map the following paper to five symbols from this universal equation:
|
||
|
||
Ω = Ψ [ B(θ) ⊗ C(n, α) ] ⊕ Δ(n, θ, α)
|
||
|
||
Definitions:
|
||
- Ω (Omega): The observable output, the measured phenomenon, the result.
|
||
- Ψ (Psi): The operator, mechanism, or theory that combines basis and context.
|
||
- B (Basis): The conserved, reusable, fundamental component (basis vectors, genes, spacetime metric, etc.).
|
||
- C (Context): The dynamic, adaptive, variable parameter (environment, initial conditions, input data).
|
||
- Δ (Delta): The irreducible residual — noise, uncertainty, error, fundamental limit.
|
||
|
||
Paper:
|
||
Title: {title}
|
||
Abstract: {summary}
|
||
|
||
Respond ONLY in valid JSON with exactly these five keys and short (≤15 words) values:
|
||
{{"Ω": "...", "Ψ": "...", "B": "...", "C": "...", "Δ": "..."}}
|
||
"""
|
||
|
||
|
||
def call_openai(title: str, summary: str, api_key: str, model: str = "gpt-4o-mini") -> Optional[dict]:
|
||
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
||
payload = {
|
||
"model": model,
|
||
"messages": [
|
||
{"role": "system", "content": "You are a scientific classifier."},
|
||
{"role": "user", "content": PROMPT_TEMPLATE.format(title=title, summary=summary)},
|
||
],
|
||
"temperature": 0.1,
|
||
"max_tokens": 150,
|
||
}
|
||
try:
|
||
r = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload, timeout=30)
|
||
r.raise_for_status()
|
||
content = r.json()["choices"][0]["message"]["content"]
|
||
# Extract JSON
|
||
match = re.search(r'\{[^}]+\}', content)
|
||
if match:
|
||
return json.loads(match.group(0))
|
||
except Exception as e:
|
||
print(f" OpenAI error: {e}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def call_anthropic(title: str, summary: str, api_key: str, model: str = "claude-3-haiku-20240307") -> Optional[dict]:
|
||
headers = {"x-api-key": api_key, "Content-Type": "application/json", "anthropic-version": "2023-06-01"}
|
||
payload = {
|
||
"model": model,
|
||
"max_tokens": 150,
|
||
"temperature": 0.1,
|
||
"messages": [{"role": "user", "content": PROMPT_TEMPLATE.format(title=title, summary=summary)}],
|
||
}
|
||
try:
|
||
r = requests.post("https://api.anthropic.com/v1/messages", headers=headers, json=payload, timeout=30)
|
||
r.raise_for_status()
|
||
content = r.json()["content"][0]["text"]
|
||
match = re.search(r'\{[^}]+\}', content)
|
||
if match:
|
||
return json.loads(match.group(0))
|
||
except Exception as e:
|
||
print(f" Anthropic error: {e}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def call_ollama(title: str, summary: str, model: str = "llama3.2", host: str = "http://localhost:11434") -> Optional[dict]:
|
||
payload = {
|
||
"model": model,
|
||
"prompt": PROMPT_TEMPLATE.format(title=title, summary=summary),
|
||
"stream": False,
|
||
"options": {"temperature": 0.1, "num_predict": 150},
|
||
}
|
||
try:
|
||
r = requests.post(f"{host}/api/generate", json=payload, timeout=60)
|
||
r.raise_for_status()
|
||
content = r.json()["response"]
|
||
match = re.search(r'\{[^}]+\}', content)
|
||
if match:
|
||
return json.loads(match.group(0))
|
||
except Exception as e:
|
||
print(f" Ollama error: {e}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def map_entry(entry: Entry, provider: str, model: str, api_key: Optional[str]) -> Optional[dict]:
|
||
if provider == "openai":
|
||
if not api_key:
|
||
print("ERROR: OPENAI_API_KEY not set", file=sys.stderr)
|
||
return None
|
||
return call_openai(entry.title, entry.summary, api_key, model)
|
||
elif provider == "anthropic":
|
||
if not api_key:
|
||
print("ERROR: ANTHROPIC_API_KEY not set", file=sys.stderr)
|
||
return None
|
||
return call_anthropic(entry.title, entry.summary, api_key, model)
|
||
elif provider == "ollama":
|
||
return call_ollama(entry.title, entry.summary, model)
|
||
else:
|
||
print(f"ERROR: Unknown provider {provider}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def render_entry(e: Entry, mapping: dict) -> str:
|
||
return (
|
||
f"## {e.number}. {e.title}\n\n"
|
||
f"**Source:** [{e.url}]({e.url}) ({e.year})\n\n"
|
||
f"**Summary:** {e.summary[:500]}\n\n"
|
||
f"| Symbol | Mapping |\n"
|
||
f"|--------|---------|\n"
|
||
f"| Ω | {mapping.get('Ω', 'N/A')} |\n"
|
||
f"| Ψ | {mapping.get('Ψ', 'N/A')} |\n"
|
||
f"| B | {mapping.get('B', 'N/A')} |\n"
|
||
f"| C | {mapping.get('C', 'N/A')} |\n"
|
||
f"| Δ | {mapping.get('Δ', 'N/A')} |\n\n"
|
||
f"---\n\n"
|
||
)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--input", required=True, help="Input markdown file")
|
||
parser.add_argument("--output", required=True, help="Output markdown file")
|
||
parser.add_argument("--provider", choices=["openai", "anthropic", "ollama"], default="ollama")
|
||
parser.add_argument("--model", default="llama3.2", help="Model name")
|
||
parser.add_argument("--workers", type=int, default=4, help="Parallel workers")
|
||
parser.add_argument("--max", type=int, default=0, help="Max entries (0 = all)")
|
||
args = parser.parse_args()
|
||
|
||
api_key = os.getenv("OPENAI_API_KEY") if args.provider == "openai" else os.getenv("ANTHROPIC_API_KEY")
|
||
|
||
with open(args.input, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
|
||
entries = parse_entries(text)
|
||
if args.max:
|
||
entries = entries[:args.max]
|
||
|
||
print(f"Entries: {len(entries)} | Provider: {args.provider} | Model: {args.model}")
|
||
print("=" * 60)
|
||
|
||
results = {}
|
||
|
||
with ThreadPoolExecutor(max_workers=args.workers) as executor:
|
||
futures = {executor.submit(map_entry, e, args.provider, args.model, api_key): e for e in entries}
|
||
for future in as_completed(futures):
|
||
e = futures[future]
|
||
mapping = future.result()
|
||
if mapping:
|
||
results[e.number] = mapping
|
||
print(f" ✓ #{e.number}: {e.title[:50]}...")
|
||
else:
|
||
print(f" ✗ #{e.number}: FAILED")
|
||
time.sleep(0.5) # rate limit politeness
|
||
|
||
# Build output
|
||
lines = [
|
||
f"# ArXiv Findings — LLM-Mapped ({args.provider}/{args.model})",
|
||
"",
|
||
f"**Papers:** {len(entries)}",
|
||
f"**Successfully mapped:** {len(results)}",
|
||
f"**Equation:** Ω = Ψ [ B(θ) ⊗ C(n, α) ] ⊕ Δ(n, θ, α)",
|
||
"",
|
||
"---",
|
||
"",
|
||
]
|
||
|
||
for e in entries:
|
||
mapping = results.get(e.number, {
|
||
"Ω": "N/A", "Ψ": "N/A", "B": "N/A", "C": "N/A", "Δ": "N/A"
|
||
})
|
||
lines.append(render_entry(e, mapping))
|
||
|
||
with open(args.output, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines))
|
||
|
||
print(f"\nWrote {args.output}")
|
||
print(f"Success rate: {len(results)}/{len(entries)} ({100*len(results)//len(entries)}%)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|