Google Agent Development Kit (ADK): A hands-on tutorial

pip install google-adk weave google-genai

import sys
import base64
import asyncio

# ====== AGENTS CODE (always imported/defined) ======
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.agents.callback_context import CallbackContext
from google.genai.types import Part, UserContent
from google.adk.tools import google_search
from google.adk.runners import InMemoryRunner

import os
os.environ["GOOGLE_CLOUD_PROJECT"] = "your_project"
os.environ["GOOGLE_CLOUD_LOCATION"] = "us-central1"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"


research_agent = LlmAgent(
    name="Researcher",
    model="gemini-2.5-flash",
    instruction="""
If 'greeted' is not set in state, greet the user with "👋 Hello! I'm your assistant. Hope all is well.".
Then, conduct thorough web research on the subject given in the input. 
Gather key facts, recent information, and relevant details about the topic.
Summarize your research findings in a comprehensive but organized manner.
This research will be used by other agents to write content, so be thorough and factual.
""",
    output_key="research",
    tools=[google_search],
)

generator_agent = LlmAgent(
    name="DraftWriter",
    model="gemini-2.5-flash",
    instruction="""
Using the research findings stored in state['research'], write a short, factual paragraph about the subject.
Base your writing on the research provided - don't make up facts.
Be concise and clear. Your output should be plain text only.
""",
    output_key="draft_text",
)

reviewer_agent = LlmAgent(
    name="Critic",
    model="gemini-2.5-flash",
    instruction="""
Analyze the paragraph stored in state['draft_text'].
Also reference the research in state['research'] to check for factual accuracy.
Always give a detailed critique, even if the text is factually correct, comment on style, clarity, possible improvements, or anything that could make it better. If there are factual problems, mention those. Output a short critique, not just 'valid'.
""",
    output_key="critique",
)

revision_agent = LlmAgent(
    name="Rewriter",
    model="gemini-2.5-flash",
    instruction="""
Your goal is to revise the paragraph in state['draft_text'] based on the feedback in state['critique'].
You can also reference the research in state['research'] to ensure accuracy and completeness.
Output only the improved paragraph, rewritten as needed.
""",
    output_key="revised_text",
)

def greet_on_first_message(callback_context: CallbackContext):
    if not callback_context.state.get("greeted"):
        callback_context.state["greeted"] = True
    return None

root_agent = SequentialAgent(
    name="ResearchWriteCritiqueRewrite",
    before_agent_callback=greet_on_first_message, 
    sub_agents=[
        research_agent,
        generator_agent,
        reviewer_agent,
        revision_agent
    ]
)

# =========== ONLY DO THIS FOR CLI MODE ===========
if 'web' not in sys.argv:
    # --- OTEL Weave tracing ---
    WANDB_BASE_URL = "https://trace.wandb.ai"
    PROJECT_ID = "byyoung3/adk_demov2"
    WANDB_API_KEY = "your_api_key"
    
    if not WANDB_API_KEY or not PROJECT_ID:
        raise ValueError("Set WANDB_API_KEY and PROJECT_ID as variables!")

    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.sdk import trace as trace_sdk
    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
    from opentelemetry import trace

    OTEL_EXPORTER_OTLP_ENDPOINT = f"{WANDB_BASE_URL}/otel/v1/traces"
    AUTH = base64.b64encode(f"api:{WANDB_API_KEY}".encode()).decode()
    OTEL_EXPORTER_OTLP_HEADERS = {
        "Authorization": f"Basic {AUTH}",
        "project_id": PROJECT_ID,
    }
    exporter = OTLPSpanExporter(
        endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
        headers=OTEL_EXPORTER_OTLP_HEADERS,
    )
    tracer_provider = trace_sdk.TracerProvider()
    tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(tracer_provider)

    runner = InMemoryRunner(agent=root_agent)

    async def create_session():
        return await runner.session_service.create_session(
            app_name=runner.app_name,
            user_id="terminal_user"
        )
    session = asyncio.run(create_session())

    print("==== Research, Write, Critique, Rewrite Terminal ====")
    print("Type a subject or question to research (or 'quit' to exit):")
    while True:
        try:
            user_input = input('\nEnter subject: ').strip()
        except EOFError:
            break
        if user_input.lower() == 'quit':
            print("Goodbye!")
            break

        content = UserContent(parts=[Part(text=user_input)])

        try:
            for event in runner.run(
                user_id=session.user_id,
                session_id=session.id,
                new_message=content
            ):
                if not event or not getattr(event, "content", None):
                    print("> [ERROR: No event or no content in event!]")
                    continue
                for part in getattr(event.content, "parts", []) or []:
                    if getattr(part, "text", None):
                        print('> ', part.text)
        except Exception as e:
            print(f"ERROR: {e}")

            
    print("Session ended.")


