Research Agents

Agents specialized for information gathering, synthesis, hypothesis generation, and experimental design.

Research Agent Capabilities

Research Agent Architectures

1. Literature Review Agent

class LiteratureReviewAgent:
    def __init__(self):
        self.sources = {
            "semantic_scholar": SemanticScholarAPI(),
            "arxiv": ArxivAPI(),
            "pubmed": PubMedAPI(),
            "crossref": CrossrefAPI(),
            "google_scholar": GoogleScholarScraper()
        }
        self.embedder = EmbeddingModel("specter2")  # Scientific embeddings

    async def search(self, query: SearchQuery) -> SearchResults:
        # Parallel search across sources
        tasks = [
            source.search(query) for source in self.sources.values()
        ]
        raw_results = await asyncio.gather(*tasks)

        # Deduplicate by DOI/title
        papers = self.deduplicate(raw_results)

        # Rank by relevance
        ranked = self.rank_by_relevance(papers, query)

        return SearchResults(papers=ranked[:query.max_results])

    async def synthesize(self, papers: list[Paper], focus: str) -> Synthesis:
        # Extract key information
        extractions = await asyncio.gather(*[
            self.extract_key_info(p, focus) for p in papers
        ])

        # Identify themes
        themes = self.identify_themes(extractions)

        # Find contradictions
        contradictions = self.find_contradictions(extractions)

        # Identify gaps
        gaps = self.identify_gaps(extractions, focus)

        return Synthesis(
            summary=self.generate_summary(extractions),
            themes=themes,
            contradictions=contradictions,
            gaps=gaps,
            citations=self.format_citations(papers)
        )

2. Hypothesis Generation Agent

class HypothesisAgent:
    def __init__(self, knowledge_graph: KnowledgeGraph):
        self.kg = knowledge_graph
        self.embedder = EmbeddingModel("specter2")

    async def generate_hypotheses(self, 
                                  domain: str, 
                                  observation: str,
                                  num_hypotheses: int = 5) -> list[Hypothesis]:
        # 1. Retrieve relevant knowledge
        context = await self.kg.query(domain, observation)

        # 2. Identify gaps & anomalies
        gaps = await self.identify_gaps(context, observation)

        # 3. Generate hypotheses via analogical reasoning
        hypotheses = []
        for gap in gaps[:num_hypotheses]:
            hypothesis = await self.generate_from_gap(gap, context)
            hypotheses.append(hypothesis)

        # 4. Rank by plausibility & testability
        ranked = await self.rank_hypotheses(hypotheses)

        return ranked

    async def generate_from_gap(self, gap: KnowledgeGap, context: str) -> Hypothesis:
        prompt = f"""
        Domain knowledge: {context}
        Knowledge gap: {gap.description}

        Generate a testable hypothesis that addresses this gap.
        Format: If [condition], then [prediction], because [mechanism].
        Testable via: [experimental approach].
        """
        response = await self.llm.generate(prompt)
        return self.parse_hypothesis(response)

3. Experimental Design Agent

class ExperimentalDesignAgent:
    def __init__(self):
        self.stats_tools = StatisticalTools()
        self.simulator = ExperimentSimulator()

    async def design_experiment(self, hypothesis: Hypothesis) -> ExperimentDesign:
        # 1. Identify variables
        variables = self.identify_variables(hypothesis)

        # 2. Determine design type
        design_type = self.select_design(variables)

        # 3. Power analysis
        power_analysis = await self.stats_tools.power_analysis(
            effect_size=hypothesis.expected_effect,
            alpha=0.05,
            power=0.8,
            design=design_type
        )

        # 3. Randomization & blocking
        randomization = self.design_randomization(variables)

        # 4. Analysis plan
        analysis_plan = self.create_analysis_plan(variables, design_type)

        return ExperimentDesign(
            hypothesis=hypothesis,
            design_type=design_type,
            variables=variables,
            sample_size=power_analysis.sample_size,
            power_analysis=power_analysis,
            randomization=randomization,
            analysis_plan=analysis_plan,
            timeline=self.estimate_timeline(power_analysis.sample_size)
        )

    def select_design(self, variables: Variables) -> DesignType:
        if variables.has_treatment and variables.has_control:
            return DesignType.RCT
        elif variables.has_repeated_measures:
            return DesignType.REPEATED_MEASURES
        elif variables.has_multiple_factors:
            return DesignType.FACTORIAL
        else:
            return DesignType.OBSERVATIONAL

