Uploading a few agents: web_search, arxiv_search.

module_agent_arxiv.py

@@ -0,0 +1,206 @@
+"""
+File: web_app/module_agent_arxiv.py
+Description: an agent with a tool to search arXiv papers.
+Author: Didier Guillevic
+Date: 2025-10-21
+"""
+
+import gradio as gr
+
+from google.adk.agents import Agent
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+from google.adk.tools import google_search
+from google.genai import types
+
+import asyncio
+import uuid
+
+import pydantic
+import arxiv
+
+APP_NAME="arxiv_agent"
+SESSION_ID="1234"
+
+model = "gemini-2.5-flash"
+
+#
+# ===== tool: arXiv search =====
+#
+class PaperArxivInfo(pydantic.BaseModel):
+    paper_id: str
+    title: str
+    authors: list[str]
+    summary: str
+    pdf_url: str
+    published: str
+
+def search_arxiv(query: str, max_results: int=3) -> list[PaperArxivInfo]:
+    """Search for scientific papers on arXiv. By default returns only top 3 results unless otherwise requested by user.
+    
+    Parameters:
+        query: The search query.
+        max_results: Maximum number of results (typically between 1 and 10). Default is 3.
+    """
+    print(f"[DEBUG] Tool search_arxiv received: query={query!r}, max_results={max_results!r}")
+    print(f"Calling search_arxiv with query: {query} and max_results: {max_results}")
+
+    # max_results is set to 3000000000000000 by the agent when not specified, so we cap it here
+    if max_results > 10:
+        max_results = 10
+    print(f"max_results capped to: {max_results}")
+
+    search = arxiv.Search(
+        query=query,
+        max_results=max_results,
+        sort_by=arxiv.SortCriterion.Relevance
+    )
+    results = arxiv.Client().results(search)
+
+    papers = []
+    for result in results:
+        paper_info = PaperArxivInfo(
+            paper_id=result.get_short_id(),
+            title=result.title,
+            authors=[author.name for author in result.authors],
+            summary=result.summary,
+            pdf_url=result.pdf_url,
+            published=result.published.strftime("%Y-%m-%d")
+        )
+        print(f"{paper_info=}")
+        papers.append(paper_info)
+    
+    return papers
+
+#
+# ===== agent =====
+#
+root_agent = Agent(
+    name="arxiv_search_agent",
+    model=model,
+    description=(
+        "Agent to answer questions with the option to call arXiv Search "
+        "if needed for up-to-date scientific paper information."
+    ),
+    instruction=(
+        "I can answer your questions from my own knowledge or by searching the "
+        "arXiv paper repository using the arXiv Search tool. When returning "
+        "results about arXiv paperr, I should provide paper titles, authors, "
+        "summaries, as well as links to the papers. "
+        "Just ask me anything!"
+    ),
+    tools=[search_arxiv]
+)
+
+#
+# ==== Session and Runner =====
+#
+async def setup_session_and_runner(user_id: str):
+    session_service = InMemorySessionService()
+    session = await session_service.create_session(
+        app_name=APP_NAME,
+        user_id=user_id,
+        session_id=SESSION_ID
+    )
+    runner = Runner(
+        agent=root_agent,
+        app_name=APP_NAME,
+        session_service=session_service
+    )
+    return session, runner
+
+#
+# ==== Call Agent Asynchronously =====
+#
+async def call_agent_async(query: str, user_id: str):
+    content = types.Content(role='user', parts=[types.Part(text=query)])
+    session, runner = await setup_session_and_runner(user_id=user_id)
+    events = runner.run_async(
+        user_id=user_id,
+        session_id=SESSION_ID,
+        new_message=content
+    )
+
+    final_response = ""
+
+    async for event in events:
+        if event.is_final_response():
+            final_response = event.content.parts[0].text
+
+    return final_response
+
+#
+# ===== User interface Block =====
+#
+def agent_arxiv_search(query: str, user_id=None):
+    """Calls a language model agent with arXiv Search tool to answer the query.
+
+    Args:
+        query (str): The user query.
+        user_id (str, optional): The user ID for session management. If None, a new ID is generated. Defaults to None.
+
+    Returns:
+        the agent's response (str).
+    """
+    if user_id is None:
+        user_id = str(uuid.uuid4())  # Generate a unique user ID
+
+    response = asyncio.run(call_agent_async(query, user_id))
+    return response, user_id
+
+
+with gr.Blocks() as demo:
+    gr.Markdown(
+        """
+        **Agent with arXiv search tool**: be patient :-) Currently looking into (async) streaming support...
+        """
+    )
+
+    with gr.Row():
+        input_text = gr.Textbox(
+            lines=2,
+            placeholder="Enter your query here...",
+            label="Query",
+            render=True
+        )
+
+    user_id = gr.State(None)
+
+    with gr.Row():
+        submit_button = gr.Button("Submit", variant="primary")
+        clear_button = gr.Button("Clear", variant="secondary")
+    
+    with gr.Row():
+        output_text = gr.Markdown(
+            label="Agent Response",
+            render=True
+        )
+    
+    with gr.Accordion("Examples", open=False):
+        examples = gr.Examples(
+            examples=[
+                ["What is the prime number factorization of 21?",], # no need got Google Search
+                ["Can you search for a few papers on arXiv related to privacy preserving machine learning applied to language models.",],
+                ["Find five papers on arXiv about graph neural networks applied to financial transactions.",],
+            ],
+            inputs=[input_text,],
+            cache_examples=False,
+            label="Click to use an example"
+        )
+
+    # ===== Button Actions =====
+    submit_button.click(
+        fn=agent_arxiv_search,
+        inputs=[input_text, user_id],
+        outputs=[output_text, user_id]
+    )
+    clear_button.click(
+        fn=lambda : ('', None),
+        inputs=None,
+        outputs=[input_text, output_text]
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)
+

