Email GTM B2B Agent and 9_2, 9_3 of google adk
This commit is contained in:
Madhu
parent
d1a248088a
commit
3b6abb675c
13 changed files with 1090 additions and 11 deletions
|
|
@ -0,0 +1,59 @@
|
||||||
|
### AI Email GTM Outreach Agent
|
||||||
|
|
||||||
|
An end-to-end, multi-agent Streamlit app that automates B2B outreach using GPT-5 and Exa. It discovers relevant companies, finds the right contacts (Founder's Office, GTM/Sales leadership, Partnerships/BD, Product Marketing), researches website + Reddit insights, and drafts tailored emails in your selected style.
|
||||||
|
|
||||||
|
## Features
|
||||||
|
|
||||||
|
- **Multi-agent workflow**:
|
||||||
|
- **Company Finder**: Uses Exa to discover companies matching your targeting and offering.
|
||||||
|
- **Contact Finder**: Finds 2–3 relevant decision makers per company and emails (marks inferred emails clearly if needed).
|
||||||
|
- **Researcher**: Pulls 2–4 interesting insights per company from their website and Reddit to enable genuine personalization.
|
||||||
|
- **Email Writer**: Uses GPT-5 to produce concise, structured outreach emails.
|
||||||
|
|
||||||
|
- **Operator controls**:
|
||||||
|
- **Number of companies** to target (1–10)
|
||||||
|
- **Email style**: Professional, Casual, Cold, or Consultative
|
||||||
|
- Live stage-by-stage progress UI and results with clean section dividers
|
||||||
|
|
||||||
|
- **Security-first**:
|
||||||
|
- API keys entered in the Streamlit sidebar; not hardcoded or committed
|
||||||
|
|
||||||
|
## Requirements
|
||||||
|
|
||||||
|
Install dependencies from `requirements.txt`:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install -r advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
Required environment variables (set via sidebar or your shell):
|
||||||
|
|
||||||
|
- `OPENAI_API_KEY`
|
||||||
|
- `EXA_API_KEY`
|
||||||
|
|
||||||
|
## How to Run
|
||||||
|
|
||||||
|
```bash
|
||||||
|
streamlit run advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py
|
||||||
|
```
|
||||||
|
|
||||||
|
## Usage
|
||||||
|
|
||||||
|
1. Enter your `OPENAI_API_KEY` and `EXA_API_KEY` in the left sidebar.
|
||||||
|
2. Provide targeting description and offering.
|
||||||
|
3. Choose number of companies and an email style.
|
||||||
|
4. Click “Start Outreach”. Watch the stages: Companies → Contacts → Research → Emails.
|
||||||
|
5. Review companies, contacts, research insights, and download or copy suggested emails.
|
||||||
|
|
||||||
|
## Notes
|
||||||
|
|
||||||
|
- The app uses the `gpt-5` model via OpenAI. If unavailable in your account, switch the model in `ai_email_gtm_outreach_agent.py` to one you have access to.
|
||||||
|
- Exa is used for web discovery; ensure your `EXA_API_KEY` is valid.
|
||||||
|
|
||||||
|
## Troubleshooting
|
||||||
|
|
||||||
|
- If the app stalls on a stage, verify your API keys and network connectivity.
|
||||||
|
- If JSON parsing errors occur, rerun the stage; models occasionally add extra text around JSON.
|
||||||
|
- For rate limits, reduce number of companies.
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,355 @@
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
|
import streamlit as st
|
||||||
|
from agno.agent import Agent
|
||||||
|
from agno.memory.v2 import Memory
|
||||||
|
from agno.models.openai import OpenAIChat
|
||||||
|
from agno.tools.exa import ExaTools
|
||||||
|
|
||||||
|
|
||||||
|
def require_env(var_name: str) -> None:
|
||||||
|
if not os.getenv(var_name):
|
||||||
|
print(f"Error: {var_name} not set. export {var_name}=...")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def create_company_finder_agent() -> Agent:
|
||||||
|
exa_tools = ExaTools(category="company")
|
||||||
|
memory = Memory()
|
||||||
|
return Agent(
|
||||||
|
model=OpenAIChat(id="gpt-5"),
|
||||||
|
tools=[exa_tools],
|
||||||
|
memory=memory,
|
||||||
|
add_history_to_messages=True,
|
||||||
|
num_history_responses=6,
|
||||||
|
session_id="gtm_outreach_company_finder",
|
||||||
|
show_tool_calls=True,
|
||||||
|
instructions=[
|
||||||
|
"You are CompanyFinderAgent. Use ExaTools to search the web for companies that match the targeting criteria.",
|
||||||
|
"Return ONLY valid JSON with key 'companies' as a list; respect the requested limit provided in the user prompt.",
|
||||||
|
"Each item must have: name, website, why_fit (1-2 lines).",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def create_contact_finder_agent() -> Agent:
|
||||||
|
exa_tools = ExaTools()
|
||||||
|
memory = Memory()
|
||||||
|
return Agent(
|
||||||
|
model=OpenAIChat(id="gpt-4o"),
|
||||||
|
tools=[exa_tools],
|
||||||
|
memory=memory,
|
||||||
|
add_history_to_messages=True,
|
||||||
|
num_history_responses=6,
|
||||||
|
session_id="gtm_outreach_contact_finder",
|
||||||
|
show_tool_calls=True,
|
||||||
|
instructions=[
|
||||||
|
"You are ContactFinderAgent. Use ExaTools to find 1-2 relevant decision makers per company and their emails if available.",
|
||||||
|
"Prioritize roles from Founder's Office, GTM (Marketing/Growth), Sales leadership, Partnerships/Business Development, and Product Marketing.",
|
||||||
|
"Search queries can include patterns like '<Company> email format', 'contact', 'team', 'leadership', and role titles.",
|
||||||
|
"If direct emails are not found, infer likely email using common formats (e.g., first.last@domain), but mark inferred=true.",
|
||||||
|
"Return ONLY valid JSON with key 'companies' as a list; each has: name, contacts: [{full_name, title, email, inferred}]",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def get_email_style_instruction(style_key: str) -> str:
|
||||||
|
styles = {
|
||||||
|
"Professional": "Style: Professional. Clear, respectful, and businesslike. Short paragraphs; no slang.",
|
||||||
|
"Casual": "Style: Casual. Friendly, approachable, first-name basis. No slang or emojis; keep it human.",
|
||||||
|
"Cold": "Style: Cold email. Strong hook in opening 2 lines, tight value proposition, minimal fluff, strong CTA.",
|
||||||
|
"Consultative": "Style: Consultative. Insight-led, frames observed problems and tailored solution hypotheses; soft CTA.",
|
||||||
|
}
|
||||||
|
return styles.get(style_key, styles["Professional"])
|
||||||
|
|
||||||
|
|
||||||
|
def create_email_writer_agent(style_key: str = "Professional") -> Agent:
|
||||||
|
memory = Memory()
|
||||||
|
style_instruction = get_email_style_instruction(style_key)
|
||||||
|
return Agent(
|
||||||
|
model=OpenAIChat(id="gpt-5"),
|
||||||
|
tools=[],
|
||||||
|
memory=memory,
|
||||||
|
add_history_to_messages=True,
|
||||||
|
num_history_responses=6,
|
||||||
|
session_id="gtm_outreach_email_writer",
|
||||||
|
show_tool_calls=False,
|
||||||
|
instructions=[
|
||||||
|
"You are EmailWriterAgent. Write concise, personalized B2B outreach emails.",
|
||||||
|
style_instruction,
|
||||||
|
"Return ONLY valid JSON with key 'emails' as a list of items: {company, contact, subject, body}.",
|
||||||
|
"Length: 120-160 words. Include 1-2 lines of strong personalization referencing research insights (company website and Reddit findings).",
|
||||||
|
"CTA: suggest a short intro call; include sender company name and calendar link if provided.",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def create_research_agent() -> Agent:
|
||||||
|
"""Agent to gather interesting insights from company websites and Reddit."""
|
||||||
|
exa_tools = ExaTools()
|
||||||
|
memory = Memory()
|
||||||
|
return Agent(
|
||||||
|
model=OpenAIChat(id="gpt-5"),
|
||||||
|
tools=[exa_tools],
|
||||||
|
memory=memory,
|
||||||
|
add_history_to_messages=True,
|
||||||
|
num_history_responses=6,
|
||||||
|
session_id="gtm_outreach_researcher",
|
||||||
|
show_tool_calls=True,
|
||||||
|
instructions=[
|
||||||
|
"You are ResearchAgent. For each company, collect concise, valuable insights from:",
|
||||||
|
"1) Their official website (about, blog, product pages)",
|
||||||
|
"2) Reddit discussions (site:reddit.com mentions)",
|
||||||
|
"Summarize 2-4 interesting, non-generic points per company that a human would bring up in an email to show genuine effort.",
|
||||||
|
"Return ONLY valid JSON with key 'companies' as a list; each has: name, insights: [strings].",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def extract_json_or_raise(text: str) -> Dict[str, Any]:
|
||||||
|
"""Extract JSON from a model response. Assumes the response is pure JSON."""
|
||||||
|
try:
|
||||||
|
return json.loads(text)
|
||||||
|
except Exception as e:
|
||||||
|
# Try to locate a JSON block if extra text snuck in
|
||||||
|
start = text.find("{")
|
||||||
|
end = text.rfind("}")
|
||||||
|
if start != -1 and end != -1 and end > start:
|
||||||
|
candidate = text[start : end + 1]
|
||||||
|
return json.loads(candidate)
|
||||||
|
raise ValueError(f"Failed to parse JSON: {e}\nResponse was:\n{text}")
|
||||||
|
|
||||||
|
|
||||||
|
def run_company_finder(agent: Agent, target_desc: str, offering_desc: str, max_companies: int) -> List[Dict[str, str]]:
|
||||||
|
prompt = (
|
||||||
|
f"Find exactly {max_companies} companies that are a strong B2B fit given the user inputs.\n"
|
||||||
|
f"Targeting: {target_desc}\n"
|
||||||
|
f"Offering: {offering_desc}\n"
|
||||||
|
"For each, provide: name, website, why_fit (1-2 lines)."
|
||||||
|
)
|
||||||
|
resp = agent.run(prompt)
|
||||||
|
data = extract_json_or_raise(str(resp.content))
|
||||||
|
companies = data.get("companies", [])
|
||||||
|
return companies[: max(1, min(max_companies, 10))]
|
||||||
|
|
||||||
|
|
||||||
|
def run_contact_finder(agent: Agent, companies: List[Dict[str, str]], target_desc: str, offering_desc: str) -> List[Dict[str, Any]]:
|
||||||
|
prompt = (
|
||||||
|
"For each company below, find 2-3 relevant decision makers and emails (if available). Ensure at least 2 per company when possible, and cap at 3.\n"
|
||||||
|
"If not available, infer likely email and mark inferred=true.\n"
|
||||||
|
f"Targeting: {target_desc}\nOffering: {offering_desc}\n"
|
||||||
|
f"Companies JSON: {json.dumps(companies, ensure_ascii=False)}\n"
|
||||||
|
"Return JSON: {companies: [{name, contacts: [{full_name, title, email, inferred}]}]}"
|
||||||
|
)
|
||||||
|
resp = agent.run(prompt)
|
||||||
|
data = extract_json_or_raise(str(resp.content))
|
||||||
|
return data.get("companies", [])
|
||||||
|
|
||||||
|
|
||||||
|
def run_research(agent: Agent, companies: List[Dict[str, str]]) -> List[Dict[str, Any]]:
|
||||||
|
prompt = (
|
||||||
|
"For each company, gather 2-4 interesting insights from their website and Reddit that would help personalize outreach.\n"
|
||||||
|
f"Companies JSON: {json.dumps(companies, ensure_ascii=False)}\n"
|
||||||
|
"Return JSON: {companies: [{name, insights: [string, ...]}]}"
|
||||||
|
)
|
||||||
|
resp = agent.run(prompt)
|
||||||
|
data = extract_json_or_raise(str(resp.content))
|
||||||
|
return data.get("companies", [])
|
||||||
|
|
||||||
|
|
||||||
|
def run_email_writer(agent: Agent, contacts_data: List[Dict[str, Any]], research_data: List[Dict[str, Any]], offering_desc: str, sender_name: str, sender_company: str, calendar_link: Optional[str]) -> List[Dict[str, str]]:
|
||||||
|
prompt = (
|
||||||
|
"Write personalized outreach emails for the following contacts.\n"
|
||||||
|
f"Sender: {sender_name} at {sender_company}.\n"
|
||||||
|
f"Offering: {offering_desc}.\n"
|
||||||
|
f"Calendar link: {calendar_link or 'N/A'}.\n"
|
||||||
|
f"Contacts JSON: {json.dumps(contacts_data, ensure_ascii=False)}\n"
|
||||||
|
f"Research JSON: {json.dumps(research_data, ensure_ascii=False)}\n"
|
||||||
|
"Return JSON with key 'emails' as a list of {company, contact, subject, body}."
|
||||||
|
)
|
||||||
|
resp = agent.run(prompt)
|
||||||
|
data = extract_json_or_raise(str(resp.content))
|
||||||
|
return data.get("emails", [])
|
||||||
|
|
||||||
|
|
||||||
|
def run_pipeline(target_desc: str, offering_desc: str, sender_name: str, sender_company: str, calendar_link: Optional[str], num_companies: int):
|
||||||
|
company_agent = create_company_finder_agent()
|
||||||
|
contact_agent = create_contact_finder_agent()
|
||||||
|
research_agent = create_research_agent()
|
||||||
|
|
||||||
|
companies = run_company_finder(company_agent, target_desc, offering_desc, max_companies=num_companies)
|
||||||
|
contacts_data = run_contact_finder(contact_agent, companies, target_desc, offering_desc) if companies else []
|
||||||
|
research_data = run_research(research_agent, companies) if companies else []
|
||||||
|
return {
|
||||||
|
"companies": companies,
|
||||||
|
"contacts": contacts_data,
|
||||||
|
"research": research_data,
|
||||||
|
"emails": [],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
st.set_page_config(page_title="GTM B2B Outreach", layout="wide")
|
||||||
|
|
||||||
|
# Sidebar: API keys
|
||||||
|
st.sidebar.header("API Configuration")
|
||||||
|
openai_key = st.sidebar.text_input("OpenAI API Key", type="password", value=os.getenv("OPENAI_API_KEY", ""))
|
||||||
|
exa_key = st.sidebar.text_input("Exa API Key", type="password", value=os.getenv("EXA_API_KEY", ""))
|
||||||
|
if openai_key:
|
||||||
|
os.environ["OPENAI_API_KEY"] = openai_key
|
||||||
|
if exa_key:
|
||||||
|
os.environ["EXA_API_KEY"] = exa_key
|
||||||
|
|
||||||
|
if not openai_key or not exa_key:
|
||||||
|
st.sidebar.warning("Enter both API keys to enable the app")
|
||||||
|
|
||||||
|
# Inputs
|
||||||
|
st.title("GTM B2B Outreach Multi Agent Team")
|
||||||
|
st.info(
|
||||||
|
"GTM teams often need to reach out for demos and discovery calls, but manual research and personalization is slow. "
|
||||||
|
"This app uses GPT-5 with a multi-agent workflow to find target companies, identify contacts, research genuine insights (website + Reddit), "
|
||||||
|
"and generate tailored outreach emails in your chosen style."
|
||||||
|
)
|
||||||
|
col1, col2 = st.columns(2)
|
||||||
|
with col1:
|
||||||
|
target_desc = st.text_area("Target companies (industry, size, region, tech, etc.)", height=100)
|
||||||
|
offering_desc = st.text_area("Your product/service offering (1-3 sentences)", height=100)
|
||||||
|
with col2:
|
||||||
|
sender_name = st.text_input("Your name", value="Sales Team")
|
||||||
|
sender_company = st.text_input("Your company", value="Our Company")
|
||||||
|
calendar_link = st.text_input("Calendar link (optional)", value="")
|
||||||
|
num_companies = st.number_input("Number of companies", min_value=1, max_value=10, value=5)
|
||||||
|
email_style = st.selectbox(
|
||||||
|
"Email style",
|
||||||
|
options=["Professional", "Casual", "Cold", "Consultative"],
|
||||||
|
index=0,
|
||||||
|
help="Choose the tone/format for the generated emails",
|
||||||
|
)
|
||||||
|
|
||||||
|
if st.button("Start Outreach", type="primary"):
|
||||||
|
# Validate
|
||||||
|
if not openai_key or not exa_key:
|
||||||
|
st.error("Please provide API keys in the sidebar")
|
||||||
|
elif not target_desc or not offering_desc:
|
||||||
|
st.error("Please fill in target companies and offering")
|
||||||
|
else:
|
||||||
|
# Stage-by-stage progress UI
|
||||||
|
progress = st.progress(0)
|
||||||
|
stage_msg = st.empty()
|
||||||
|
details = st.empty()
|
||||||
|
try:
|
||||||
|
# Prepare agents
|
||||||
|
company_agent = create_company_finder_agent()
|
||||||
|
contact_agent = create_contact_finder_agent()
|
||||||
|
research_agent = create_research_agent()
|
||||||
|
email_agent = create_email_writer_agent(email_style)
|
||||||
|
|
||||||
|
# 1. Companies
|
||||||
|
stage_msg.info("1/4 Finding companies...")
|
||||||
|
companies = run_company_finder(
|
||||||
|
company_agent,
|
||||||
|
target_desc.strip(),
|
||||||
|
offering_desc.strip(),
|
||||||
|
max_companies=int(num_companies),
|
||||||
|
)
|
||||||
|
progress.progress(25)
|
||||||
|
details.write(f"Found {len(companies)} companies")
|
||||||
|
|
||||||
|
# 2. Contacts
|
||||||
|
stage_msg.info("2/4 Finding contacts (2–3 per company)...")
|
||||||
|
contacts_data = run_contact_finder(
|
||||||
|
contact_agent,
|
||||||
|
companies,
|
||||||
|
target_desc.strip(),
|
||||||
|
offering_desc.strip(),
|
||||||
|
) if companies else []
|
||||||
|
progress.progress(50)
|
||||||
|
details.write(f"Collected contacts for {len(contacts_data)} companies")
|
||||||
|
|
||||||
|
# 3. Research
|
||||||
|
stage_msg.info("3/4 Researching insights (website + Reddit)...")
|
||||||
|
research_data = run_research(research_agent, companies) if companies else []
|
||||||
|
progress.progress(75)
|
||||||
|
details.write(f"Compiled research for {len(research_data)} companies")
|
||||||
|
|
||||||
|
# 4. Emails
|
||||||
|
stage_msg.info("4/4 Writing personalized emails...")
|
||||||
|
emails = run_email_writer(
|
||||||
|
email_agent,
|
||||||
|
contacts_data,
|
||||||
|
research_data,
|
||||||
|
offering_desc.strip(),
|
||||||
|
sender_name.strip() or "Sales Team",
|
||||||
|
sender_company.strip() or "Our Company",
|
||||||
|
calendar_link.strip() or None,
|
||||||
|
) if contacts_data else []
|
||||||
|
progress.progress(100)
|
||||||
|
details.write(f"Generated {len(emails)} emails")
|
||||||
|
|
||||||
|
st.session_state["gtm_results"] = {
|
||||||
|
"companies": companies,
|
||||||
|
"contacts": contacts_data,
|
||||||
|
"research": research_data,
|
||||||
|
"emails": emails,
|
||||||
|
}
|
||||||
|
stage_msg.success("Completed")
|
||||||
|
except Exception as e:
|
||||||
|
stage_msg.error("Pipeline failed")
|
||||||
|
st.error(f"{e}")
|
||||||
|
|
||||||
|
# Show results if present
|
||||||
|
results = st.session_state.get("gtm_results")
|
||||||
|
if results:
|
||||||
|
companies = results.get("companies", [])
|
||||||
|
contacts = results.get("contacts", [])
|
||||||
|
research = results.get("research", [])
|
||||||
|
emails = results.get("emails", [])
|
||||||
|
|
||||||
|
st.subheader("Top target companies")
|
||||||
|
if companies:
|
||||||
|
for idx, c in enumerate(companies, 1):
|
||||||
|
st.markdown(f"**{idx}. {c.get('name','')}** ")
|
||||||
|
st.write(c.get("website", ""))
|
||||||
|
st.write(c.get("why_fit", ""))
|
||||||
|
else:
|
||||||
|
st.info("No companies found")
|
||||||
|
st.divider()
|
||||||
|
|
||||||
|
st.subheader("Contacts found")
|
||||||
|
if contacts:
|
||||||
|
for c in contacts:
|
||||||
|
st.markdown(f"**{c.get('name','')}**")
|
||||||
|
for p in c.get("contacts", [])[:3]:
|
||||||
|
inferred = " (inferred)" if p.get("inferred") else ""
|
||||||
|
st.write(f"- {p.get('full_name','')} | {p.get('title','')} | {p.get('email','')}{inferred}")
|
||||||
|
else:
|
||||||
|
st.info("No contacts found")
|
||||||
|
st.divider()
|
||||||
|
|
||||||
|
st.subheader("Research insights")
|
||||||
|
if research:
|
||||||
|
for r in research:
|
||||||
|
st.markdown(f"**{r.get('name','')}**")
|
||||||
|
for insight in r.get("insights", [])[:4]:
|
||||||
|
st.write(f"- {insight}")
|
||||||
|
else:
|
||||||
|
st.info("No research insights")
|
||||||
|
st.divider()
|
||||||
|
|
||||||
|
st.subheader("Suggested Outreach Emails")
|
||||||
|
if emails:
|
||||||
|
for i, e in enumerate(emails, 1):
|
||||||
|
with st.expander(f"{i}. {e.get('company','')} → {e.get('contact','')}"):
|
||||||
|
st.write(f"Subject: {e.get('subject','')}")
|
||||||
|
st.text(e.get("body", ""))
|
||||||
|
else:
|
||||||
|
st.info("No emails generated")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,5 @@
|
||||||
|
agno>=0.4.2
|
||||||
|
streamlit>=1.33.0
|
||||||
|
pydantic>=2.7.0
|
||||||
|
openai>=1.30.0
|
||||||
|
exa_py>=1.0.7
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
import os
|
import os
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import inspect
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
from google.adk.agents import LlmAgent, SequentialAgent
|
from google.adk.agents import LlmAgent, SequentialAgent
|
||||||
from google.adk.tools import google_search
|
from google.adk.tools import google_search
|
||||||
|
|
@ -112,19 +113,22 @@ async def analyze_business_intelligence(user_id: str, business_topic: str) -> st
|
||||||
"""Process business intelligence through the sequential pipeline"""
|
"""Process business intelligence through the sequential pipeline"""
|
||||||
session_id = f"bi_session_{user_id}"
|
session_id = f"bi_session_{user_id}"
|
||||||
|
|
||||||
# Create or get session
|
# Support both sync and async session service
|
||||||
session = await session_service.get_session(
|
async def _maybe_await(value):
|
||||||
|
return await value if inspect.isawaitable(value) else value
|
||||||
|
|
||||||
|
session = await _maybe_await(session_service.get_session(
|
||||||
app_name="business_intelligence",
|
app_name="business_intelligence",
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
session_id=session_id
|
session_id=session_id
|
||||||
)
|
))
|
||||||
if not session:
|
if not session:
|
||||||
session = await session_service.create_session(
|
session = await _maybe_await(session_service.create_session(
|
||||||
app_name="business_intelligence",
|
app_name="business_intelligence",
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
session_id=session_id,
|
session_id=session_id,
|
||||||
state={"business_topic": business_topic, "conversation_history": []}
|
state={"business_topic": business_topic, "conversation_history": []}
|
||||||
)
|
))
|
||||||
|
|
||||||
# Create user content
|
# Create user content
|
||||||
user_content = types.Content(
|
user_content = types.Content(
|
||||||
|
|
@ -132,15 +136,22 @@ async def analyze_business_intelligence(user_id: str, business_topic: str) -> st
|
||||||
parts=[types.Part(text=f"Please analyze this business topic: {business_topic}")]
|
parts=[types.Part(text=f"Please analyze this business topic: {business_topic}")]
|
||||||
)
|
)
|
||||||
|
|
||||||
# Run the sequential pipeline
|
# Run the sequential pipeline (support async or sync stream)
|
||||||
response_text = ""
|
response_text = ""
|
||||||
async for event in runner.run_async(
|
stream = runner.run_async(
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
session_id=session_id,
|
session_id=session_id,
|
||||||
new_message=user_content
|
new_message=user_content
|
||||||
):
|
)
|
||||||
if event.is_final_response():
|
if inspect.isasyncgen(stream):
|
||||||
if event.content and event.content.parts:
|
async for event in stream:
|
||||||
response_text = event.content.parts[0].text
|
if event.is_final_response():
|
||||||
|
if event.content and event.content.parts:
|
||||||
|
response_text = event.content.parts[0].text
|
||||||
|
else:
|
||||||
|
for event in stream:
|
||||||
|
if getattr(event, "is_final_response", lambda: False)():
|
||||||
|
if event.content and event.content.parts:
|
||||||
|
response_text = event.content.parts[0].text
|
||||||
|
|
||||||
return response_text
|
return response_text
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,3 @@
|
||||||
|
# If using Gemini via Google AI Studio
|
||||||
|
GOOGLE_GENAI_USE_VERTEXAI=False
|
||||||
|
GOOGLE_API_KEY="your-api-key"
|
||||||
|
|
@ -0,0 +1,87 @@
|
||||||
|
# 🔁 Tutorial 9.2: Loop Agents - Iterative Plan Refiner
|
||||||
|
|
||||||
|
## 🎯 What You'll Learn
|
||||||
|
|
||||||
|
- **Loop Agent Composition**: Execute sub-agents sequentially in a loop
|
||||||
|
- **Stateful Iterations**: Persist counters and flags across iterations
|
||||||
|
- **Termination Conditions**: Stop by reaching a max or when a sub-agent escalates
|
||||||
|
- **Streamlit Web Interface**: Interactive UI to run iterative refinements
|
||||||
|
|
||||||
|
## 🧠 Core Concept: LoopAgent with Condition
|
||||||
|
|
||||||
|
According to the ADK workflow agents documentation, **LoopAgent** repeats a set of sub-agents while sharing the same context/state across iterations. This tutorial demonstrates an **Iterative Plan Refiner** that improves a plan over multiple iterations and stops when a condition is met.
|
||||||
|
|
||||||
|
```
|
||||||
|
Topic → LoopAgent → [Refine Plan] → [Increment Iteration] → [Check Completion]
|
||||||
|
↑ │
|
||||||
|
└──────────────────────────── Repeat until stop ─────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
**Termination**: The loop stops if the optional `max_iterations` is reached, or if any sub-agent returns an `Event` with `escalate=True` in its `EventActions`.
|
||||||
|
|
||||||
|
**Context & State**: The same `InvocationContext` and `session.state` are used across iterations, allowing values like `iteration`, `target_iterations`, and `accepted` to persist and control the loop.
|
||||||
|
|
||||||
|
## 📁 Project Structure
|
||||||
|
|
||||||
|
```
|
||||||
|
9_2_loop agent/
|
||||||
|
├── agent.py # LoopAgent with 3 sub-agents and session-state control
|
||||||
|
├── app.py # Streamlit UI to run the loop refinement
|
||||||
|
└── README.md # This documentation
|
||||||
|
```
|
||||||
|
|
||||||
|
## 🚀 Getting Started
|
||||||
|
|
||||||
|
### 1. Install Dependencies
|
||||||
|
```bash
|
||||||
|
cd "9_2_loop agent"
|
||||||
|
pip install -r ../9_1_sequential_agent/requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Set Up Environment
|
||||||
|
Create a `.env` file with your Google API key (or reuse from the sequential example):
|
||||||
|
```bash
|
||||||
|
echo "GOOGLE_API_KEY=your_ai_studio_key_here" > .env
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3. Run the Streamlit App
|
||||||
|
```bash
|
||||||
|
streamlit run app.py
|
||||||
|
```
|
||||||
|
|
||||||
|
## 🧪 How It Works
|
||||||
|
|
||||||
|
- **plan_refiner (LlmAgent)**: Produces a concise, improved plan each iteration.
|
||||||
|
- **increment_iteration (BaseAgent)**: Increments `session.state['iteration']`.
|
||||||
|
- **check_completion (BaseAgent)**: Escalates (to stop) if `accepted=True` or `iteration >= target_iterations`.
|
||||||
|
|
||||||
|
The `LoopAgent` sequences these sub-agents on every iteration, persisting and updating state until a stop condition is met.
|
||||||
|
|
||||||
|
### Session State Keys
|
||||||
|
- **topic**: The subject being refined.
|
||||||
|
- **iteration**: Current iteration counter.
|
||||||
|
- **target_iterations**: Loop budget before stopping.
|
||||||
|
- **accepted**: When set to `True`, the loop stops immediately.
|
||||||
|
|
||||||
|
## 🧪 Try It
|
||||||
|
- Enter a topic (e.g., "AI-powered customer support platform launch plan").
|
||||||
|
- Set `Target iterations` to 3–5.
|
||||||
|
- Run and observe the final refined plan and run metadata.
|
||||||
|
|
||||||
|
## 🔧 ADK Concepts Demonstrated
|
||||||
|
- **LoopAgent pattern** with sequential sub-agents.
|
||||||
|
- **Session state persistence** across iterations.
|
||||||
|
- **Escalation-based termination** with `EventActions(escalate=True)`.
|
||||||
|
- **Runner + SessionService** execution pattern.
|
||||||
|
|
||||||
|
## 🔎 Troubleshooting
|
||||||
|
- Ensure `GOOGLE_API_KEY` is set in `.env`.
|
||||||
|
- Run from the directory containing `app.py`.
|
||||||
|
- If you previously ran the app, the same session id is reused; changing the topic or target updates state accordingly.
|
||||||
|
|
||||||
|
## 📚 Key Takeaways
|
||||||
|
- **LoopAgent** enables iterative refinement workflows.
|
||||||
|
- **Shared state** allows complex control signals to accumulate across iterations.
|
||||||
|
- **Clean, modular sub-agents** keep the loop logic clear and maintainable.
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,222 @@
|
||||||
|
import os
|
||||||
|
import asyncio
|
||||||
|
import inspect
|
||||||
|
from typing import AsyncGenerator, Dict, Any
|
||||||
|
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
from google.adk.agents import LlmAgent, LoopAgent
|
||||||
|
from google.adk.agents.base_agent import BaseAgent
|
||||||
|
from google.adk.agents.invocation_context import InvocationContext
|
||||||
|
from google.adk.sessions import InMemorySessionService
|
||||||
|
from google.adk.runners import Runner
|
||||||
|
from google.adk.events import Event, EventActions
|
||||||
|
from google.genai import types
|
||||||
|
|
||||||
|
|
||||||
|
# Load environment variables
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# Sub-agent 1: LLM refiner that improves the plan each iteration
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
plan_refiner = LlmAgent(
|
||||||
|
name="plan_refiner",
|
||||||
|
model="gemini-2.5-flash",
|
||||||
|
description="Iteratively refines a brief product/launch plan given topic and prior context",
|
||||||
|
instruction=(
|
||||||
|
"You are an iterative planner. On each turn:\n"
|
||||||
|
"- Improve and tighten the current plan for the topic in session state\n"
|
||||||
|
"- Keep it concise (5-8 bullets) and avoid repeating prior text verbatim\n"
|
||||||
|
"- Incorporate clarity, feasibility, and crisp sequencing\n"
|
||||||
|
"- Assume this output will be refined again in subsequent iterations\n\n"
|
||||||
|
"Output format:\n"
|
||||||
|
"Title line\n"
|
||||||
|
"- Bullet 1\n- Bullet 2\n- Bullet 3 ..."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# Sub-agent 2: Progress tracker increments iteration counter
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
class IncrementIteration(BaseAgent):
|
||||||
|
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
|
||||||
|
current_iteration = int(ctx.session.state.get("iteration", 0)) + 1
|
||||||
|
ctx.session.state["iteration"] = current_iteration
|
||||||
|
yield Event(
|
||||||
|
author=self.name,
|
||||||
|
content=types.Content(
|
||||||
|
role="model",
|
||||||
|
parts=[
|
||||||
|
types.Part(
|
||||||
|
text=f"Iteration advanced to {current_iteration}"
|
||||||
|
)
|
||||||
|
],
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# Sub-agent 3: Completion check with optional early stop
|
||||||
|
# - Stops if iteration >= target_iterations OR session flag 'accepted' is True
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
class CheckCompletion(BaseAgent):
|
||||||
|
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
|
||||||
|
target_iterations = int(ctx.session.state.get("target_iterations", 3))
|
||||||
|
current_iteration = int(ctx.session.state.get("iteration", 0))
|
||||||
|
accepted = bool(ctx.session.state.get("accepted", False))
|
||||||
|
|
||||||
|
reached_limit = current_iteration >= target_iterations
|
||||||
|
should_stop = accepted or reached_limit
|
||||||
|
|
||||||
|
yield Event(
|
||||||
|
author=self.name,
|
||||||
|
actions=EventActions(escalate=should_stop),
|
||||||
|
content=types.Content(
|
||||||
|
role="model",
|
||||||
|
parts=[
|
||||||
|
types.Part(
|
||||||
|
text=(
|
||||||
|
"Stopping criteria met"
|
||||||
|
if should_stop
|
||||||
|
else "Continuing loop"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
],
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
increment_iteration = IncrementIteration(name="increment_iteration")
|
||||||
|
check_completion = CheckCompletion(name="check_completion")
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# LoopAgent: Executes sub-agents sequentially in a loop
|
||||||
|
# - Termination: max_iterations, or CheckCompletion escalates
|
||||||
|
# - Context & State: Same InvocationContext across iterations
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
spec_refinement_loop = LoopAgent(
|
||||||
|
name="spec_refinement_loop",
|
||||||
|
description=(
|
||||||
|
"Iteratively refines a plan using LLM, tracks iterations, and stops when target iterations "
|
||||||
|
"are reached or an 'accepted' flag is set in session state."
|
||||||
|
),
|
||||||
|
max_iterations=10,
|
||||||
|
sub_agents=[
|
||||||
|
plan_refiner,
|
||||||
|
increment_iteration,
|
||||||
|
check_completion,
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# Runner setup
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
session_service = InMemorySessionService()
|
||||||
|
runner = Runner(
|
||||||
|
agent=spec_refinement_loop,
|
||||||
|
app_name="loop_refinement_app",
|
||||||
|
session_service=session_service,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
# Public API: run the loop refinement for a topic
|
||||||
|
# ------------------------------------------------------------
|
||||||
|
async def iterate_spec_until_acceptance(
|
||||||
|
user_id: str, topic: str, target_iterations: int = 3
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""Run the LoopAgent to iteratively refine a plan.
|
||||||
|
|
||||||
|
Returns a dictionary with final plan text and iteration metadata.
|
||||||
|
"""
|
||||||
|
session_id = f"loop_refinement_{user_id}"
|
||||||
|
|
||||||
|
async def _maybe_await(value):
|
||||||
|
return await value if inspect.isawaitable(value) else value
|
||||||
|
|
||||||
|
# Create or get session (support both sync/async services)
|
||||||
|
session = await _maybe_await(session_service.get_session(
|
||||||
|
app_name="loop_refinement_app",
|
||||||
|
user_id=user_id,
|
||||||
|
session_id=session_id,
|
||||||
|
))
|
||||||
|
if not session:
|
||||||
|
session = await _maybe_await(session_service.create_session(
|
||||||
|
app_name="loop_refinement_app",
|
||||||
|
user_id=user_id,
|
||||||
|
session_id=session_id,
|
||||||
|
state={
|
||||||
|
"topic": topic,
|
||||||
|
"iteration": 0,
|
||||||
|
"target_iterations": int(target_iterations),
|
||||||
|
# Optionally, an external process or UI could set this to True to stop early
|
||||||
|
"accepted": False,
|
||||||
|
},
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
# Refresh topic/target if user re-runs on UI
|
||||||
|
if hasattr(session, "state") and isinstance(session.state, dict):
|
||||||
|
session.state["topic"] = topic
|
||||||
|
session.state["target_iterations"] = int(target_iterations)
|
||||||
|
|
||||||
|
# Seed message for LLM
|
||||||
|
user_content = types.Content(
|
||||||
|
role="user",
|
||||||
|
parts=[
|
||||||
|
types.Part(
|
||||||
|
text=(
|
||||||
|
"Topic: "
|
||||||
|
+ topic
|
||||||
|
+ "\nPlease produce or refine a concise plan."
|
||||||
|
)
|
||||||
|
)
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
final_text = ""
|
||||||
|
last_plan_text = ""
|
||||||
|
stream = runner.run_async(user_id=user_id, session_id=session_id, new_message=user_content)
|
||||||
|
# Support both async generators and plain iterables
|
||||||
|
if inspect.isasyncgen(stream):
|
||||||
|
async for event in stream:
|
||||||
|
if event.content and getattr(event.content, "parts", None):
|
||||||
|
for part in event.content.parts:
|
||||||
|
if hasattr(part, "text") and part.text:
|
||||||
|
# Keep last text from plan_refiner preferentially
|
||||||
|
if getattr(event, "author", "") == plan_refiner.name:
|
||||||
|
last_plan_text = part.text
|
||||||
|
if event.is_final_response():
|
||||||
|
final_text = part.text
|
||||||
|
else:
|
||||||
|
for event in stream:
|
||||||
|
if event.content and getattr(event.content, "parts", None):
|
||||||
|
for part in event.content.parts:
|
||||||
|
if hasattr(part, "text") and part.text:
|
||||||
|
if getattr(event, "author", "") == plan_refiner.name:
|
||||||
|
last_plan_text = part.text
|
||||||
|
# final events in sync mode
|
||||||
|
final_text = part.text
|
||||||
|
if event.content and getattr(event.content, "parts", None):
|
||||||
|
for part in event.content.parts:
|
||||||
|
if hasattr(part, "text") and part.text:
|
||||||
|
# Keep last text from plan_refiner preferentially
|
||||||
|
if getattr(event, "author", "") == plan_refiner.name:
|
||||||
|
last_plan_text = part.text
|
||||||
|
if event.is_final_response():
|
||||||
|
final_text = part.text
|
||||||
|
|
||||||
|
current_iteration = int(session.state.get("iteration", 0))
|
||||||
|
reached = current_iteration >= int(session.state.get("target_iterations", 0))
|
||||||
|
accepted = bool(session.state.get("accepted", False))
|
||||||
|
|
||||||
|
return {
|
||||||
|
"final_plan": last_plan_text or final_text,
|
||||||
|
"iterations": current_iteration,
|
||||||
|
"stopped_reason": "accepted" if accepted else ("target_iterations" if reached else "max_iterations_or_other"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,76 @@
|
||||||
|
import streamlit as st
|
||||||
|
import asyncio
|
||||||
|
from agent import iterate_spec_until_acceptance
|
||||||
|
|
||||||
|
|
||||||
|
st.set_page_config(page_title="Loop Agent Demo", page_icon=":repeat:", layout="wide")
|
||||||
|
|
||||||
|
st.title("🔁 Iterative Plan Refiner (Loop Agent)")
|
||||||
|
st.markdown(
|
||||||
|
"""
|
||||||
|
This demo runs a LoopAgent that repeatedly executes sub-agents to iteratively refine a plan.
|
||||||
|
|
||||||
|
Loop characteristics:
|
||||||
|
- Executes its sub-agents sequentially in a loop
|
||||||
|
- Terminates when the session's `accepted` flag is set or after the target iterations
|
||||||
|
- Shares the same session state across iterations, so counters/flags persist
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
user_id = "demo_loop_user"
|
||||||
|
|
||||||
|
st.header("Run an iterative refinement")
|
||||||
|
topic = st.text_area(
|
||||||
|
"Topic",
|
||||||
|
value="AI-powered customer support platform launch plan",
|
||||||
|
height=100,
|
||||||
|
placeholder="What plan/topic should be refined iteratively?",
|
||||||
|
)
|
||||||
|
|
||||||
|
col_a, col_b = st.columns([1, 1])
|
||||||
|
with col_a:
|
||||||
|
target_iterations = st.number_input(
|
||||||
|
"Target iterations (early stop possible)", min_value=1, max_value=20, value=3, step=1
|
||||||
|
)
|
||||||
|
with col_b:
|
||||||
|
st.caption(
|
||||||
|
"Set a reasonable number of iterations. The loop may stop earlier if the session state flag `accepted` becomes True."
|
||||||
|
)
|
||||||
|
|
||||||
|
if st.button("Run Loop Refinement", type="primary"):
|
||||||
|
if topic.strip():
|
||||||
|
st.info("Refining plan in a loop…")
|
||||||
|
with st.spinner("Working…"):
|
||||||
|
try:
|
||||||
|
results = asyncio.run(
|
||||||
|
iterate_spec_until_acceptance(user_id, topic, int(target_iterations))
|
||||||
|
)
|
||||||
|
st.success("Loop finished")
|
||||||
|
|
||||||
|
st.subheader("Final Refined Plan")
|
||||||
|
st.write(results.get("final_plan", ""))
|
||||||
|
|
||||||
|
st.subheader("Run Metadata")
|
||||||
|
st.write({
|
||||||
|
"iterations": results.get("iterations"),
|
||||||
|
"stopped_reason": results.get("stopped_reason"),
|
||||||
|
})
|
||||||
|
except Exception as e:
|
||||||
|
st.error(f"Error: {e}")
|
||||||
|
else:
|
||||||
|
st.error("Please enter a topic")
|
||||||
|
|
||||||
|
with st.sidebar:
|
||||||
|
st.header("How it works")
|
||||||
|
st.markdown(
|
||||||
|
"""
|
||||||
|
- Uses `LoopAgent` with 3 sub-agents:
|
||||||
|
1) `plan_refiner` (LLM) refines the plan
|
||||||
|
2) `increment_iteration` updates the iteration counter in session state
|
||||||
|
3) `check_completion` escalates when done (accepted flag or target reached)
|
||||||
|
- The same `InvocationContext` and session state are reused every iteration
|
||||||
|
- The loop stops if `accepted` is True or the `target_iterations` is reached.
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,3 @@
|
||||||
|
# If using Gemini via Google AI Studio
|
||||||
|
GOOGLE_GENAI_USE_VERTEXAI=False
|
||||||
|
GOOGLE_API_KEY="your-api-key"
|
||||||
|
|
@ -0,0 +1,75 @@
|
||||||
|
# ⚡ Tutorial 9.3: Parallel Agents - Market Snapshot Team
|
||||||
|
|
||||||
|
## 🎯 What You'll Learn
|
||||||
|
|
||||||
|
- **Parallel Agent Composition**: How to orchestrate multiple specialized agents concurrently
|
||||||
|
- **Shared State**: How parallel children write to a common `session.state` safely
|
||||||
|
- **Branching Context**: Invocation branches for clean, isolated tool/memory context
|
||||||
|
- **Streamlit Interface**: A simple UI to run and visualize parallel results
|
||||||
|
|
||||||
|
## 🧠 Core Concept: ParallelAgent with Shared State
|
||||||
|
|
||||||
|
According to the ADK docs, **Parallel Agents** execute their sub-agents concurrently. Each child runs on its own invocation branch but shares the same `session.state`.
|
||||||
|
|
||||||
|
```
|
||||||
|
Topic → ParallelAgent → 3 Sub-agents (Concurrent Execution)
|
||||||
|
↓
|
||||||
|
[Market Trends] + [Competitors] + [Funding News]
|
||||||
|
↓
|
||||||
|
Snapshot in state
|
||||||
|
```
|
||||||
|
|
||||||
|
Each child agent writes results to a distinct key in shared state to avoid overwrites: `market_trends`, `competitors`, `funding_news`.
|
||||||
|
|
||||||
|
## 📁 Project Structure
|
||||||
|
|
||||||
|
```
|
||||||
|
9_3_parallel agent/
|
||||||
|
├── agent.py # Parallel workflow (3 research agents + ParallelAgent)
|
||||||
|
├── app.py # Streamlit UI to run and view snapshot
|
||||||
|
├── requirements.txt # Python dependencies
|
||||||
|
├── README.md # This documentation
|
||||||
|
└── .env.example # Example environment variables
|
||||||
|
```
|
||||||
|
|
||||||
|
## 🚀 Getting Started
|
||||||
|
|
||||||
|
### 1. Install Dependencies
|
||||||
|
```bash
|
||||||
|
cd "9_3_parallel agent"
|
||||||
|
pip install -r requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Set Up Environment
|
||||||
|
Create a `.env` file with your Google API key:
|
||||||
|
```bash
|
||||||
|
echo "GOOGLE_API_KEY=your_ai_studio_key_here" > .env
|
||||||
|
```
|
||||||
|
|
||||||
|
> Get your key from Google AI Studio.
|
||||||
|
|
||||||
|
### 3. Run the Streamlit App
|
||||||
|
```bash
|
||||||
|
streamlit run app.py
|
||||||
|
```
|
||||||
|
|
||||||
|
## 🧪 How It Works
|
||||||
|
|
||||||
|
- `ParallelAgent` executes `market_trends_agent`, `competitor_intel_agent`, and `funding_news_agent` concurrently.
|
||||||
|
- Each child uses web search and writes to a unique `output_key` in `session.state`.
|
||||||
|
- The UI reads `session.state` and displays a 3-column snapshot.
|
||||||
|
|
||||||
|
## 🔧 ADK Concepts Demonstrated
|
||||||
|
|
||||||
|
- ParallelAgent pattern and event interleaving
|
||||||
|
- Shared `session.state` with distinct keys per child
|
||||||
|
- Invocation branches for contextual separation
|
||||||
|
- Runner + Session services for execution
|
||||||
|
|
||||||
|
## 📚 Key Takeaways
|
||||||
|
|
||||||
|
- Parallel fan-out is ideal for independent data gathering
|
||||||
|
- Keep output keys distinct to avoid overwrites in shared state
|
||||||
|
- Combine with a downstream synthesizer agent if you need a single report
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,116 @@
|
||||||
|
from typing import Dict, Any
|
||||||
|
import inspect
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
from google.adk.agents import LlmAgent, ParallelAgent
|
||||||
|
from google.adk.sessions import InMemorySessionService
|
||||||
|
from google.adk.runners import Runner
|
||||||
|
from google.genai import types
|
||||||
|
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
# Child agents write to distinct keys in session.state for UI consumption
|
||||||
|
market_trends_agent = LlmAgent(
|
||||||
|
name="market_trends_agent",
|
||||||
|
model="gemini-2.5-flash",
|
||||||
|
description="Summarizes recent market trends for the topic",
|
||||||
|
instruction=(
|
||||||
|
"Summarize 3-5 recent market trends for the topic in session.state['topic'].\n"
|
||||||
|
"Output a concise markdown list."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
competitor_intel_agent = LlmAgent(
|
||||||
|
name="competitor_intel_agent",
|
||||||
|
model="gemini-2.5-flash",
|
||||||
|
description="Identifies key competitors and positioning",
|
||||||
|
instruction=(
|
||||||
|
"List 3-5 notable competitors for session.state['topic'] and describe their positioning briefly."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
funding_news_agent = LlmAgent(
|
||||||
|
name="funding_news_agent",
|
||||||
|
model="gemini-2.5-flash",
|
||||||
|
description="Reports funding/partnership news",
|
||||||
|
instruction=(
|
||||||
|
"Provide a short digest (bulleted) of recent funding or partnership news related to session.state['topic']."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
# Parallel orchestrator
|
||||||
|
market_snapshot_team = ParallelAgent(
|
||||||
|
name="market_snapshot_team",
|
||||||
|
description="Runs multiple research agents concurrently to produce a market snapshot",
|
||||||
|
sub_agents=[
|
||||||
|
market_trends_agent,
|
||||||
|
competitor_intel_agent,
|
||||||
|
funding_news_agent,
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
# Runner and session service
|
||||||
|
session_service = InMemorySessionService()
|
||||||
|
runner = Runner(agent=market_snapshot_team, app_name="parallel_snapshot_app", session_service=session_service)
|
||||||
|
|
||||||
|
|
||||||
|
async def gather_market_snapshot(user_id: str, topic: str) -> Dict[str, Any]:
|
||||||
|
"""Execute the parallel agents and return combined snapshot text blocks.
|
||||||
|
|
||||||
|
Returns keys: 'market_trends', 'competitors', 'funding_news'.
|
||||||
|
"""
|
||||||
|
session_id = f"parallel_snapshot_{user_id}"
|
||||||
|
|
||||||
|
async def _maybe_await(v):
|
||||||
|
return await v if inspect.isawaitable(v) else v
|
||||||
|
|
||||||
|
session = await _maybe_await(
|
||||||
|
session_service.get_session(
|
||||||
|
app_name="parallel_snapshot_app", user_id=user_id, session_id=session_id
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if not session:
|
||||||
|
session = await _maybe_await(
|
||||||
|
session_service.create_session(
|
||||||
|
app_name="parallel_snapshot_app",
|
||||||
|
user_id=user_id,
|
||||||
|
session_id=session_id,
|
||||||
|
state={"topic": topic},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
if hasattr(session, "state") and isinstance(session.state, dict):
|
||||||
|
session.state["topic"] = topic
|
||||||
|
|
||||||
|
user_content = types.Content(
|
||||||
|
role="user",
|
||||||
|
parts=[types.Part(text=f"Topic: {topic}. Provide a concise snapshot per agent focus.")],
|
||||||
|
)
|
||||||
|
|
||||||
|
# Collect last text emitted per agent
|
||||||
|
last_text_by_agent: Dict[str, str] = {}
|
||||||
|
|
||||||
|
stream = runner.run_async(user_id=user_id, session_id=session_id, new_message=user_content)
|
||||||
|
if inspect.isasyncgen(stream):
|
||||||
|
async for event in stream:
|
||||||
|
if getattr(event, "content", None) and getattr(event.content, "parts", None):
|
||||||
|
for part in event.content.parts:
|
||||||
|
if hasattr(part, "text") and part.text:
|
||||||
|
author = getattr(event, "author", "")
|
||||||
|
if author:
|
||||||
|
last_text_by_agent[author] = part.text
|
||||||
|
else:
|
||||||
|
for event in stream:
|
||||||
|
if getattr(event, "content", None) and getattr(event.content, "parts", None):
|
||||||
|
for part in event.content.parts:
|
||||||
|
if hasattr(part, "text") and part.text:
|
||||||
|
author = getattr(event, "author", "")
|
||||||
|
if author:
|
||||||
|
last_text_by_agent[author] = part.text
|
||||||
|
|
||||||
|
return {
|
||||||
|
"market_trends": last_text_by_agent.get(market_trends_agent.name, ""),
|
||||||
|
"competitors": last_text_by_agent.get(competitor_intel_agent.name, ""),
|
||||||
|
"funding_news": last_text_by_agent.get(funding_news_agent.name, ""),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -0,0 +1,62 @@
|
||||||
|
import streamlit as st
|
||||||
|
import asyncio
|
||||||
|
from agent import market_snapshot_team, gather_market_snapshot
|
||||||
|
|
||||||
|
st.set_page_config(page_title="Parallel Agent Demo", page_icon=":fast_forward:", layout="wide")
|
||||||
|
|
||||||
|
st.title("⚡ Market Snapshot (Parallel Agents)")
|
||||||
|
st.markdown(
|
||||||
|
"""
|
||||||
|
This demo runs multiple research agents in parallel using a ParallelAgent:
|
||||||
|
|
||||||
|
- Market trends analysis
|
||||||
|
- Competitor intelligence
|
||||||
|
- Funding and partnerships news
|
||||||
|
|
||||||
|
Each sub-agent writes its results into a shared session.state under distinct keys. A subsequent step (or this UI) can read the combined snapshot.
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
user_id = "demo_parallel_user"
|
||||||
|
|
||||||
|
st.header("Run a market snapshot")
|
||||||
|
topic = st.text_input(
|
||||||
|
"Research topic",
|
||||||
|
value="AI-powered customer support platforms",
|
||||||
|
placeholder="What market/topic do you want a quick parallel snapshot on?",
|
||||||
|
)
|
||||||
|
|
||||||
|
if st.button("Run Parallel Research", type="primary"):
|
||||||
|
if topic.strip():
|
||||||
|
st.info("Running parallel agents… market trends, competitors, and funding news")
|
||||||
|
with st.spinner("Gathering snapshot…"):
|
||||||
|
try:
|
||||||
|
results = asyncio.run(gather_market_snapshot(user_id, topic))
|
||||||
|
st.success("Snapshot ready")
|
||||||
|
|
||||||
|
col1, col2, col3 = st.columns(3)
|
||||||
|
with col1:
|
||||||
|
st.subheader("Market Trends")
|
||||||
|
st.write(results.get("market_trends", ""))
|
||||||
|
with col2:
|
||||||
|
st.subheader("Competitors")
|
||||||
|
st.write(results.get("competitors", ""))
|
||||||
|
with col3:
|
||||||
|
st.subheader("Funding News")
|
||||||
|
st.write(results.get("funding_news", ""))
|
||||||
|
except Exception as e:
|
||||||
|
st.error(f"Error: {e}")
|
||||||
|
else:
|
||||||
|
st.error("Please enter a topic")
|
||||||
|
|
||||||
|
with st.sidebar:
|
||||||
|
st.header("How it works")
|
||||||
|
st.markdown(
|
||||||
|
"""
|
||||||
|
- Uses `ParallelAgent` to execute sub-agents concurrently
|
||||||
|
- Each child runs on its own invocation branch, but shares the same session.state
|
||||||
|
- Distinct `output_key`s prevent overwrites in the shared state
|
||||||
|
- This pattern is ideal for fan-out data gathering before synthesis
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
|
@ -0,0 +1,5 @@
|
||||||
|
google-adk>=1.9.0
|
||||||
|
streamlit>=1.28.0
|
||||||
|
python-dotenv>=1.1.1
|
||||||
|
pydantic>=2.0.0
|
||||||
|
|
||||||
Loading…
Reference in a new issue