4. Data Analysis Agent

class DataAnalysisAgent:
    def __init__(self):
        self.tools = {
            "pandas": PandasTools(),
            "statsmodels": StatsModelsTools(),
            "scipy": SciPyTools(),
            "plotly": PlottingTools()
        }

    async def analyze(self, dataset: Dataset, questions: list[str]) -> AnalysisReport:
        # 1. Exploratory analysis
        eda = await self.exploratory_analysis(dataset)

        # 2. Answer specific questions
        answers = await asyncio.gather(*[
            self.answer_question(dataset, q) for q in questions
        ])

        # 3. Statistical tests
        stats = await self.run_statistical_tests(dataset, questions)

        # 4. Visualizations
        viz = await self.create_visualizations(dataset, answers)

        return AnalysisReport(
            eda=eda,
            answers=answers,
            statistics=stats,
            visualizations=viz,
            reproducibility=self.generate_notebook(dataset, answers)
        )

    async def answer_question(self, dataset: Dataset, question: str) -> Answer:
        # Use LLM to plan analysis
        plan = await self.llm.plan_analysis(dataset.schema, question)

        # Execute plan
        result = await self.execute_plan(dataset, plan)

        # Interpret
        interpretation = await self.llm.interpret(result, question)

        return Answer(
            question=question,
            result=result,
            interpretation=interpretation,
            confidence=self.assess_confidence(result)
        )

Specialized Research Domains

1. Systematic Review Agent

class SystematicReviewAgent:
    def __init__(self):
        self.prisma_checker = PRISMAChecker()

    async def conduct_review(self, protocol: ReviewProtocol) -> SystematicReview:
        # 1. Search strategy
        search_results = await self.execute_search(protocol.search_strategy)

        # 2. Screening (title/abstract)
        screened = await self.screen_abstracts(search_results, protocol.inclusion_criteria)

        # 3. Full-text review
        included = await self.full_text_review(screened, protocol.inclusion_criteria)

        # 4. Data extraction
        extracted = await self.extract_data(included, protocol.extraction_form)

        # 5. Risk of bias
        rob = await self.assess_bias(included, protocol.bias_tool)

        # 5. Synthesis
        synthesis = await self.synthesize(extracted, protocol.synthesis_method)

        # 6. PRISMA flowchart
        prisma = self.generate_prisma_flowchart(search_results, screened, included)

        return SystematicReview(
            protocol=protocol,
            search_results=search_results,
            included_studies=included,
            extracted_data=extracted,
            risk_of_bias=rob,
            synthesis=synthesis,
            prisma_flowchart=prisma
        )

2. Computational Research Agent

class ComputationalResearchAgent:
    def __init__(self):
        self.hpc = HPCClient()
        self.containers = ContainerManager()

    async def run_simulation(self, spec: SimulationSpec) -> SimulationResult:
        # 1. Prepare environment
        container = await self.containers.prepare(spec.environment)

        # 2. Submit job
        job_id = await self.hpc.submit(
            container=container,
            command=spec.command,
            resources=spec.resources,
            walltime=spec.walltime
        )

        # 3. Monitor
        result = await self.monitor_job(job_id)

        # 4. Post-process
        return await self.post_process(result, spec.output_format)

    async def parameter_sweep(self, spec: SweepSpec) -> SweepResult:
        # Generate parameter combinations
        param_sets = list(ParameterGrid(spec.param_grid))

        # Submit batch
        jobs = await asyncio.gather(*[
            self.hpc.submit_array(
                command=f"{spec.command} --params {json.dumps(params)}",
                resources=spec.resources
            ) for params in param_sets
        ])

        # Aggregate results
        return await self.aggregate_results(jobs)

