Research Agent Architecture

Research Agent Architecture

Autonomous Research Discovery, Validation, and Implementation Pipeline

Author: Claude (Architecture Design)
Date: 2026-01-14
Status: DESIGN PROPOSAL

Executive Summary

This document describes an autonomous research agent system that:
1. Discovers new research (papers, code, discussions)
2. Validates claims technically and mathematically
3. Implements validated concepts as code modules
4. Integrates with version control (git workflow)
5. Reports findings to human for review

The system operates in a human-in-the-loop mode: agents propose, humans approve critical decisions.

System Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                        RESEARCH AGENT ORCHESTRATOR                       │
│                                                                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐   │
│  │  Discovery  │  │ Validation  │  │Implementation│  │  Version    │   │
│  │   Agent     │──│   Agent     │──│    Agent    │──│  Control    │   │
│  │             │  │             │  │             │  │   Agent     │   │
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘   │
│         │                │                │                │           │
│         ▼                ▼                ▼                ▼           │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │                     SHARED STATE (research_state.json)           │  │
│  │  - discovered_papers[]                                           │  │
│  │  - validation_results[]                                          │  │
│  │  - implementation_queue[]                                        │  │
│  │  - git_branches[]                                                │  │
│  └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐  │
│  │                     HUMAN APPROVAL GATES                         │  │
│  │  - New research topics to track                                  │  │
│  │  - Claims marked for rejection                                   │  │
│  │  - Implementation PRs to merge                                   │  │
│  └─────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

Agent Specifications

1. Discovery Agent

Purpose: Find new research in target domains

Data Sources:

• arXiv API (cs.IR, cs.AI, cs.CL, quant-ph)
• Semantic Scholar API
• GitHub trending repositories
• HuggingFace new models
• Twitter/X academic accounts
• Conference proceedings (NeurIPS, ICML, ACL, EMNLP)

Capabilities:

class DiscoveryAgent:
    """Autonomous research discovery."""

    def __init__(self, config: DiscoveryConfig):
        self.topics = config.topics  # ["hyperdimensional computing", "GraphRAG", ...]
        self.sources = config.sources
        self.state_file = config.state_file

    async def discover(self) -> List[ResearchItem]:
        """Run discovery cycle."""
        items = []

        # 1. arXiv search
        items.extend(await self._search_arxiv())

        # 2. Semantic Scholar citations
        items.extend(await self._check_citations())

        # 3. GitHub repos
        items.extend(await self._search_github())

        # 4. Deduplicate and rank
        ranked = self._rank_by_relevance(items)

        return ranked[:self.config.max_items_per_cycle]

    async def _search_arxiv(self) -> List[ResearchItem]:
        """Search arXiv for new papers."""
        # Uses arxiv API: https://arxiv.org/help/api
        pass

    def _rank_by_relevance(self, items: List[ResearchItem]) -> List[ResearchItem]:
        """Rank by relevance to project goals."""
        # Score based on:
        # - Keyword match to topics
        # - Citation count / velocity
        # - Author reputation
        # - Recency
        pass

Output: discovered_papers.json with structured entries

2. Validation Agent

Purpose: Technically validate research claims

Validation Types:
1. Mathematical: Check proofs, complexity claims, bounds
2. Empirical: Verify benchmarks are reproducible
3. Conceptual: Check logical consistency
4. Reference: Verify citations exist and support claims

Capabilities:

