Python Research Agent Using Google ADK
The Google Agent Development Kit (ADK) is an open-source Python framework for building multi-agent AI systems. In this chapter we build a deep research agent that plans investigations, executes web searches, evaluates its own findings, and produces a professionally cited report, all orchestrated by cooperating LLM-powered agents running on the Gemini model family.
Overview of Google ADK
The ADK provides composable agent primitives, LlmAgent, SequentialAgent, LoopAgent, and BaseAgent, that let you assemble complex workflows from simple, single-purpose agents. Each agent has its own instruction prompt, optional tools (such as Google Search), and structured output schemas enforced via Pydantic models. Agents communicate through shared session state, and the framework handles the event loop, tool dispatch, and callback lifecycle automatically. This design makes it straightforward to build systems where one agent plans, another researches, a third evaluates quality, and a fourth composes the final output.
Research Agent
The example lives in source-code/a_google_adk_example. It needs Python 3.12 or newer and the google-adk package, which pulls in its own copy of the Gemini SDK:
1 cd source-code/a_google_adk_example
2 uv sync
3 uv run python deep-search.py
I rewrote Google’s full research agent web app example program, simplifying it as a command line utility.
The following listing shows the complete deep search agent. It defines seven specialized agents wired together in a sequential pipeline with an inner refinement loop. The interactive_planner root agent receives a research topic from the user, delegates plan creation to the plan_generator, and upon approval hands off to the research_pipeline. Inside the pipeline, the section_researcher executes Google searches and synthesizes findings, the research_evaluator grades coverage quality, and the enhanced_search_executor fills any gaps, looping up to three times until the evaluator passes. Finally, the report_composer writes a Markdown report with inline source citations.
1 # Derived from: https://github.com/google/adk-samples/tree/main/python/agents/deep-search
2
3 import re
4 import datetime
5 from typing import Literal, AsyncGenerator
6
7 from pydantic import BaseModel, Field
8 from google.genai import types as genai_types
9
10 from google.adk.agents import BaseAgent, LlmAgent, LoopAgent, SequentialAgent
11 from google.adk.agents.callback_context import CallbackContext
12 from google.adk.agents.invocation_context import InvocationContext
13 from google.adk.apps.app import App
14 from google.adk.events import Event, EventActions
15 from google.adk.planners import BuiltInPlanner
16 from google.adk.tools import google_search
17 from google.adk.tools.agent_tool import AgentTool
18
19 # Google ships new Flash models several times a year. See the model naming
20 # note in python_auth_setup.md when this model ID stops working.
21 MODEL_NAME = "gemini-3.8-flash"
22
23 # Note: GOOGLE_API_KEY should be in your environment
24
25
26 # --- Structured Outputs ---
27 class SearchQuery(BaseModel):
28 search_query: str = Field(description="A specific, targeted query for web search.")
29
30 class Feedback(BaseModel):
31 grade: Literal["pass", "fail"]
32 comment: str
33 follow_up_queries: list[SearchQuery] | None = Field(default=None)
34
35 # --- Callbacks ---
36 def collect_sources(callback_context: CallbackContext) -> None:
37 """Aggregates sources from grounding metadata into state."""
38 session, state = callback_context._invocation_context.session, callback_context.state
39 url_map, sources = state.get("url_to_short_id", {}), state.get("sources", {})
40 next_id = len(url_map) + 1
41
42 for event in session.events:
43 if not (md := event.grounding_metadata): continue
44
45 # Map URLs to short IDs (src-1, src-2)
46 chunk_map = {}
47 for idx, chunk in enumerate(md.grounding_chunks or []):
48 if not chunk.web: continue
49 url = chunk.web.uri
50 if url not in url_map:
51 short_id = f"src-{next_id}"
52 url_map[url] = short_id
53 sources[short_id] = {"title": chunk.web.title or chunk.web.domain, "url": url}
54 next_id += 1
55 chunk_map[idx] = url_map[url]
56
57 state["url_to_short_id"] = url_map
58 state["sources"] = sources
59
60 def replace_citations(callback_context: CallbackContext) -> genai_types.Content:
61 """Converts <cite source='src-1'/> tags to Markdown links."""
62 text = callback_context.state.get("final_cited_report", "")
63 sources = callback_context.state.get("sources", {})
64
65 def replacer(match):
66 sid = match.group(1)
67 info = sources.get(sid)
68 return f" [{info['title']}]({info['url']})" if info else ""
69
70 # Replace tags and fix spacing
71 text = re.sub(r'<cite\s+source\s*=\s*["\']?(src-\d+)["\']?\s*/>', replacer, text)
72 text = re.sub(r"\s+([.,;:])", r"\1", text)
73 return genai_types.Content(parts=[genai_types.Part(text=text)])
74
75 # --- Agents ---
76
77 # 1. Plan Generator: Creates the initial strategy
78 plan_generator = LlmAgent(
79 model=MODEL_NAME,
80 name="plan_generator",
81 tools=[google_search],
82 instruction=f"""
83 Create a 5-step research plan.
84 Prefix every step with either:
85 - **`[RESEARCH]`**: For information gathering.
86 - **`[DELIVERABLE]`**: For synthesis/output creation.
87
88 Start with 5 `[RESEARCH]` goals. If these imply a specific output (like a table), add a `[DELIVERABLE][IMPLIED]` step immediately after.
89 If refining a plan based on feedback, mark changes with `[MODIFIED]` or `[NEW]`.
90 Only use search if strictly necessary to clarify ambiguous topics.
91 Date: {datetime.datetime.now().strftime("%Y-%m-%d")}
92 """
93 )
94
95 # 2. Section Planner: Outlines the report structure
96 section_planner = LlmAgent(
97 model=MODEL_NAME,
98 name="section_planner",
99 output_key="report_sections",
100 instruction="""
101 Using the 'research_plan', design a Markdown outline (4-6 sections) for the final report.
102 Do not include a References section.
103 Format: # Section Name \n Brief overview...
104 """
105 )
106
107 # 3. Researcher: The heavy lifter (Search -> Synthesize)
108 section_researcher = LlmAgent(
109 model=MODEL_NAME,
110 name="section_researcher",
111 tools=[google_search],
112 output_key="section_research_findings",
113 after_agent_callback=collect_sources,
114 planner=BuiltInPlanner(thinking_config=genai_types.ThinkingConfig(include_thoughts=True)),
115 instruction="""
116 Execute the `research_plan` in two strict phases:
117
118 **Phase 1: Research**
119 Process all `[RESEARCH]` goals first. For each, generate 4-5 search queries, execute them, and summarize findings. Store these summaries internally.
120
121 **Phase 2: Synthesis**
122 Once Phase 1 is complete, process `[DELIVERABLE]` goals.
123 Use the stored summaries to build the requested artifacts (tables, reports, etc).
124 Do NOT search during this phase.
125
126 Final output must include all research summaries and deliverable artifacts.
127 """
128 )
129
130 # 4. Evaluator: Checks quality
131 research_evaluator = LlmAgent(
132 model=MODEL_NAME,
133 name="research_evaluator",
134 output_key="research_evaluation",
135 output_schema=Feedback,
136 instruction="""
137 Evaluate 'section_research_findings'.
138 Pass if coverage is comprehensive. Fail if there are gaps.
139 If Fail, provide 'follow_up_queries' to fix the gaps.
140 """
141 )
142
143 # 5. Search Executor: Fixes gaps found by Evaluator
144 enhanced_search_executor = LlmAgent(
145 model=MODEL_NAME,
146 name="enhanced_search_executor",
147 tools=[google_search],
148 output_key="section_research_findings", # Merges results back
149 after_agent_callback=collect_sources,
150 instruction="""
151 You are fixing a failed research attempt.
152 1. Execute all 'follow_up_queries' from the evaluation.
153 2. Synthesize new findings and merge them into 'section_research_findings'.
154 """
155 )
156
157 # 6. Escalation Checker: Breaks the loop if Passed
158 class EscalationChecker(BaseAgent):
159 async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
160 result = ctx.session.state.get("research_evaluation", {})
161 if result.get("grade") == "pass":
162 yield Event(author=self.name, actions=EventActions(escalate=True))
163 else:
164 yield Event(author=self.name)
165
166 # 7. Composer: Writes final report with citations
167 report_composer = LlmAgent(
168 model=MODEL_NAME,
169 name="report_composer",
170 output_key="final_cited_report",
171 after_agent_callback=replace_citations,
172 instruction="""
173 Write a professional report using 'section_research_findings' and 'report_sections'.
174 **CITATIONS:** You MUST cite sources using this format: `<cite source="src-ID" />`.
175 Do not create a bibliography; use inline citations only.
176 """
177 )
178
179 # --- Pipelines ---
180
181 research_pipeline = SequentialAgent(
182 name="research_pipeline",
183 description="Executes plan, refines via loop, writes report.",
184 sub_agents=[
185 section_planner,
186 section_researcher,
187 LoopAgent(
188 name="refinement_loop",
189 max_iterations=3,
190 sub_agents=[research_evaluator, EscalationChecker(name="checker"), enhanced_search_executor],
191 ),
192 report_composer,
193 ],
194 )
195
196 # The Root Agent: Interfaces with the user
197 interactive_planner = LlmAgent(
198 name="interactive_planner",
199 model=MODEL_NAME,
200 output_key="research_plan",
201 tools=[AgentTool(plan_generator)], # Uses the generator as a tool
202 sub_agents=[research_pipeline], # Delegates to pipeline upon approval
203 instruction=f"""
204 You are a research assistant.
205 1. Receive user topic.
206 2. Call `plan_generator` to create a plan.
207 3. Show plan to user.
208 4. If user requests changes, call `plan_generator` again.
209 5. If user agrees, delegate to `research_pipeline`.
210 """
211 )
212
213 # --- Application ---
214 app = App(root_agent=interactive_planner, name="DeepSearchApp")
215
216 if __name__ == "__main__":
217 import asyncio
218 import sys
219 from google.adk.runners import InMemoryRunner
220
221 async def main():
222 print("\n--- Deep Search Agent Initialized ---")
223 topic = input("Enter the research topic: ")
224 print(f"Starting research for topic: {topic}\n")
225
226 runner = InMemoryRunner(app=app)
227
228 user_id = "cli_user"
229 session_id = "cli_session"
230
231 # Ensure session exists
232 await runner.session_service.create_session(
233 app_name=app.name,
234 user_id=user_id,
235 session_id=session_id
236 )
237
238 current_input = topic
239
240 while True:
241 try:
242 message = genai_types.Content(parts=[genai_types.Part(text=current_input)])
243 print("\n--- Agent Response ---")
244
245 async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=message):
246 # Debug printing
247 print(f"DEBUG: Event: {type(event)}")
248 if hasattr(event, 'content') and event.content:
249 print(f"DEBUG: Content parts: {len(event.content.parts)}")
250 for p in event.content.parts:
251 print(f"DEBUG: Part type: {type(p)}")
252 if p.function_call:
253 print(f"DEBUG: Function call: {p.function_call.name}")
254
255 if hasattr(event, 'content') and event.content and event.content.parts:
256 for part in event.content.parts:
257 if part.text:
258 print(part.text, end="", flush=True)
259
260 print("\n----------------------")
261
262 current_input = input("\n(Enter to continue, or type feedback/instruction. Type 'quit' to exit)\n> ")
263 if current_input.lower() in ["quit", "exit"]:
264 break
265 if not current_input.strip():
266 current_input = "proceed"
267
268 except KeyboardInterrupt:
269 print("\nExiting...")
270 break
271 except Exception as e:
272 print(f"\nError: {e}")
273 import traceback
274 traceback.print_exc()
275 break
276
277 asyncio.run(main())
The deep search agent demonstrates several patterns that are broadly useful when building agentic AI systems: breaking a complex task into focused sub-agents, using structured Pydantic outputs to enforce data contracts between agents, implementing self-improving loops with quality evaluation, and tracking provenance through source citation callbacks. These same patterns can be adapted for other workflows such as competitive analysis, literature review, market research, or any task where iterative search and synthesis produces better results than a single LLM call.
Keeping This Example Working
Agent frameworks move quickly, and this chapter depends on two of them: google-adk and the Gemini SDK that it pulls in. Two habits keep the example running:
-
Check that the code still loads before you run a session. Change into
source-code/a_google_adk_exampleand run:1 uv sync 2 GOOGLE_API_KEY=placeholder uv run python -c " 3 import importlib.util 4 spec = importlib.util.spec_from_file_location('ds', 'deep-search.py') 5 m = importlib.util.module_from_spec(spec) 6 spec.loader.exec_module(m) 7 print('root agent:', m.app.root_agent.name) 8 "
This builds every agent and prints the root agent name without making a model call. It catches a removed class or a changed import in a few seconds, which is far cheaper than discovering it mid research session.
Treat the model name as data.
MODEL_NAMEis one constant at the top of the file. When Google ships a newer Flash model, change that line. The model naming note in the authentication chapter explains stable and preview names.
The source-code/ directory has a make verify target that runs a check like this one for every chapter, and a GitHub Actions workflow runs it on each push. If you copy these examples into your own project, copy that habit too.