3. Scientific Writing Agent

class ScientificWritingAgent:
    def __init__(self):
        self.citation_manager = CitationManager()
        self.latex_compiler = LaTeXCompiler()

    async def write_paper(self, 
                          structure: PaperStructure, 
                          data: ResearchData) -> PaperDraft:
        # 1. Generate outline
        outline = await self.generate_outline(structure, data)

        # 2. Write sections
        sections = await asyncio.gather(*[
            self.write_section(section, data, outline) 
            for section in outline.sections
        ])

        # 2. Generate citations
        citations = await self.citation_manager.format_citations(
            data.references, style=structure.citation_style
        )

        # 3. Assemble
        draft = self.assemble_paper(outline, sections, citations)

        # 3. Compile LaTeX
        pdf = await self.latex_compiler.compile(draft.latex)

        return PaperDraft(
            latex=draft.latex,
            pdf=pdf,
            word_count=self.count_words(draft.latex),
            sections=sections
        )

    async def write_section(self, section: SectionSpec, data: ResearchData, outline: Outline) -> Section:
        prompt = f"""
        Write the {section.name} section for a paper with this outline:
        {outline}

        Available data: {data.summary}
        Key findings for this section: {section.key_points}
        Target length: {section.target_words} words
        Tone: Formal academic
        """
        content = await self.llm.generate(prompt)

        # Add citations
        cited = await self.add_citations(content, section.required_refs)

        return Section(name=section.name, content=cited, word_count=len(content.split()))

Research Tools & APIs

Literature APIs

Knowledge Graphs

Statistical Tools

Visualization

Research Workflows

1. Literature Review Workflow

graph TD
    A[Define Research Question] --> B[Design Search Strategy]
    B --> C[Execute Searches]
    C --> D[Deduplicate Results]
    D --> E[Screen Title/Abstract]
    E --> F[Full-Text Review]
    F --> G[Extract Data]
    G --> H[Assess Quality]
    H --> H[Synthesize]
    H --> I[Write Report]

2. Hypothesis-Driven Research

graph TD
    A[Observation] --> B[Literature Gap Analysis]
    B --> C[Hypothesis Generation]
    C --> D[Experimental Design]
    D --> E[Power Analysis]
    E --> F[IRB/Ethics Approval]
    F --> G[Data Collection]
    G --> H[Analysis]
    H --> I[Interpretation]
    I --> J[Manuscript Writing]

3. Computational Experiment

graph TD
    A[Hypothesis] --> B[Simulation Design]
    B --> C[Parameter Sweep Design]
    C --> D[HPC Job Submission]
    D --> E[Monitor & Monitor]
    E --> F[Result Aggregation]
    F --> G[Analysis & Visualization]
    G --> H[Interpretation]

Evaluation Metrics

Related Notes

• RAG Fundamentals — RAG for literature
• Advanced RAG — Advanced retrieval
• Knowledge Graphs — KG for research
• Memory Systems — Research memory
• API Integration — API integration
• Code Execution — Computational research
• Browser Automation — Web research
• Static Analysis — Code quality for research code
• Reasoning Models — Reasoning for hypotheses
• Small Language Models — SLMs for research
• Coding Agents — Code for research
• Data Analysis Agents — Data analysis
• Agent Benchmarks — Research benchmarks
• Scientific Agents — Scientific discovery agents
• Self Improving Agents — Self-improving research
• Research Directions — Research directions

Sources: Semantic Scholar API, arXiv API, PubMed API, OpenAlex, PRISMA guidelines, Cochrane Handbook, Campbell Collaboration, Systematic Review Toolbox, Rayyan, Covidence, DistillerSR, EPPI-Reviewer, Statistical Power Analysis, Experimental Design textbooks, Causal Inference (Pearl, Rubin), CausalML, DoWhy, HPC scheduling, Slurm, Kubernetes, Singularity/Apptainer, Nextflow, Snakemake, LaTeX, Overleaf, Zotero, Mendeley, Paperpile, Scholarcy, Elicit, Scite, Connected Papers, Research Rabbit, Litmaps, OpenAlex.