Agentic Architectures

• AI researcher & software engineer
• Works with RL, RLHF, and Graphs
• Enjoys clean abstractions & messy experiments
• Knowledge is power
• Trustee @ Society of Research Software Engineering
• Senior Member @ IEEE

import os
from datetime import datetime
from typing import Dict, List, Literal, TypedDict

from google.adk.agents.llm_agent import Agent
from google.adk.tools.function_tool import FunctionTool
import arxiv


GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")


class ArxivPaper(TypedDict):
    title: str
    authors: List[str]
    year: int
    arxiv_id: str
    url: str
    summary: str
    topic: str


def _build_arxiv_query_string(topics: List[str]) -> str:
    """Build an arXiv API query string from a list of topics.

    We keep this simple and AND the topic phrases together.
    """

    cleaned = list(dict.fromkeys(t.strip() for t in topics if t.strip()))
    if not cleaned:
        raise ValueError("At least one topic is required")

    # Example: "(ti:\"graph neural networks\" OR abs:\"graph neural networks\")"
    parts = []
    for t in cleaned:
        phrase = t.replace('"', '\\"')
        parts.append(f'(ti:"{phrase}" OR abs:"{phrase}")')

    return " AND ".join(parts)


def search_arxiv_with_client(
    topics: List[str],
    *,
    max_results: int = 50,
    sort_by: Literal[
        "relevance",
        "lastUpdatedDate",
        "submittedDate",
    ] = "submittedDate",
) -> List[ArxivPaper]:
    """Use the `arxiv` Python client to fetch recent papers.

    This is a low-level utility that the LLM agent can call via tool usage,
    or that you can call directly from Python.
    """

    query = _build_arxiv_query_string(topics)

    sort_criterion_map: Dict[str, arxiv.SortCriterion] = {
        "relevance": arxiv.SortCriterion.Relevance,
        "lastUpdatedDate": arxiv.SortCriterion.LastUpdatedDate,
        "submittedDate": arxiv.SortCriterion.SubmittedDate,
    }

    search = arxiv.Search(
        query=query,
        max_results=max_results,
        sort_by=sort_criterion_map[sort_by],
    )

    client = arxiv.Client()

    papers: List[ArxivPaper] = []
    for result in client.results(search):
        year = result.published.year if result.published else datetime.utcnow().year
        papers.append(
            ArxivPaper(
                title=result.title,
                authors=[a.name for a in result.authors],
                year=year,
                arxiv_id=result.get_short_id(),
                url=result.entry_id,
                summary=result.summary,
                topic=", ".join(topics),
            )
        )

    return papers


search_arxiv_with_client_tool = FunctionTool(func=search_arxiv_with_client)

arxiv_research_agent = Agent(
    model=GEMINI_MODEL,
    name="arxiv_research_agent",
    description=(
        "Sub-agent that can call a custom arXiv search tool "
        "to find recent papers for given research topics."
    ),
    instruction=(
        "Role: academic research assistant specialized in arXiv.\n"
        "Capabilities: access to search_arxiv_with_client(topics, max_results, sort_by) returning structured paper metadata.\n"
        "Workflow:\n"
        "1. Normalize the supplied topics and decide whether to call the tool once or multiple times (e.g., one per topic cluster).\n"
        "2. Choose max_results (default 50, adjust when many topics) and an appropriate sort order (use submittedDate unless directed otherwise), then call the tool.\n"
        "3. Merge the results, de-duplicate by arXiv ID, and organize them by topic or theme.\n"
        "Output: Provide each paper's title, authors, year, arXiv ID, URL, and a 1–2 sentence summary. Conclude with key takeaways or gaps."
    ),
    tools=[search_arxiv_with_client_tool],
    output_key="arxiv_research_result",
)

pip install arxiv

import os
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.agents.parallel_agent import ParallelAgent
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.google_search_tool import google_search

from arxiv_agent import arxiv_research_agent
from email_agent import email_agent

GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")


google_agent = LlmAgent(
    name="GoogleSearchResearchAgent",
    model=GEMINI_MODEL,
    instruction="""You are an AI research assistant specializing in finding latest research using web search.
1. Invoke the Google Search tool with a focused query before drafting your answer.
2. Ground every statement in the retrieved results and synthesize them into one concise (1–2 sentence) summary.
3. Output only that summary; include citations and extra commentary.
""",
    description="Researches using Google.",
    tools=[google_search],
    # Store result in state for the merger agent
    output_key="google_research_result",
)

parallel_research_agent = ParallelAgent(
    name="ParallelWebResearchAgent",
    sub_agents=[google_agent, arxiv_research_agent],
    description="Runs multiple research agents in parallel to gather information.",
)

merger_agent = LlmAgent(
    name="MergeSynthesisAgent",
    model=GEMINI_MODEL,
    instruction="""You are an AI assistant merging research outputs into a structured report.

Follow this workflow:
1. Read the Google, arXiv, Duckie, and Hacker News summaries below; do not use external knowledge.
2. Identify the main topics or themes present and organize them into clear headings.
3. Under each heading, integrate the relevant findings, attributing them to their source (Google, arXiv, Duckie, or Hacker News) in-line.
4. Conclude with a brief overall insight section synthesizing cross-source takeaways.
5. Provide a suggested email subject line on the final line in the format: "Suggested Subject: ..."

Input Summaries:
- Google Search Results:
  {google_research_result}

- arXiv Papers:
  {arxiv_research_result}

Output Requirements:
- Use Markdown headings (##) for each topic.
- Write concise paragraphs or bullet points under each heading.
- Ensure all content is grounded exclusively in the input summaries.
- End with the required subject line and nothing else.
""",
    description="Combines research findings from parallel agents into a structured, cited report, strictly grounded on provided inputs.",
)

sequential_pipeline_agent = SequentialAgent(
    name="ResearchAndSynthesisPipeline",
    sub_agents=[parallel_research_agent, merger_agent, email_agent],
    description="Coordinates parallel research and synthesizes the results.",
)


root_agent = sequential_pipeline_agent

adk run .