import os
import shutil
import uuid
import asyncio
import PyPDF2
import base64
from flask import Flask, request, render_template_string, jsonify
import asyncio
import os
import base64

from google.adk.agents import Agent, LlmAgent, SequentialAgent
from google.adk.runners import InMemoryRunner
from google.genai import types
from litellm import completion
import base64
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk import trace as trace_sdk
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry import trace

from google import genai
from google.genai import types
from PIL import Image
from io import BytesIO



import os
# switch to Gemini 
os.environ["GOOGLE_CLOUD_PROJECT"] = "your_project"
os.environ["GOOGLE_CLOUD_LOCATION"] = "us-central1"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"


def init_weave_otel_adk(
    wandb_api_key: str,
    wandb_project_id: str,
    base_url: str = "https://trace.wandb.ai"
):
    """
    Initialize OpenTelemetry tracing for Weave + Google ADK.

    Args:
        wandb_api_key (str): Your W&B API key.
        wandb_project_id (str): Your W&B entity/project (e.g. "myteam/myproject").
        base_url (str): The Weave trace API base URL (default is "https://trace.wandb.ai").
    """
    otel_endpoint = f"{base_url}/otel/v1/traces"
    auth = base64.b64encode(f"api:{wandb_api_key}".encode()).decode()
    headers = {
        "Authorization": f"Basic {auth}",
        "project_id": wandb_project_id,
    }
    exporter = OTLPSpanExporter(
        endpoint=otel_endpoint,
        headers=headers,
    )
    tracer_provider = trace_sdk.TracerProvider()
    tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(tracer_provider)

PROJECT_ID = "byyoung3/adk_demo"   # e.g. youruser/yourproject
WANDB_API_KEY = "your_api_key"   # from https://wandb.ai/authorize

init_weave_otel_adk(wandb_api_key=WANDB_API_KEY, wandb_project_id=PROJECT_ID)
import weave; weave.init("adk_demov4")


# ==== Setup directories (cleared every run) ====
PDF_DIR = "pdfs"
IMG_DIR = "imgs"

for dir_path in [PDF_DIR, IMG_DIR]:
    if os.path.exists(dir_path):
        shutil.rmtree(dir_path)
    os.makedirs(dir_path, exist_ok=True)



@weave.op
async def analyze_pdf_claim_tool(tool_context) -> dict:
    print("RUNNING PDF", flush=True)
    pdfs = [os.path.join(PDF_DIR, f) for f in os.listdir(PDF_DIR) if f.lower().endswith('.pdf')]
    if not pdfs:
        return {"status": "no_pdf", "message": "No PDF uploaded."}
    latest = max(pdfs, key=os.path.getmtime)

    try:
        with open(latest, "rb") as f:
            reader = PyPDF2.PdfReader(f)
            all_text = "\n".join(page.extract_text() or '' for page in reader.pages)
        print("all Text: " + all_text, flush=True)
        if not all_text.strip():
            return {"status": "error", "message": "No readable text in PDF."}
    except Exception as e:
        return {"status": "error", "message": f"Failed to read PDF: {e}"}

    prompt = (
        "You are an insurance expert. Review the following insurance claim document and list several questions "
        "an insurance company should ask the claimant to clarify missing, ambiguous, incomplete, or suspicious information.\n\n"
        "Claim Document:\n"
        "-----------------------\n"
        f"{all_text[:8000]}"
        "\n-----------------------\n"
        "Questions to ask the claimant:"
    )

    try:
        client = genai.Client()
        resp = await asyncio.to_thread(
            client.models.generate_content,
            model="gemini-2.5-flash",
            contents=[prompt]
        )
        ans = getattr(resp, "text", None) or "No response."
        print(ans, flush=True)
        return {
            "status": "success",
            "claim_filename": os.path.basename(latest),
            "questions": ans,
        }
    except Exception as e:
        return {"status": "error", "message": f"Error from Gemini: {e}"}



