Agentic Design Patterns

The Goal Setting and Monitoring pattern involves an iterative loop where an agent generates an output, a reviewer evaluates that output against specific goals, and the agent refines the output based on feedback until the goals are met or a maximum number of iterations is reached.

In this implementation:

1. Generate: generate_prompt constructs a prompt containing the use case, goals, previous code, and feedback.
2. Execute: The LLM generates code.
3. Review: get_code_feedback uses an LLM to critique the code against the provided goals.
4. Monitor: goals_met uses an LLM to determine if the feedback indicates the goals have been satisfied (returning True or False).
5. Iterate: If goals are not met, the code and feedback are fed back into the next iteration.

# Core loop logic for the pattern
for i in range(max_iterations):
    # 1. Generate
    prompt = generate_prompt(use_case, goals, previous_code, feedback)
    code_response = llm.invoke(prompt)
    code = clean_code_block(code_response.content)

    # 2. Review
    feedback = get_code_feedback(code, goals)
    feedback_text = feedback.content.strip()

    # 3. Monitor
    if goals_met(feedback_text, goals):
        break

    # 4. Prepare for next iteration
    previous_code = code

To enable multi-agent collaboration, you can wrap a specialized LlmAgent inside an AgentTool. This allows a parent agent to treat a sub-agent as a standard tool.

When a parent agent calls an AgentTool, it passes its query to the sub-agent using the input parameter. The sub-agent then executes its own reasoning and tool-calling logic before returning a result to the parent.

Workflow:

