Exploring Google’s Agent Development Kit (ADK)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define a simple tool
def get_weather(city: str) -> str:
    """Gets the current weather for a city.

    Args:
        city: The name of the city to get weather for.

    Returns:
        A string with the weather description.
    """
    return f"The weather in {city} is sunny."

# Create the agent
agent = LlmAgent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
    description="A helpful assistant that can check weather."
)

# Set up runner and session for execution
APP_NAME = "weather_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent using the runner
def run_query(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    # Run the agent with the runner
    events = runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    )

    # Process events to get the final response
    for event in events:
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example of usage
response = run_query("What's the weather like in Berlin?")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tools as functions
def extract_data(input_text: str) -> str:
    """Extracts raw data from input.

    Args:
        input_text: The text to extract data from.

    Returns:
        The extracted raw data.
    """
    return input_text.upper()

def clean_data(data: str) -> str:
    """Cleans the provided data.

    Args:
        data: The data to clean.

    Returns:
        The cleaned data.
    """
    return data.strip()

# Create individual agents for each step
extract_agent = LlmAgent(
    name="extract_agent",
    model="gemini-2.0-flash",
    tools=[extract_data],
    instruction=(
        "You are a data extraction agent. "
        "When given the user message as `input_text`, "
        "call the Python function `extract_data(input_text)` and return *only* its output."
    ),
    description="Extracts data from input",
    output_key="raw_data"
)

clean_agent = LlmAgent(
    name="clean_agent",
    model="gemini-2.0-flash",
    tools=[clean_data],
    instruction=(
        "You are a data cleaning agent. "
        "Take the extracted data in `raw_data`, "
        "call the Python function `clean_data(raw_data)`, and return *only* the cleaned data."
    ),
    description="Cleans extracted data"
)

# Create sequential agent with the proper sub-agents
sequential_agent = SequentialAgent(
    name="sequential_pipeline",
    sub_agents=[extract_agent, clean_agent],
    description="Runs extract_agent then clean_agent, in order."
)

# Set up runner and session for execution
APP_NAME = "sequential_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=sequential_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the sequential agent using the runner
def run_sequential_agent(input_text):
    # Create content from user input
    content = types.Content(
        role="user",
        parts=[types.Part(text=input_text)]
    )

    # Run the agent with the runner
    final_response = "No response received."
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            final_response = event.content.parts[0].text

    return final_response

# Example usage
result = run_sequential_agent("start")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
from google.adk.agents import ParallelAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define some example tools
def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    return f"The weather in {city} is sunny."

def get_news(topic: str) -> str:
    """Gets the latest news on a topic."""
    return f"Latest news about {topic}: Everything is great!"

# Create individual agents for different tasks
weather_agent = LlmAgent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
    instruction=(
        "Extract the city name from the user query, call get_weather(city), "
        "and return only that result."
    ),
    description="Provides weather information",
    output_key = "weather_info"
)

news_agent = LlmAgent(
    name="news_agent",
    model="gemini-2.0-flash",
    tools=[get_news],
    instruction=(
        "Extract the topic from the user query, call get_news(topic), "
        "and return only that result."
    ),
    description="Provides news updates",
    output_key="news_info"
)

# Create a parallel agent with proper agents as sub-agents
parallel_agent = ParallelAgent(
    name="parallel_fetcher",
    sub_agents=[weather_agent, news_agent],
    description="Fetch weather & news at the same time"
)