@weave.op
async def analyze_one_image(pil_image, fname, criteria, client):
    try:
        buf = BytesIO()
        pil_image.save(buf, format="PNG")
        img_bytes = buf.getvalue()
    except Exception as e:
        return {
            "filename": fname,
            "error": f"Failed to process image: {e}",
        }

    parts = [
        types.Part.from_bytes(data=img_bytes, mime_type="image/png"),
        criteria
    ]
    try:
        resp = await asyncio.to_thread(
            client.models.generate_content,
            model="gemini-2.5-flash",
            contents=parts
        )
        ans = getattr(resp, "text", None) or "No response."
        return {
            "filename": fname,
            "analysis": ans,
        }
    except Exception as e:
        return {
            "filename": fname,
            "error": f"Error analyzing image: {str(e)}",

        }
    
    
@weave.op
async def analyze_imgs_tool(tool_context) -> dict:
    current_call = weave.get_current_call()
    call_id = getattr(current_call, 'id', None)

    print("RUNNING IMG (parallel)", flush=True)
    criteria = (
        "As an insurance claims photo analyst, verify if this photo meets ALL of the following:\n"
        "1. The image is sharp and clear (no blur, good lighting).\n"
        "2. There are NO photo filters applied, and the image is not cropped to hide context.\n"
        "3. The photo shows surroundings to prove location (for example, the full car, not just a dent).\n"
        "For each point, state YES or NO and why.\n"
        "Then, give a one-line verdict: Would this photo be acceptable for an insurance claim? (Yes/No and reason)."
    )

    # --- Collect all PIL images and associated filenames
    files = sorted([
        f for f in os.listdir(IMG_DIR)
        if f.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp'))
    ])
    if not files:
        return {"status": "no_images", "message": "No images available to analyze."}

    pil_images = []
    for fname in files:
        fpath = os.path.join(IMG_DIR, fname)
        try:
            pil_img = Image.open(fpath).convert("RGB")
            pil_images.append((fname, pil_img))
        except Exception as e:
            print(f"Failed to load image {fname}: {e}", flush=True)

    client = genai.Client()

    # --- Now analyze_one_image takes (PIL image, filename)


    # Gather all results concurrently, passing (PIL image, filename)
    results = await asyncio.gather(
        *(analyze_one_image(pil_img, fname, criteria=criteria, client=client) for fname, pil_img in pil_images)
    )

    print(results, flush=True)
    return {"status": "success", "analyses": results, 'call_id': call_id}

# EXAMPLE: Flask wrapper (for HTTP error codes) for PDF endpoint
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/analyze_pdf_claim', methods=['POST'])
async def analyze_pdf_claim():
    result = await analyze_pdf_claim_tool({})
    # HTTP STATUS CODES
    if result["status"] == "success":
        return jsonify(result), 200
    elif result["status"] == "no_pdf":
        return jsonify(result), 400
    else:
        return jsonify(result), 500

@app.route('/analyze_images', methods=['POST'])
async def analyze_images():
    result = await analyze_imgs_tool({})
    if result["status"] == "success":
        return jsonify(result), 200
    elif result["status"] == "no_images":
        return jsonify(result), 400
    else:
        return jsonify(result), 500


# ==== Individual Agents ====
pdf_agent = Agent(
    model="gemini-2.5-flash",
    name="pdf_claim_analyzer",
    instruction="use the analyze_pdf_claim_tool and generate follow-up questions for unclear or missing information. Just call the tool to start. It will handle the analysis",
    description="Analyzes PDF insurance claim documents.",
    tools=[analyze_pdf_claim_tool],
    output_key="pdf_feedback" 
)

photo_agent = Agent(
    model="gemini-2.5-flash",
    name="photo_claim_analyzer", 
    instruction="1. Read the photos using the analyze_imgs_tool. Next, Analyze insurance claim photos for quality and compliance with evidence standards.",
    description="Analyzes insurance claim photos for acceptability.",
    tools=[analyze_imgs_tool],
    output_key="photo_feedback" 
)

# --- Revision Agent ---
summarizer_agent = LlmAgent(
    name="Rewriter",
    model="gemini-2.5-flash",
    instruction="""
    
Your goal is to combine the reports from state['photo_feedback'] and state['pdf_feedback'].
Create a comprehensive insurance claim analysis report that includes:
1. PDF document analysis and recommended questions
2. Photo quality assessment and acceptability
3. Overall claim assessment and recommendations

Output only the improved combined report, rewritten as needed.
""",
    output_key="data_summary"
)