1. Parent Agent decides on a task and calls the AgentTool.
2. AgentTool invokes the Sub-Agent using the provided input.
3. Sub-Agent performs its specific reasoning and calls its own internal tools.
4. Sub-Agent returns the final result to the **AgentTool`.
5. Parent Agent receives the result from the tool call.

from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool

# 1. Define the specialized sub-agent
image_generator_agent = LlmAgent(
    name="ImageGen",
    model="gemini-2.0-flash",
    description="Generates an image based on a detailed text prompt.",
    instruction="You are an image generation specialist...",
    tools=[generate_image]
)

# 2. Wrap the sub-agent in an AgentTool so a parent can use it
image_tool = agent_tool.AgentTool(
    agent=image_generator_agent,
    description="Use this tool to generate an image. The input should be a descriptive prompt."
)

# 3. Provide the AgentTool to the parent agent
artist_agent = LlmAgent(
    name="Artist",
    model="gemini-2.0-flash",
    instruction="You are a creative artist...",
    tools=[image_tool]
)

To interact with an agent, use a Runner in conjunction with a SessionService (such as InMemorySessionService).

1. Initialize Session Service: Create an instance of InMemorySessionService.
2. Create a Session: Use session_service.create_session with app_name, user_id, and session_id.
3. Initialize Runner: Create a Runner passing the agent, app_name, and session_service.
4. Run Query: Call runner.run() with the user_id, session_id, and a new_message object (constructed using google.genai.types.Content).
5. Handle Events: The run() method returns an event stream. Iterate through these events and check event.is_final_response() to extract the final text response from event.content.parts[0].text.

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

# Setup
session_service = InMemorySessionService()
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)

# Execution
query = "what's the latest ai news?"
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

for event in events:
    if event.is_final_response():
        print("Agent Response: ", event.content.parts[0].text)

The Google ADK provides high-level abstractions for building agents with built-in capabilities.

Search Agent

Use ADKAgent to create an agent that uses the google_search tool. You must manage sessions using InMemorySessionService and run the agent via a Runner.

Code Execution Agent

Use LlmAgent with a BuiltInCodeExecutor to create an agent capable of writing and running Python code. This is useful for mathematical or data processing tasks.

Implementation Pattern

1. Define Agent: Specify model, instruction, and tools (or code_executor).
2. Setup Session: Create an InMemorySessionService and a session via session_service.create_session.
3. Run via Runner: Use runner.run (sync) or runner.run_async (async) to send messages.
4. Handle Events: Iterate through the events returned by the runner to capture the final response or debug intermediate steps like executable_code or code_execution_result.

from google.adk.agents import Agent as ADKAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.code_executors import BuiltInCodeExecutor
from google.adk.tools import google_search
from google.genai import types

# Example: Code Execution Agent
code_agent = LlmAgent(
   name="calculator_agent",
   model="gemini-2.0-flash",
   code_executor=BuiltInCodeExecutor(),
   instruction="Write and execute Python code to calculate results."
)

async def call_agent_async(query):
   session_service = InMemorySessionService()
   await session_service.create_session(app_name="calc", user_id="u1", session_id="s1")
   runner = Runner(agent=code_agent, app_name="calc", session_service=session_service)
   
   content = types.Content(role='user', parts=[types.Part(text=query)])
   async for event in runner.run_async(user_id="u1", session_id="s1", new_message=content):
       if event.is_final_response():
           print(event.content.parts[0].text)

To create specialized logic within an agentic workflow, define a custom agent as a complete, self-describing class by inheriting from BaseAgent. You must implement the _run_async_impl method, which accepts an InvocationContext and returns an AsyncGenerator of Event objects.

Inside _run_async_impl, you can access the shared session state via context.session.state. To control the flow of a LoopAgent, you can yield events with specific EventActions:

• To terminate a loop: Yield an Event with actions=EventActions(escalate=True).
• To continue a loop: Yield a standard Event with content describing the current state.

from google.adk.agents import BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator

class ConditionChecker(BaseAgent):
    name: str = "ConditionChecker"
    description: str = "Checks if a process is complete and signals the loop to stop."

    async def _run_async_impl(
        self, context: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        status = context.session.state.get("status", "pending")
        is_done = (status == "completed")

        if is_done:
            # Escalate to terminate the loop
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            # Yield a simple event to continue the loop
            yield Event(author=self.name, content="Condition not met, continuing loop.")

To build a tool-calling agent in LangChain, follow these steps:

1. Initialize the LLM: Use a model with function/tool calling capabilities (e.g., gemini-2.0-flash).
2. Define Tools: Use the @langchain_tool decorator on a function. The function's docstring is crucial as it provides the description the LLM uses to decide when to call the tool.
3. Create the Agent: Use create_tool_calling_agent. This requires a ChatPromptTemplate that includes a "placeholder": "{agent_scratchpad}" to track internal reasoning steps.
4. Execute: Wrap the agent in an AgentExecutor to handle the runtime loop of calling tools and processing results.

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 1. Setup LLM
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)

# 2. Define Tool
@langchain_tool
def search_information(query: str) -> str:
   """Provides factual information on a given topic."""
   return "Simulated result"

tools = [search_information]

# 3. Create Agent
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, agent_prompt)

# 4. Setup Executor
agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

# Run
await agent_executor.ainvoke({"input": "What is the capital of France?"})

To interact with an agent in an asynchronous environment, use the Runner class combined with an InMemorySessionService. The runner.run_async method returns an async iterator that yields events during the agent's execution lifecycle.

Key steps:

1. Initialize an InMemorySessionService.
2. Create a session using session_service.create_session.
3. Initialize a Runner with the agent and session service.
4. Iterate over runner.run_async using async for.

When processing events, check event.is_final_response() to identify the final output. Within the final response parts, you can inspect part.executable_code for the code generated by the agent and part.code_execution_result for the outcome and output of that code.

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

# Setup
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=code_agent, app_name=APP_NAME, session_service=session_service)

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

async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
    if event.content and event.content.parts and event.is_final_response():
        for part in event.content.parts:
            if part.executable_code:
                print(f"Code: {part.executable_code.code}")
            elif part.code_execution_result:
                print(f"Result: {part.code_execution_result.outcome} - Output: {part.code_execution_result.output}")
            elif part.text:
                print(f"Text: {part.text}")