# Set up runner and session for execution
APP_NAME = "parallel_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=parallel_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the parallel agent using the runner
def run_parallel_agent(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    replies = []

    # Run the agent with the runner
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.content:
            replies.append(event.content.parts[0].text)
    return replies

# Example usage
result = run_parallel_agent("Berlin")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event, EventActions

# Define tools
def guess_number(input_text: str) -> str:
    """Makes a guess at the target number.

    Args:
        input_text: Context for the guess.

    Returns:
        A string containing the guessed number.
    """
    # This is just a mockup - would have actual logic in a real implementation
    return "Is it 42?"

# Create an agent for guessing
guess_agent = LlmAgent(
    name="guesser",
    model="gemini-2.0-flash",
    description="Makes a guess at the target number.",
    instruction=(
        "You are a number-guessing agent. On each turn, "
        "call `guess_number(input_text)` and return *only* its output."
    ),
    tools=[guess_number],
    output_key="last_response"
)

# Create a custom checker agent that determines when to stop looping
class CheckerAgent(BaseAgent):
    """Agent that checks if the guessed number is correct."""

    def __init__(self, name: str):
        super().__init__(name=name)

    async def _run_async_impl(self, context):
        # pull the last guess out of state
        last = context.session.state.get("last_response", "")
        # keep looping until we actually saw “42”
        found = "42" in last

        # “continue” vs “stop” is your protocol
        verdict = "stop" if found else "continue"

        # ALWAYS supply an EventActions instance (cannot be None)
        actions = EventActions(escalate=found)

        yield Event(
            author=self.name,
            content=types.Content(
                role="assistant",
                parts=[types.Part(text=verdict)]
            ),
            actions=actions
        )

# Create the checker agent
checker_agent = CheckerAgent(name="checker")

# Create loop agent with proper configuration
loop_agent = LoopAgent(
    name="guessing_loop",
    sub_agents=[guess_agent, checker_agent],
    max_iterations=5,
    description="Repeatedly guesses a number until correct or max iterations reached",
)

# Set up runner and session for execution
APP_NAME = "loop_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=loop_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the loop agent using the runner
def run_loop_agent(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    # Run the agent with the runner
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example usage
result = run_loop_agent("Start guessing")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from google.adk.agents import LlmAgent
from google.adk.tools.agent_tool import AgentTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Assuming we have a pre-existing agent
sub_agent = LlmAgent(
    name="specialized_agent",
    model="gemini-2.0-flash",
    description="A specialized agent with specific capabilities",
    instruction=(
        "You are the specialized agent. Take the user’s request string "
        "and return exactly: “Specialized result: <their request>”."
    )
)

# Create an agent tool from the sub-agent
nested_tool = AgentTool(agent=sub_agent)

# Use this tool in another agent
super_agent = LlmAgent(
    name="supervisor",
    model="gemini-2.0-flash",
    tools=[nested_tool],
    output_key="delegated_response",
    description="Delegates the user’s request to a specialist and returns the result.",
    instruction=(
        "You are the supervisor. When the user gives you a request, "
        "call `specialized_tool(request)` and return *only* what that tool returns."
    ))

# Set up runner and session for execution
APP_NAME = "delegation_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=super_agent,
    app_name=APP_NAME,
    session_service=session_service
)

def run_supervisor(query: str) -> str:
    msg = types.Content(role="user", parts=[types.Part(text=query)])
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=msg):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."

# Example usage
response = run_supervisor("Please transform this text")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

def greet_user(name: str) -> str:
    """Greets the user by name.

    Args:
        name: The name of the user to greet.

    Returns:
        A greeting message.
    """
    return f"Hello, {name}!"

# Create the agent with the function as a tool
greeting_agent = LlmAgent(
    name="greeter",
    model="gemini-2.0-flash",
    tools=[greet_user],
    description="Agent that can greet users"
)

# Set up runner and session
APP_NAME = "greeting_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=greeting_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent
def run_greeting_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    for event in runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

response = run_greeting_agent("Say hello to Alice")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.genai import types

# Tool with context access
def process_document(document_name: str, analysis_query: str, tool_context: ToolContext) -> dict:
    """Analyzes a document using context from memory.

    Args:
        document_name: Name of the document to analyze.
        analysis_query: The specific analysis to perform.
        tool_context: Access to state and other context variables.

    Returns:
        A dictionary with analysis results or error information.
    """
    # Access session state
    previous_queries = tool_context.state.get("previous_queries", [])

    # Update state with the new query
    tool_context.actions.state_delta = {
        "previous_queries": previous_queries + [analysis_query]
    }

    # Actual document processing would go here
    return {
        "status": "success",
        "analysis": f"Analysis of '{document_name}' regarding '{analysis_query}'"
    }

# Create the agent with the tool
document_agent = LlmAgent(
    name="document_analyzer",
    model="gemini-2.0-flash",
    output_key="analysis_result",
    tools=[process_document],
    description="Analyzes a document and records each query in history.",
    instruction=(
        "You are a document analysis agent. When the user asks "
        "'Analyze <document_name> for <analysis_query>', extract both parts, "
        "call `process_document(document_name, analysis_query, tool_context)`, "
        "and return *only* the resulting JSON dictionary."
    )
)