module_agent_web_search.py

@@ -0,0 +1,233 @@
+"""
+File: web_app/module_agent_web_search.py
+Description: Gradio module for the Agent Web Search functionality.
+Author: Didier Guillevic
+Date: 2025-10-20
+"""
+
+import gradio as gr
+
+from google.adk.agents import Agent
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+from google.adk.tools import google_search
+from google.genai import types
+
+import asyncio
+import uuid
+
+APP_NAME="google_search_agent"
+SESSION_ID="1234"
+
+model = "gemini-2.5-flash"
+
+#
+# ===== agent =====
+#
+root_agent = Agent(
+    name="basic_search_agent",
+    model=model,
+    description=(
+        "Agent to answer questions with the option to call Google Search "
+        "if needed for up-to-date information."
+    ),
+    instruction=(
+        "I can answer your questions from my own knowledge or by searching the "
+        "web using Google Search. Just ask me anything!"
+    ),
+    # google_search: pre-built tool allows agent to perform Google searches.
+    tools=[google_search]
+)
+
+#
+# ===== Session and Runner =====
+#
+async def setup_session_and_runner(user_id: str):
+    session_service = InMemorySessionService()
+    session = await session_service.create_session(
+        app_name=APP_NAME,
+        user_id=user_id,
+        session_id=SESSION_ID
+    )
+    runner = Runner(
+        agent=root_agent,
+        app_name=APP_NAME,
+        session_service=session_service
+    )
+    return session, runner
+
+
+#
+# ===== Call Agent Asynchronously =====
+#
+async def call_agent_async(query: str, user_id: str):
+    content = types.Content(role='user', parts=[types.Part(text=query)])
+    session, runner = await setup_session_and_runner(user_id=user_id)
+    events = runner.run_async(
+        user_id=user_id,
+        session_id=SESSION_ID,
+        new_message=content
+    )
+
+    final_response = ""
+    rendered_content = ""
+
+    async for event in events:
+        if event.is_final_response():
+            final_response = event.content.parts[0].text
+
+            # Check if the event has grounding metadata and rendered content
+            if (
+                event.grounding_metadata and
+                event.grounding_metadata.search_entry_point and
+                event.grounding_metadata.search_entry_point.rendered_content
+            ):
+                rendered_content = event.grounding_metadata.search_entry_point.rendered_content
+            else:
+                rendered_content = None
+
+    return final_response, rendered_content
+
+
+#
+# ===== Call Agent Asynchronously with Streaming =====
+#
+async def call_agent_streaming(query: str, user_id: str):
+    content = types.Content(role='user', parts=[types.Part(text=query)])
+    session, runner = await setup_session_and_runner(user_id=user_id)
+    events = runner.run_async(
+        user_id=user_id,
+        session_id=SESSION_ID,
+        new_message=content
+    )
+
+    accumulated_response = ""
+    rendered_content = None # Initialize to None
+
+    async for event in events:
+        # Check for intermediate text parts to stream
+        if event.content and event.content.parts and event.content.parts[0].text:
+            # Accumulate and yield the new text
+            new_text = event.content.parts[0].text
+            accumulated_response += new_text
+            yield accumulated_response, None, user_id # Yield the current text and empty grounding
+
+        # When the final response event is received, capture the grounding content
+        if event.is_final_response():
+            # The final response text should already be in accumulated_response from earlier yields,
+            # but we can ensure it's fully captured here.
+            # accumulated_response = event.content.parts[0].text # The final text
+
+            # Capture the rendered_content from grounding_metadata
+            if (
+                event.grounding_metadata and
+                event.grounding_metadata.search_entry_point and
+                event.grounding_metadata.search_entry_point.rendered_content
+            ):
+                rendered_content = event.grounding_metadata.search_entry_point.rendered_content
+            
+            # After the final response, yield one last time with the accumulated text AND the grounding content
+            # This final yield updates the grounding block.
+            yield accumulated_response, rendered_content, user_id 
+
+    # If the grounding content wasn't in the final event (e.g., if no search was performed), 
+    # make sure to yield the final accumulated text.
+    if rendered_content is None:
+        yield accumulated_response, None, user_id
+
+
+#
+# ===== User interface Block =====
+#
+def agent_web_search(query: str, user_id=None):
+    """Calls a language model agent with Google Search tool to answer the query.
+
+    Args:
+        query (str): The user query.
+        user_id (str, optional): The user ID for session management. If None, a new ID is generated. Defaults to None.
+
+    Returns:
+        tuple: A tuple containing the agent's response (str), rendered grounding content (str or None), and user_id (str).
+    """
+    if user_id is None:
+        user_id = str(uuid.uuid4())  # Generate a unique user ID
+
+    response, rendered_content = asyncio.run(call_agent_async(query, user_id))
+    return response, rendered_content, user_id
+
+
+async def agent_web_search_streaming(query: str, current_user_id: str | None):
+    # If the user ID state is None (first run), generate a new one
+    if current_user_id is None:
+        user_id = str(uuid.uuid4())
+    else:
+        user_id = current_user_id
+
+    # The user_id is passed as part of the yield from the generator
+    # but we need to ensure the Gradio state is updated initially for the generator to use the correct ID.
+    
+    # Gradio handles the asynchronous generator return and streams the output to the UI.
+    return call_agent_streaming(query, user_id)
+
+
+with gr.Blocks() as demo:
+    gr.Markdown(
+        """
+        **Agent with Google Search tool**: be patient :-) Currently looking into (async) streaming support...
+        """
+    )
+
+    with gr.Row():
+        input_text = gr.Textbox(
+            lines=2,
+            placeholder="Enter your query here...",
+            label="Query",
+            render=True
+        )
+
+    user_id = gr.State(None)
+
+    with gr.Row():
+        submit_button = gr.Button("Submit", variant="primary")
+        clear_button = gr.Button("Clear", variant="secondary")
+    
+    with gr.Row():
+        output_text = gr.Markdown(
+            label="Agent Response",
+            render=True
+        )
+    
+    with gr.Row():
+        grounding = gr.HTML(
+            label="Grounding Content",
+            render=True
+        )
+    
+    with gr.Accordion("Examples", open=False):
+        examples = gr.Examples(
+            examples=[
+                ["What is the prime number factorization of 15?",], # no need got Google Search
+                ["Who won the Nobel Peace Prize in 2025?",],
+                ["What is the weather like tomorrow in Montreal, Canada?",],
+                ["What are the latest news about Graph Neural Networks?",],
+            ],
+            inputs=[input_text,],
+            cache_examples=False,
+            label="Click to use an example"
+        )
+
+    # ===== Button Actions =====
+    submit_button.click(
+        fn=agent_web_search,
+        inputs=[input_text, user_id],
+        outputs=[output_text, grounding, user_id]
+    )
+    clear_button.click(
+        fn=lambda : ('', '', None),
+        inputs=None,
+        outputs=[input_text, output_text, grounding]
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)