coordinator_agent = SequentialAgent(
    name="insurance_claim_pipeline",
    sub_agents=[pdf_agent, photo_agent, summarizer_agent]
)

# Create runner and session
runner = InMemoryRunner(agent=coordinator_agent)

# Create session synchronously at startup
async def create_session():
    return await runner.session_service.create_session(
        app_name=runner.app_name, user_id="insurance_user"
    )

session = asyncio.run(create_session())

# ==== FLASK UI ====
app = Flask(__name__)

COMBINED_HTML = """
<!DOCTYPE html>
<html>
<head>
  <title>Insurance Claim Analyzer</title>
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <style>
    body { font-family:sans-serif; background:#f3f5f9;}
    .main { max-width:650px; margin:40px auto; background:white; border-radius:8px; box-shadow:0 3px 12px #ddd; padding: 30px; }
    .upload-section { margin-bottom: 30px; padding: 20px; background:#f8f9fa; border-radius: 6px; }
    input[type=file] { margin-bottom:18px; width: 100%; }
    .result { background:#f3f7fa;margin:20px 0;padding:18px; border-radius:7px; font-size:1.05em;}
    .err { color:red;}
    .imglist img { max-width:80px; margin:4px; border-radius:6px;}
    .analysis-section { margin:15px 0; padding:15px; border-left: 3px solid #4a9fd6; background:#fafbfc; }
    button { padding:10px 22px; margin:5px; border:none; background:#4a9fd6; color:white; border-radius:6px; cursor:pointer;}
    button:disabled { background:#ccc; cursor:not-allowed; }
    .file-info { font-size:0.9em; color:#666; margin:5px 0; }
  </style>
</head>
<body>
  <div class="main">
    <h2>🔍 Insurance Claim Analyzer</h2>
    <p>Upload PDF claim documents and/or photos. The AI will analyze both types of evidence and provide a comprehensive assessment.</p>
    
    <div class="upload-section">
      <h3>📄 PDF Claims</h3>
      <form id="pdfForm" enctype="multipart/form-data" onsubmit="return uploadPdf();">
        <input id="pdfFile" type="file" accept="application/pdf">
        <button id="pdfBtn" type="submit">Upload PDF</button>
      </form>
      <div id="pdfStatus" class="file-info"></div>
    </div>

    <div class="upload-section">
      <h3>📸 Claim Photos</h3>
      <form id="photoForm" enctype="multipart/form-data" onsubmit="return uploadPhotos();">
        <input id="photoFiles" type="file" accept="image/*" multiple>
        <button id="photoBtn" type="submit">Upload Photos</button>
      </form>
      <div id="preview" class="imglist"></div>
      <div id="photoStatus" class="file-info"></div>
    </div>

    <div style="text-align:center; margin:20px 0;">
      <button id="analyzeBtn" onclick="analyzeAll()" style="background:#28a745; font-size:1.1em;">
        🔍 Analyze All Uploaded Files
      </button>
    </div>

    <div id="results"></div>
  </div>

  <script>
    // PDF Upload
    function uploadPdf(){
      let f = document.getElementById('pdfFile');
      let btn = document.getElementById('pdfBtn');
      let status = document.getElementById('pdfStatus');
      if(!f.files.length) return false;
      
      btn.disabled = true;
      status.innerHTML = "<i>Uploading PDF...</i>";
      
      let data = new FormData();
      data.append('claimpdf', f.files[0]);
      
      fetch('/upload_pdf', { method:'POST', body:data })
      .then(r=>r.json())
      .then(obj=>{
        btn.disabled = false;
        if(obj.success){
          status.innerHTML = "✓ PDF uploaded successfully";
          status.style.color = 'green';
        }else{
          status.innerHTML = "✗ " + (obj.error||"Upload failed");
          status.style.color = 'red';
        }
      }).catch(e=>{
        btn.disabled = false;
        status.innerHTML = "✗ Error: "+e;
        status.style.color = 'red';
      });
      return false;
    }

    // Photo Upload  
    document.getElementById('photoFiles').onchange = function(ev){
      let preview = document.getElementById('preview');
      preview.innerHTML = '';
      for(let f of ev.target.files){
        let img = document.createElement('img');
        img.src = URL.createObjectURL(f);
        preview.appendChild(img);
      }
    };

    function uploadPhotos(){
      let f = document.getElementById('photoFiles');
      let btn = document.getElementById('photoBtn');
      let status = document.getElementById('photoStatus');
      if(f.files.length==0) return false;
      
      btn.disabled = true;
      status.innerHTML = "<i>Uploading photos...</i>";
      
      let data = new FormData();
      for(let i=0;i<f.files.length;i++) data.append('photos', f.files[i]);
      
      fetch('/upload_photos', { method:'POST', body:data })
      .then(r=>r.json())
      .then(obj=>{
        btn.disabled = false;
        if(obj.success){
          status.innerHTML = `✓ ${obj.count} photos uploaded successfully`;
          status.style.color = 'green';
        }else{
          status.innerHTML = "✗ " + (obj.error||"Upload failed");
          status.style.color = 'red';
        }
      }).catch(e=>{
        btn.disabled = false;
        status.innerHTML = "✗ Error: "+e;
        status.style.color = 'red';
      });
      return false;
    }

    // Analysis
    function analyzeAll(){
      let btn = document.getElementById('analyzeBtn');
      let results = document.getElementById('results');
      
      btn.disabled = true;
      results.innerHTML = "<div style='text-align:center; padding:20px;'><i>🔄 Analyzing claim data... (this may take up to 30 seconds)</i></div>";
      
      fetch('/analyze')
      .then(r=>r.json())
      .then(obj=>{
        btn.disabled = false;
        if(obj.error){
          results.innerHTML = `<div class='err'>Error: ${obj.error}</div>`;
        } else {
          results.innerHTML = `<div class='analysis-section'><h3>📋 Complete Analysis</h3><div>${obj.analysis}</div></div>`;
        }
      }).catch(e=>{
        btn.disabled = false;
        results.innerHTML = `<div class='err'>Error: ${e}</div>`;
      });
    }
  </script>
</body>
</html>
"""