# Set up runner and session
APP_NAME = "document_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=document_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent
def run_document_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example usage
response = run_document_agent("Analyze report.pdf for sales trends")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Basic tool
def get_data(source: str) -> str:
    """Gets raw data from a specified source.

    Args:
        source: The name or URL of the data source.

    Returns:
        The raw data as a string.
    """
    return f"Raw data from {source}: 42, 17, 23, 8, 15"

# Tool that uses the first tool
def analyze_data(source: str) -> str:
    """Analyzes data from a specified source.

    Args:
        source: The name or URL of the data source.

    Returns:
        Analysis of the data.
    """
    # Get the data first
    raw_data = get_data(source)

    # Parse and analyze it
    numbers = [int(n.strip()) for n in raw_data.split(":", 1)[1].split(",")]
    total = sum(numbers)
    average = total / len(numbers)

    return f"Analysis of {source}:\n- Total: {total}\n- Average: {average:.2f}"

# Create the agent with both tools
data_agent = LlmAgent(
    name="data_analyzer",
    model="gemini-2.0-flash",
    tools=[analyze_data],
    output_key="analysis_summary",
    description="Fetches raw data and returns its analysis.",
    instruction=(
        "You are the data analyzer. When the user says "
        "'Analyze data from <source>', extract the <source> string, "
        "call `analyze_data(source)`, and return *only* its output."
    )
)

# Set up runner and session
APP_NAME = "data_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=data_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent
def run_data_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example usage
response = run_data_agent("Analyze data from database_alpha")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tool functions
def extract(input_text: str) -> str:
    """Extracts data from input."""
    return "Extracted: " + input_text

def clean(data: str) -> str:
    """Cleans and processes the data."""
    return data.strip()

def summarize(data: str) -> str:
    """Summarizes the provided data."""
    return f"Summary of: {data}"

# Create individual agents with direct function references
extract_agent = LlmAgent(
    name="extractor",
    model="gemini-2.0-flash",
    tools=[extract],
    output_key="extracted_data",
    description="Extracts raw data from the user’s input.",
    instruction=(
        "You are the extractor. Given the user’s message as `input_text`, "
        "call `extract(input_text)` and return *only* its output."
    )
)

clean_agent = LlmAgent(
    name="cleaner",
    model="gemini-2.0-flash",
    tools=[clean],
    output_key="cleaned_data",
    description="Cleans the extracted data.",
    instruction=(
        "You are the cleaner. Take the extracted data in `extracted_data`, "
        "call `clean(extracted_data)`, and return *only* the cleaned result."
    )
)

summarize_agent = LlmAgent(
    name="summarizer",
    model="gemini-2.0-flash",
    tools=[summarize],
    output_key="final_summary",
    description="Summarizes the cleaned data.",
    instruction=(
        "You are the summarizer. Take the cleaned data in `cleaned_data`, "
        "call `summarize(cleaned_data)`, and return *only* the summary."
    )
)

# Create a sequential pipeline
pipeline = SequentialAgent(
    name="data_pipeline",
    sub_agents=[extract_agent, clean_agent, summarize_agent],
    description="Processes data in three steps: extract, clean, summarize"
)

# For ADK tools compatibility, the root agent must be named `root_agent`
root_agent = pipeline

# Set up runner and session for execution
APP_NAME = "pipeline_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=root_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Function to run the pipeline
def run_pipeline(input_text):
    # Create content from user input
    content = types.Content(
        role="user",
        parts=[types.Part(text=input_text)]
    )

    # Run the agent with the runner
    final_response = "No response received."

    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        # Check for intermediate state changes (debugging)
        if event.actions and event.actions.state_delta:
            print(f"State update: {event.actions.state_delta}")

        # Get final response
        if event.is_final_response():
            final_response = event.content.parts[0].text

    return final_response

# Example usage
result = run_pipeline("This is some sample input data.")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tool functions for different stages
def extract_text(document_path: str) -> str:
    """Extracts text from a document file."""
    return f"This is sample content for processing from {document_path}."