class ValidationAgent:
    """Technical validation of research claims."""

    def __init__(self, config: ValidationConfig):
        self.llm = config.llm  # Claude for reasoning
        self.reference_db = config.reference_db

    async def validate(self, paper: ResearchItem) -> ValidationResult:
        """Validate a research paper."""

        # 1. Extract claims
        claims = await self._extract_claims(paper)

        # 2. Validate each claim
        results = []
        for claim in claims:
            result = await self._validate_claim(claim)
            results.append(result)

        # 3. Aggregate verdict
        return self._aggregate_results(results)

    async def _extract_claims(self, paper: ResearchItem) -> List[Claim]:
        """Extract testable claims from paper."""
        prompt = f"""
        Extract all testable claims from this paper:

        Title: {paper.title}
        Abstract: {paper.abstract}

        For each claim, identify:
        1. The claim statement
        2. Claim type: [correctness, quality, runtime, hardware]
        3. Evidence provided
        4. Required validation method
        """
        # Use LLM to extract
        pass

    async def _validate_claim(self, claim: Claim) -> ClaimValidation:
        """Validate a single claim."""

        if claim.type == "mathematical":
            return await self._validate_math(claim)
        elif claim.type == "empirical":
            return await self._validate_empirical(claim)
        elif claim.type == "reference":
            return await self._validate_reference(claim)
        else:
            return await self._validate_conceptual(claim)

    async def _validate_math(self, claim: Claim) -> ClaimValidation:
        """Validate mathematical claims."""
        prompt = f"""
        Validate this mathematical claim:

        Claim: {claim.statement}
        Proof provided: {claim.evidence}

        Check:
        1. Is the proof logically sound?
        2. Are the assumptions stated?
        3. Does the conclusion follow?
        4. Are there known counterexamples?

        Provide verdict: CONFIRMED / NEEDS_CORRECTION / REJECTED
        With detailed reasoning.
        """
        pass

Output: validation_results.json with per-claim verdicts

3. Implementation Agent

Purpose: Convert validated research into code

Capabilities:

class ImplementationAgent:
    """Convert research to code modules."""

    def __init__(self, config: ImplementationConfig):
        self.codebase_path = config.codebase_path
        self.test_runner = config.test_runner

    async def implement(self, validated: ValidationResult) -> ImplementationResult:
        """Implement validated research."""

        # 1. Design module structure
        design = await self._design_module(validated)

        # 2. Generate code
        code = await self._generate_code(design)

        # 3. Generate tests
        tests = await self._generate_tests(code)

        # 4. Run tests
        test_results = await self._run_tests(tests)

        # 5. Create integration plan
        integration = await self._plan_integration(code)

        return ImplementationResult(
            code=code,
            tests=tests,
            test_results=test_results,
            integration_plan=integration
        )

    async def _design_module(self, validated: ValidationResult) -> ModuleDesign:
        """Design module structure."""
        prompt = f"""
        Design a Python module to implement:

        Research: {validated.paper.title}
        Key concepts: {validated.accepted_claims}

        Existing codebase structure:
        - enhanced_quantum_rag.py (main RAG system)
        - HolographicMemory class (outer product memory)
        - EnhancedGroverSearch class (Grover's algorithm)

        Design a module that:
        1. Follows existing patterns
        2. Integrates via clear interfaces
        3. Is testable in isolation
        4. Has minimal dependencies
        """
        pass

    async def _generate_code(self, design: ModuleDesign) -> str:
        """Generate implementation code."""
        pass

    async def _generate_tests(self, code: str) -> str:
        """Generate test suite for implementation."""
        pass

Output: Code files + test files in staging branch

4. Version Control Agent

Purpose: Manage git workflow for research integrations

Capabilities:

class VersionControlAgent:
    """Git workflow management."""

    def __init__(self, config: VCConfig):
        self.repo_path = config.repo_path
        self.main_branch = config.main_branch

    async def create_feature_branch(self,
                                    implementation: ImplementationResult) -> str:
        """Create feature branch for implementation."""
        branch_name = f"research/{implementation.paper_id}"

        # Create branch
        await self._run_git(f"checkout -b {branch_name}")

        # Write files
        for file in implementation.files:
            await self._write_file(file.path, file.content)

        # Commit
        await self._run_git(f"add .")
        await self._run_git(
            f'commit -m "feat: implement {implementation.paper.title[:50]}"'
        )

        return branch_name

    async def create_pr(self, branch: str,
                        validation: ValidationResult) -> PRResult:
        """Create pull request with validation summary."""

        pr_body = f"""
## Research Implementation

**Paper**: {validation.paper.title}
**arXiv**: {validation.paper.arxiv_id}

### Validation Summary
- Claims validated: {validation.confirmed_count}
- Claims corrected: {validation.corrected_count}
- Claims rejected: {validation.rejected_count}

### Changes
{self._generate_change_summary()}

### Test Results
{self._format_test_results()}

---
Generated by Research Agent Pipeline
        """

        result = await self._run_gh(
            f'pr create --title "feat: {validation.paper.title[:50]}" '
            f'--body "{pr_body}"'
        )

        return PRResult(url=result.url, number=result.number)

Orchestrator

Purpose: Coordinate all agents in research pipeline