@app.route('/')
def main_page():
    return render_template_string(COMBINED_HTML)

@app.route('/upload_pdf', methods=['POST'])
def upload_pdf():
    f = request.files.get('claimpdf')
    if not f or not f.filename.lower().endswith('.pdf'):
        return jsonify({"error":"No PDF file uploaded."})
    
    pdfbytes = f.read()
    if len(pdfbytes) > 10*1024*1024:
        return jsonify({"error":"PDF too large (10MB max)."})
    
    fname = uuid.uuid4().hex + ".pdf"
    with open(os.path.join(PDF_DIR, fname), 'wb') as out:
        out.write(pdfbytes)
    
    return jsonify({"success": True})

@app.route('/upload_photos', methods=['POST'])
def upload_photos():
    files = request.files.getlist('photos')
    if not files:
        return jsonify({"error":"No files uploaded."})
    
    count = 0
    for f in files:
        ext = os.path.splitext(f.filename)[-1].lower()
        if ext not in ('.jpg','.jpeg','.png','.gif','.webp'):
            continue
        
        imgbytes = f.read()
        if len(imgbytes) > 10*1024*1024:
            continue
        
        fname = uuid.uuid4().hex + ext
        with open(os.path.join(IMG_DIR, fname), 'wb') as out:
            out.write(imgbytes)
        count += 1
    
    if count==0:
        return jsonify({"error":"No valid image files found."})
    
    return jsonify({"success": True, "count": count})



@app.route('/analyze', methods=['GET'])
def analyze_all():
    global session
    
    try:
        from google.genai.types import Part, UserContent
        
        content = UserContent(parts=[Part(text="Analyze all available insurance claim documents and photos")])
        
        # Collect all text output from sequential agent
        all_text = ""
        for event in runner.run(
            user_id=session.user_id, 
            session_id=session.id, 
            new_message=content
        ):
            for part in event.content.parts:
                if part.text:  # Check if text is not None
                    all_text += part.text + "\n"
        
        if not all_text.strip():
            return jsonify({"error": "No analysis results were generated. Please ensure files are uploaded."})
        
        return jsonify({"analysis": all_text.strip()})
        
    except Exception as e:
        return jsonify({"error": f"Server error: {str(e)}"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)