def analyze_sentiment(text: str) -> dict:
    """Analyzes sentiment of the provided text."""
    positive = ["good", "great", "excellent"]
    negative = ["bad", "poor", "terrible"]
    lc = text.lower()
    pos = sum(w in lc for w in positive)
    neg = sum(w in lc for w in negative)
    tone = "positive" if pos > neg else "negative" if neg > pos else "neutral"
    return {"sentiment": tone, "positive_score": pos, "negative_score": neg}

def generate_summary(text: str, sentiment_data: dict) -> str:
    """Generates a summary based on text and sentiment data."""
    word_count = len(text.split())
    tone = sentiment_data["sentiment"]
    return f"Summary: {word_count} words, overall tone is {tone}."

# Create agents for each step
extract_agent = LlmAgent(
    name="extractor",
    model="gemini-2.0-flash",
    tools=[extract_text],
    output_key="extracted_text",
    description="Extracts text from the given document path.",
    instruction=(
        "You are the extractor. The user says 'Process this document: <path>'. "
        "Extract the <path> part, call `extract_text(path)`, and return only the extracted text."
    )
)

analyze_agent = LlmAgent(
    name="analyzer",
    model="gemini-2.0-flash",
    tools=[analyze_sentiment],
    description="Analyzes sentiment in text",
    output_key="sentiment_data",
    instruction=(
        "You are the analyzer. Take the text in `extracted_text`, "
        "call `analyze_sentiment(extracted_text)`, and return only the resulting JSON."
    )
)

summary_agent = LlmAgent(
    name="summarizer",
    model="gemini-2.0-flash",
    tools=[generate_summary],
    output_key="final_summary",
    description="Summarizes text with sentiment data.",
    instruction=(
        "You are the summarizer. Given `extracted_text` and `sentiment_data`, "
        "call `generate_summary(extracted_text, sentiment_data)` and return only the summary string."
    )
)

# Create the sequential pipeline
document_pipeline = SequentialAgent(
    name="document_processor",
    sub_agents=[extract_agent, analyze_agent, summary_agent],
    description="Extracts text, runs sentiment analysis, then summarizes."
)

# Set up runner and session
APP_NAME = "document_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=document_pipeline,
    app_name=APP_NAME,
    session_service=session_service
)

def run_document_pipeline(document_path: str) -> str:
    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Process this document: {document_path}")]
    )
    final_summary = None

    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # capture the summarizer's final output
        if event.is_final_response() and event.content and event.content.parts:
            final_summary = event.content.parts[0].text
        # keep looping until the generator naturally ends

    return (final_summary or "No response received.").strip()

# Example usage
result = run_document_pipeline("quarterly_report.pdf")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from google.adk.agents import ParallelAgent, LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define research tools for different sources
def search_news(topic: str) -> str:
    """Searches news sources for information on a topic."""
    return f"News results for {topic}: Latest developments include XYZ..."

def search_academic(topic: str) -> str:
    """Searches academic databases for information on a topic."""
    return f"Academic results for {topic}: Recent papers discuss ABC..."

def search_social(topic: str) -> str:
    """Searches social media for information on a topic."""
    return f"Social media trends for {topic}: People are discussing DEF..."

# Create agents for each research source
news_agent = LlmAgent(
    name="news_researcher",
    model="gemini-2.0-flash",
    tools=[search_news],
    output_key="news_results",
    description="Researches news sources",
    instruction=(
        "You are the news researcher. Extract the topic from the user query, "
        "call `search_news(topic)`, and return only the raw string."
    )
)

academic_agent = LlmAgent(
    name="academic_researcher",
    model="gemini-2.0-flash",
    tools=[search_academic],
    output_key="academic_results",
    description="Researches academic sources",
    instruction=(
        "You are the academic researcher. Extract the topic, "
        "call `search_academic(topic)`, and return only the raw string."
    )
)

social_agent = LlmAgent(
    name="social_researcher",
    model="gemini-2.0-flash",
    tools=[search_social],
    output_key="social_results",
    description="Researches social media trends",
    instruction=(
        "You are the social media researcher. Extract the topic, "
        "call `search_social(topic)`, and return only the raw string."
    )
)

# Create parallel research agent
parallel_research = ParallelAgent(
    name="parallel_researcher",
    sub_agents=[news_agent, academic_agent, social_agent],
    description="Fetches news, academic, and social results in parallel"
)