class ResearchOrchestrator:
    """Orchestrate the research agent pipeline."""

    def __init__(self, config: OrchestratorConfig):
        self.discovery = DiscoveryAgent(config.discovery)
        self.validation = ValidationAgent(config.validation)
        self.implementation = ImplementationAgent(config.implementation)
        self.version_control = VersionControlAgent(config.vc)

        self.state_file = config.state_file
        self.approval_queue = config.approval_queue

    async def run_cycle(self) -> CycleResult:
        """Run one research cycle."""

        # 1. Discovery
        print("Phase 1: Discovery...")
        discovered = await self.discovery.discover()

        # 2. Filter new papers
        new_papers = self._filter_already_processed(discovered)

        # 3. Validation (parallel)
        print(f"Phase 2: Validating {len(new_papers)} papers...")
        validations = await asyncio.gather(*[
            self.validation.validate(paper)
            for paper in new_papers
        ])

        # 4. Queue approved for implementation
        approved = [v for v in validations if v.verdict == "APPROVED"]

        # 5. Implementation (sequential - needs care)
        print(f"Phase 3: Implementing {len(approved)} papers...")
        implementations = []
        for validation in approved:
            impl = await self.implementation.implement(validation)
            implementations.append(impl)

        # 6. Version control
        print("Phase 4: Creating branches and PRs...")
        prs = []
        for impl in implementations:
            branch = await self.version_control.create_feature_branch(impl)
            pr = await self.version_control.create_pr(branch, impl.validation)
            prs.append(pr)

        # 7. Update state
        await self._update_state(discovered, validations, implementations, prs)

        # 8. Generate report
        return self._generate_cycle_report()

    async def run_continuous(self, interval_hours: int = 24):
        """Run continuous research cycles."""
        while True:
            try:
                result = await self.run_cycle()
                await self._notify_human(result)
            except Exception as e:
                await self._handle_error(e)

            await asyncio.sleep(interval_hours * 3600)

State Management

{
  "last_cycle": "2026-01-14T10:00:00Z",
  "topics": [
    "hyperdimensional computing",
    "GraphRAG",
    "sparse distributed memory",
    "neuro-symbolic AI",
    "quantum retrieval"
  ],
  "discovered_papers": [
    {
      "id": "arxiv:2501.12345",
      "title": "...",
      "discovered_at": "2026-01-14",
      "status": "validated",
      "validation_id": "val_001"
    }
  ],
  "validation_results": [
    {
      "id": "val_001",
      "paper_id": "arxiv:2501.12345",
      "verdict": "APPROVED_WITH_CORRECTIONS",
      "claims_confirmed": 5,
      "claims_corrected": 2,
      "claims_rejected": 1
    }
  ],
  "implementations": [
    {
      "id": "impl_001",
      "validation_id": "val_001",
      "branch": "research/arxiv-2501-12345",
      "pr_url": "https://github.com/.../pull/42",
      "status": "pending_review"
    }
  ],
  "human_approval_queue": [
    {
      "type": "reject_claim",
      "claim": "HDC is always faster than transformers",
      "reason": "No benchmark evidence",
      "awaiting_approval": true
    }
  ]
}

Human Approval Gates

The system pauses for human approval at these points:

Configuration

# research_agent_config.yaml

discovery:
  topics:
    - "hyperdimensional computing"
    - "GraphRAG"
    - "sparse distributed memory"
    - "modern hopfield networks"
    - "quantum machine learning"
    - "neuro-symbolic AI"
    - "vector symbolic architectures"

  sources:
    arxiv:
      categories: ["cs.IR", "cs.AI", "cs.CL", "cs.LG", "quant-ph"]
      max_results_per_query: 50

    semantic_scholar:
      fields_of_study: ["Computer Science", "Physics"]
      min_citations: 5

    github:
      topics: ["hyperdimensional-computing", "graphrag", "rag"]
      min_stars: 50

  schedule:
    interval_hours: 24
    timezone: "UTC"

validation:
  llm:
    provider: "anthropic"
    model: "claude-sonnet-4-20250514"

  claim_types:
    - correctness
    - quality
    - runtime
    - hardware

  thresholds:
    min_evidence_score: 0.7
    require_reproducible: true