# Create a merger agent to combine results
def merge_research(news: str, academic: str, social: str) -> str:
    """Merges research results from multiple sources."""
    return f"""Combined Research Report:

News Insights: {news}

Academic Findings: {academic}

Social Trends: {social}

This comprehensive view provides a well-rounded perspective on the topic.
"""

merger_agent = LlmAgent(
    name="research_merger",
    model="gemini-2.0-flash",
    tools=[merge_research],
    output_key="merged_report",
    description="Merges research from multiple sources",
    instruction=(
        "You have three pieces of state: `news_results`, `academic_results`, "
        "and `social_results`. Call `merge_research(news_results, academic_results, social_results)` "
        "and return *only* the resulting combined report string."
    )
)

# Create a sequential pipeline that does parallel research then merges
research_pipeline = SequentialAgent(
    name="research_pipeline",
    sub_agents=[parallel_research, merger_agent],
    description="Runs parallel research and then merges the outputs"
)

# Set up runner and session
APP_NAME = "research_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=research_pipeline,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the research pipeline
def run_research(topic: str) -> str:
    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Research this topic: {topic}")]
    )

    merged_report = None

    # Fully consume the event stream
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # Only capture the merger agent's final response
        if (
            event.is_final_response()
            and event.author == "research_merger"
            and event.content and event.content.parts
        ):
            merged_report = event.content.parts[0].text

    return (merged_report or "No response received.").strip()

# Example usage
result = run_research("quantum computing")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event, EventActions

# Define tools for content creation and critique
def generate_draft(topic: str) -> str:
    """Creates an initial draft on a topic."""
    return f"Initial draft about {topic}: This is a basic overview of the subject matter..."

def improve_draft(draft: str, feedback: str) -> str:
    """Improves a draft based on feedback."""
    # In a real implementation, this would make specific improvements
    return f"Improved draft based on feedback: {draft}\n\nAddressed issues: {feedback}"

# Create agents
writer_agent = LlmAgent(
    name="writer",
    model="gemini-2.0-flash",
    tools=[generate_draft, improve_draft],
    output_key="current_draft",
    description="Writes and refines content drafts",
    instruction=(
        "You are the writer. On the first iteration, extract the topic from the user message "
        "and call `generate_draft(topic)`. On subsequent iterations, take `current_draft` from state "
        "and `feedback` from state, call `improve_draft(current_draft, feedback)`. "
        "Return *only* the new draft string."
    )
)

class CriticAgent(BaseAgent):
    def __init__(self, name: str):
        super().__init__(name=name)

    async def _run_async_impl(self, context):
        state = context.session.state
        draft = state.get("current_draft", "")
        iteration = state.get("iteration", 0)

        # simple “quality” model: +25 points per revision
        quality = min(iteration * 25, 100)

        if quality >= 90:
            feedback = "Draft is excellent—no further changes needed."
            yield Event(
                author=self.name,
                content=types.Content(
                    role="assistant",
                    parts=[types.Part(text=feedback)]
                ),
                actions=EventActions(escalate=True)  # break the loop
            )
        else:
            feedback = f"Quality {quality}/100. Needs more clarity and detail."
            # bump iteration and store feedback
            yield Event(
                author=self.name,
                content=types.Content(
                    role="assistant",
                    parts=[types.Part(text=feedback)]
                ),
                actions=EventActions(state_delta={
                    "feedback": feedback,
                    "iteration": iteration + 1
                })
            )

critic_agent = CriticAgent(name="critic")

# Create loop agent for iterative refinement
content_refiner = LoopAgent(
    name="content_refiner",
    sub_agents=[writer_agent, critic_agent],
    max_iterations=5,
    description="Iteratively refines content until quality threshold is met"
)

# Set up runner and session
APP_NAME = "content_app"
USER_ID = "user_123"
SESSION_ID = "session_456"

# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

# Create runner
runner = Runner(
    agent=content_refiner,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the content refiner
def run_content_refiner(topic: str) -> str:
    # seed the loop
    svc_session = session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    svc_session.state["iteration"] = 0

    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Create high-quality content about: {topic}")]
    )
    final = None

    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # capture only when the writer_agent emits a draft _and_ loop is about to end
        if event.is_final_response() and event.author == "writer":
            final = event.content.parts[0].text

    return (final or "No response received.").strip()

# Example usage
result = run_content_refiner("artificial intelligence ethics")
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import json

# Create a simple agent that may have errors
def calculate_complex(expression: str) -> str:
    """Performs a complex calculation that might fail."""
    try:
        # Deliberately overcomplicated to demonstrate error handling
        parts = expression.split()
        result = eval(" ".join(parts))
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

debug_agent = LlmAgent(
    name="debuggable_calculator",
    model="gemini-2.0-flash",
    tools=[calculate_complex],
    description="Performs calculations with detailed error handling for debugging"
)

# Set up runner and session
APP_NAME = "debug_app"
USER_ID = "debug_user"
SESSION_ID = "debug_session"

# Create session service and runner
session_service = InMemorySessionService()
session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)

runner = Runner(
    agent=debug_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent with verbose debugging
def run_with_debugging(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    print("===== STARTING DEBUG SESSION =====")
    print(f"Input: {query}")
    print("===== EVENT STREAM =====")

    # Process every event in detail
    for i, event in enumerate(runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    )):
        print(f"\nEVENT {i + 1}:")
        print(f"  Author: {event.author}")
        print(f"  ID: {event.id}")

        # Check for content
        if event.content:
            print(f"  Content Role: {event.content.role}")
            for j, part in enumerate(event.content.parts):
                print(f"  Content Part {j + 1}: {type(part).__name__}")
                if hasattr(part, 'text') and part.text:
                    print(f"    Text: {part.text[:100]}..." if len(part.text) > 100 else f"    Text: {part.text}")

        # Check for function calls
        function_calls = event.get_function_calls()
        if function_calls:
            print(f"  Function Calls: {len(function_calls)}")
            for fc in function_calls:
                print(f"    Name: {fc.name}")
                print(f"    Args: {json.dumps(fc.args, indent=2)}")

        # Check for function responses
        function_responses = event.get_function_responses()
        if function_responses:
            print(f"  Function Responses: {len(function_responses)}")
            for fr in function_responses:
                print(f"    Name: {fr.name}")
                print(f"    Response: {fr.response}")

        # Check for state changes
        if event.actions and event.actions.state_delta:
            print(f"  State Delta: {event.actions.state_delta}")

        # Check for final response
        if event.is_final_response():
            print("  [THIS IS FINAL RESPONSE]")

    print("===== DEBUG SESSION COMPLETE =====")

# Example usage
run_with_debugging("Calculate 10 * (5 + 3)")
run_with_debugging("Calculate 10 / (5 - 5)")  # Should trigger an error

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Create a simple agent
def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"

agent = LlmAgent(
    name="simple_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A simple question-answering agent"
)

# Set up runner with InMemorySessionService for local use
session_service = InMemorySessionService()
runner = Runner(
    agent=agent,
    app_name="local_app",
    session_service=session_service
)

# Function to run the agent locally
def run_local_query(query):
    # Create session for this run
    session = session_service.create_session(
        app_name="local_app",
        user_id="local_user",
        session_id="local_session"
    )

    # Create content from query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    # Run the agent and extract the final response
    for event in runner.run(
        user_id="local_user",
        session_id="local_session",
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example usage in a local script
if __name__ == "__main__":
    while True:
        user_input = input("Ask a question (or 'exit' to quit): ")
        if user_input.lower() == 'exit':
            break

        response = run_local_query(user_input)
        print(f"Agent: {response}\n")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uuid
import uvicorn

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# --- Agent setup ----------------------------------------------------------

def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"

agent = LlmAgent(
    name="api_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A question-answering agent exposed as an API"
)

session_service = InMemorySessionService()
runner = Runner(
    agent=agent,
    app_name="api_app",
    session_service=session_service
)

# --- Request/Response models ----------------------------------------------

class QueryRequest(BaseModel):
    query: str
    session_id: str = None
    user_id: str = None

class QueryResponse(BaseModel):
    user_id: str
    session_id: str
    response: str

# --- FastAPI app ----------------------------------------------------------

app = FastAPI(title="ADK Agent API")

@app.post("/query", response_model=QueryResponse)
async def query_agent(req: QueryRequest):
    # generate IDs if missing
    user_id = req.user_id or f"user_{uuid.uuid4().hex[:8]}"
    session_id = req.session_id or f"session_{uuid.uuid4().hex[:8]}"

    # ensure session exists
    if session_service.get_session(app_name="api_app", user_id=user_id, session_id= session_id) is None:
        session_service.create_session(app_name="api_app", user_id=user_id, session_id= session_id)

    # build message
    content = types.Content(
        role="user",
        parts=[types.Part(text=req.query)]
    )

    # stream until final and grab the text
    response_text = None
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=content
    ):
        if event.is_final_response() and event.content and event.content.parts:
            response_text = event.content.parts[0].text

    return QueryResponse(
        user_id=user_id,
        session_id=session_id,
        response=response_text or ""
    )

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(ws: WebSocket, session_id: str):
    await ws.accept()
    # each WS connection gets its own user_id
    user_id = f"ws_user_{uuid.uuid4().hex[:8]}"

    # ensure session exists
    if session_service.get_session(app_name="api_app", user_id=user_id, session_id= session_id is None):
        session_service.create_session(app_name="api_app", user_id=user_id, session_id= session_id)

    # let client know its user_id
    await ws.send_json({"type": "session_init", "user_id": user_id, "session_id": session_id})

    try:
        while True:
            query = await ws.receive_text()
            content = types.Content(
                role="user",
                parts=[types.Part(text=query)]
            )

            # stream all events, pushing each as JSON
            async for event in runner.run_async(
                user_id=user_id,
                session_id=session_id,
                new_message=content
            ):
                payload = {
                    "type": "event",
                    "id": event.id,
                    "author": event.author,
                }
                if event.content and event.content.parts:
                    payload["text"] = event.content.parts[0].text
                if event.is_final_response():
                    payload["is_final"] = True

                await ws.send_json(payload)

    except WebSocketDisconnect:
        # client closed connection
        pass
    except Exception as e:
        # on error, close cleanly
        await ws.send_json({"type": "error", "message": str(e)})
        await ws.close()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from google.adk.agents import LlmAgent
from vertexai.preview.reasoning_engines import AdkApp

# Create an agent
def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"

agent = LlmAgent(
    name="vertex_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A question-answering agent deployed on Vertex AI"
)

# Create Vertex AI app from the agent
app = AdkApp(agent=agent)

# For local testing with Vertex AI integration
for event in app.stream_query(
    user_id="test_user",
    message="What is the meaning of life?"
):
    print(event)

# For deployment to Vertex AI Agent Engine
# app.deploy(project="your-gcp-project", display_name="Question Answerer")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import time
import schedule
import json

# Create an agent that summarizes data
def summarize_data(source: str) -> str:
    """Summarizes data from a source."""
    # In a real implementation, this would fetch and analyze data
    return f"Summary of data from {source}: Key metrics are trending upward."

summary_agent = LlmAgent(
    name="summary_agent",
    model="gemini-2.0-flash",
    tools=[summarize_data],
    description="An agent that summarizes data on a schedule"
)

# Set up runner
session_service = InMemorySessionService()
runner = Runner(
    agent=summary_agent,
    app_name="schedule_app",
    session_service=session_service
)

# Function to run the summary job
def run_summary_job():
    print(f"Running scheduled summary job at {time.strftime('%Y-%m-%d %H:%M:%S')}")

    # Create a unique session for this run
    session_id = f"job_{int(time.time())}"
    session = session_service.create_session(
        app_name="schedule_app",
        user_id="schedule_user",
        session_id=session_id
    )

    # Create content with the job request
    content = types.Content(
        role="user",
        parts=[types.Part(text="Summarize today's data from our analytics database")]
    )

    # Run the agent
    summary = None
    for event in runner.run(
        user_id="schedule_user",
        session_id=session_id,
        new_message=content
    ):
        if event.is_final_response():
            summary = event.content.parts[0].text

    if summary:
        # In a real implementation, you might:
        # - Send an email with the summary
        # - Store it in a database
        # - Trigger an alert if certain conditions are met
        print(f"Summary generated: {summary}")

        # Example: Write to a log file
        with open("summaries.log", "a") as log_file:
            log_entry = {
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "session_id": session_id,
                "summary": summary
            }
            log_file.write(json.dumps(log_entry) + "\n")
    else:
        print("No summary generated.")

# Schedule the job to run daily at 8:00 AM
schedule.every().day.at("08:00").do(run_summary_job)

# Run the scheduler
if __name__ == "__main__":
    print("Starting scheduler...")
    # Run once immediately for testing
    run_summary_job()

    # Then run on schedule
    while True:
        schedule.run_pending()
        time.sleep(60)  # Check every minute

1
2
3
4
5
6
7
8
9
10
11
# Example of setting clear limitations in the agent instruction
medical_info_agent = LlmAgent(
    name="health_info",
    model="gemini-2.0-flash",
    instruction="""You provide general health information only. 
    You are NOT a doctor and cannot diagnose conditions or recommend treatments.
    Always clarify that you're providing general information, not medical advice.
    If asked for medical advice, recommend consulting a healthcare professional.
    """,
    description="Provides general health information with appropriate disclaimers"
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# Example of a tool with built-in safety checks
def send_email(to: str, subject: str, body: str) -> dict:
    """Sends an email to the specified recipient.

    Args:
        to: Email address of the recipient.
        subject: Subject line of the email.
        body: Body text of the email.

    Returns:
        A dictionary with status information.
    """
    # Safety check: Validate email format
    if not re.match(r"[^@]+@[^@]+\.[^@]+", to):
        return {"status": "error", "message": "Invalid email format"}

    # Safety check: Check allowed domains
    allowed_domains = ["mycompany.com", "partner.org"]
    if not any(to.endswith(f"@{domain}") for domain in allowed_domains):
        return {
            "status": "error", 
            "message": f"Can only send to these domains: {', '.join(allowed_domains)}"
        }

    # Safety check: Check for sensitive content
    sensitive_terms = ["password", "ssn", "secret", "confidential"]
    for term in sensitive_terms:
        if term in body.lower():
            return {
                "status": "error", 
                "message": f"Cannot send emails containing sensitive terms: {term}"
            }

    # Actual email sending logic would go here
    # ...

    # Log the action for audit purposes
    logging.info(f"Email sent to {to} with subject: {subject}")

    return {"status": "success", "message": "Email sent successfully"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Tool that requires human approval
def propose_action(action_type: str, details: str) -> dict:
    """Proposes an action that requires human approval.

    Args:
        action_type: The type of action being proposed.
        details: Specific details about the action.

    Returns:
        A dictionary with the proposal status.
    """
    # In a real implementation, this might:
    # - Send the proposal to a queue for human review
    # - Send an email notification
    # - Update a dashboard

    print(f"\n=== ACTION REQUIRING APPROVAL ===")
    print(f"Type: {action_type}")
    print(f"Details: {details}")

    # Simple console-based approval for demo purposes
    approval = input("Approve? (y/n): ")

    if approval.lower() == 'y':
        return {
            "status": "approved",
            "message": "Action approved by human reviewer",
            "action_type": action_type,
            "details": details
        }
    else:
        return {
            "status": "rejected",
            "message": "Action rejected by human reviewer",
            "action_type": action_type,
            "details": details
        }

# Create agent with human-in-the-loop design
human_oversight_agent = LlmAgent(
    name="overseen_agent",
    model="gemini-2.0-flash",
    tools=[propose_action],
    instruction="""You are an agent with human oversight.
    For any significant action, use the propose_action tool to get approval before proceeding.
    Significant actions include:
    - Sending communications to customers
    - Making financial decisions above $100
    - Changing system settings
    - Creating or deleting accounts
    """,
    description="Agent that requires human approval for significant actions"
)

# Set up runner
session_service = InMemorySessionService()
runner = Runner(
    agent=human_oversight_agent,
    app_name="oversight_app",
    session_service=session_service
)

# Function to run the agent
def run_overseen_agent(query):
    # Create a session
    session = session_service.create_session(
        app_name="oversight_app",
        user_id="user_123",
        session_id="session_456"
    )

    # Create content
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    # Run the agent
    for event in runner.run(
        user_id="user_123",
        session_id="session_456",
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

    return "No response received."

# Example usage
result = run_overseen_agent("Please send a promotional email to all customers about our new product.")
print(f"\nFinal result: {result}")