implementation:
  codebase_path: "Documents/Learning/quantum_rag_demo"
  test_command: "./quantum_env/bin/python -m pytest tests/ -v"

  patterns:
    module_template: "templates/module.py.j2"
    test_template: "templates/test.py.j2"

  integration_rules:
    - "Must not break existing tests"
    - "Must have >80% test coverage"
    - "Must follow existing code patterns"

version_control:
  repo_path: "Documents/Learning/quantum_rag_demo"
  main_branch: "main"
  branch_prefix: "research/"

  commit_format: "feat: implement {paper_title}"

  pr_template: |
    ## Research Implementation

    **Paper**: {paper_title}
    **arXiv**: {arxiv_id}

    ### Validation Summary
    {validation_summary}

    ### Changes
    {change_summary}

notifications:
  slack_webhook: "${SLACK_WEBHOOK_URL}"
  email: "kulbir.minhas@aiinitiative.co.uk"

Implementation Plan

Phase 1: Core Infrastructure (1-2 days)

research_agent/
├── __init__.py
├── config.py           # Configuration management
├── state.py            # State persistence
├── agents/
│   ├── __init__.py
│   ├── discovery.py    # Discovery agent
│   ├── validation.py   # Validation agent
│   ├── implementation.py # Implementation agent
│   └── version_control.py # Git workflow
├── orchestrator.py     # Main orchestrator
├── models.py           # Data models (Pydantic)
└── cli.py              # Command-line interface

Phase 2: Discovery Agent (2-3 days)

• arXiv API integration
• Semantic Scholar API integration
• GitHub search integration
• Deduplication and ranking

Phase 3: Validation Agent (3-5 days)

• Claim extraction pipeline
• Mathematical validation prompts
• Reference verification
• Validation report generation

Phase 4: Implementation Agent (1 week)

• Code generation templates
• Test generation
• Integration planning
• Style conformance checking

Phase 5: Version Control Agent (1-2 days)

• Git operations
• PR creation with gh
• Branch management

Phase 6: Orchestration & UI (2-3 days)

• Async orchestration
• Human approval interface
• Notification system
• Dashboard (optional)

Quick Start: Minimal Viable Agent

For immediate use, here's a simplified version:

#!/usr/bin/env python3
"""
Minimal Research Agent - Single-file implementation.
Run: python research_agent.py --cycle
"""

import asyncio
import json
import os
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import List, Optional

import httpx  # pip install httpx

# ============================================================================
# Configuration
# ============================================================================

TOPICS = [
    "hyperdimensional computing retrieval",
    "GraphRAG knowledge graph",
    "sparse distributed memory transformer",
    "neuro-symbolic reasoning",
]

STATE_FILE = "research_state.json"
ARXIV_API = "http://export.arxiv.org/api/query"

# ============================================================================
# Models
# ============================================================================

@dataclass
class Paper:
    id: str
    title: str
    abstract: str
    authors: List[str]
    published: str
    arxiv_url: str
    status: str = "discovered"  # discovered, validated, implemented

@dataclass
class ResearchState:
    last_cycle: str
    papers: List[Paper]

# ============================================================================
# Discovery
# ============================================================================

async def search_arxiv(topic: str, max_results: int = 10) -> List[Paper]:
    """Search arXiv for papers on topic."""
    async with httpx.AsyncClient() as client:
        params = {
            "search_query": f"all:{topic}",
            "start": 0,
            "max_results": max_results,
            "sortBy": "submittedDate",
            "sortOrder": "descending"
        }

        response = await client.get(ARXIV_API, params=params)

        # Parse Atom XML (simplified)
        # In production, use feedparser
        papers = []
        # ... parsing logic ...

        return papers

async def discover_papers() -> List[Paper]:
    """Discover new papers across all topics."""
    all_papers = []

    for topic in TOPICS:
        papers = await search_arxiv(topic)
        all_papers.extend(papers)

    # Deduplicate by ID
    seen = set()
    unique = []
    for p in all_papers:
        if p.id not in seen:
            seen.add(p.id)
            unique.append(p)

    return unique

# ============================================================================
# Validation (Stub - Integrate with Claude)
# ============================================================================

async def validate_paper(paper: Paper) -> dict:
    """Validate a paper's claims."""
    # In production, use Claude API to:
    # 1. Extract claims from abstract
    # 2. Validate each claim
    # 3. Return structured results

    return {
        "paper_id": paper.id,
        "verdict": "NEEDS_REVIEW",
        "claims": [],
        "note": "Automated validation pending - requires Claude integration"
    }

# ============================================================================
# State Management
# ============================================================================

def load_state() -> ResearchState:
    """Load research state from file."""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            data = json.load(f)
            papers = [Paper(**p) for p in data.get("papers", [])]
            return ResearchState(
                last_cycle=data.get("last_cycle", ""),
                papers=papers
            )
    return ResearchState(last_cycle="", papers=[])

def save_state(state: ResearchState):
    """Save research state to file."""
    data = {
        "last_cycle": state.last_cycle,
        "papers": [asdict(p) for p in state.papers]
    }
    with open(STATE_FILE, "w") as f:
        json.dump(data, f, indent=2)

# ============================================================================
# Main Cycle
# ============================================================================

async def run_cycle():
    """Run one research discovery cycle."""
    print("=" * 60)
    print("RESEARCH AGENT CYCLE")
    print("=" * 60)

    # Load state
    state = load_state()
    existing_ids = {p.id for p in state.papers}

    # Discover
    print(f"\n[1/3] Discovering papers for {len(TOPICS)} topics...")
    discovered = await discover_papers()

    # Filter new
    new_papers = [p for p in discovered if p.id not in existing_ids]
    print(f"      Found {len(new_papers)} new papers")

    # Validate (stub)
    print(f"\n[2/3] Validating papers...")
    for paper in new_papers[:5]:  # Limit for demo
        result = await validate_paper(paper)
        print(f"      - {paper.title[:50]}... [{result['verdict']}]")
        paper.status = "validated"

    # Update state
    print(f"\n[3/3] Updating state...")
    state.papers.extend(new_papers)
    state.last_cycle = datetime.utcnow().isoformat()
    save_state(state)

    # Report
    print(f"\n" + "=" * 60)
    print(f"CYCLE COMPLETE")
    print(f"  New papers: {len(new_papers)}")
    print(f"  Total tracked: {len(state.papers)}")
    print(f"  State saved to: {STATE_FILE}")
    print("=" * 60)

# ============================================================================
# CLI
# ============================================================================

if __name__ == "__main__":
    import sys

    if "--cycle" in sys.argv:
        asyncio.run(run_cycle())
    elif "--status" in sys.argv:
        state = load_state()
        print(f"Last cycle: {state.last_cycle}")
        print(f"Papers tracked: {len(state.papers)}")
        for p in state.papers[-5:]:
            print(f"  - [{p.status}] {p.title[:60]}...")
    else:
        print("Usage: python research_agent.py [--cycle|--status]")

Integration with Existing Project

The research agent integrates with quantum_rag_demo via:

1. Output Location: research_agent/ directory in project root
2. State File: research_state.json tracked in git
3. Generated Code: Goes to research_agent/implementations/
4. PRs: Created against main branch

quantum_rag_demo/
├── enhanced_quantum_rag.py      # Existing
├── research_agent/              # NEW
│   ├── __init__.py
│   ├── agents/
│   ├── config.yaml
│   ├── research_state.json
│   └── implementations/         # Generated modules
│       ├── hdc_module.py
│       └── graphrag_module.py
└── docs/
    └── research/
        └── ADVANCE_RETRIEVAL.md # Existing

Running the Agent

Manual Cycle

cd quantum_rag_demo
python -m research_agent --cycle

Scheduled (cron)

# Run daily at 6 AM
0 6 * * * cd /path/to/quantum_rag_demo && ./quantum_env/bin/python -m research_agent --cycle >> research_agent.log 2>&1

Continuous (background)

nohup python -m research_agent --continuous --interval 24 &

Security Considerations

1. API Keys: Store in environment variables, not config files
2. Code Generation: All generated code goes through PR review
3. Rate Limits: Respect arXiv and other API rate limits
4. Human Gate: Critical decisions require human approval

Future Enhancements

1. Citation Graph Analysis: Track which papers cite your implemented methods
2. Benchmark Automation: Auto-run benchmarks on new implementations
3. Multi-repo Support: Track and integrate across multiple projects
4. Collaboration: Multiple humans can review and approve
5. Learning: Track which papers led to successful implementations

Architecture designed: 2026-01-14
Next step: Implement Phase 1 (Core Infrastructure)