איך יוצרים סוכן לתכנון בדיקות QA באמצעות ADK,‏ MCP ו-Gemini 2.5 Flash עם מצב Thinking

1. מבוא

782957a0266b13f6.png

הפוטנציאל של שימוש ב-AI גנרטיבי ליצירת תוכנית בדיקות נובע מהיכולת שלו לפתור שתיים מהבעיות הכי גדולות בהבטחת איכות מודרנית: מהירות ומקיפות. במחזורי הפיתוח המהירים של Agile ו-DevOps של היום, יצירה ידנית של תוכניות בדיקה מפורטות יוצרת צוואר בקבוק משמעותי ומעכבת את כל תהליך הבדיקה. סוכן מבוסס-AI גנרטיבי יכול לעבד סיפורי משתמשים ודרישות טכניות כדי ליצור תוכנית בדיקות מקיפה תוך דקות, ולא תוך ימים, וכך לוודא שתהליך בקרת האיכות מתנהל בקצב שמתאים לקצב הפיתוח. בנוסף, AI מצטיין בזיהוי תרחישים מורכבים, מקרים חריגים ונתיבים שליליים שאדם עלול לפספס, מה שמוביל לשיפור משמעותי בכיסוי הבדיקות ולצמצום משמעותי של באגים שחומקים אל הייצור.

ב-codelab הזה נסביר איך ליצור סוכן כזה שיכול לאחזר את מסמכי דרישות המוצר מ-Confluence, לתת משוב בונה וגם ליצור תוכנית בדיקות מקיפה שאפשר לייצא לקובץ CSV.

במהלך ה-codelab, תשתמשו בגישה שלב אחר שלב באופן הבא:

  1. הכנת הפרויקט ב-Google Cloud והפעלת כל ממשקי ה-API הנדרשים בו
  2. הגדרת סביבת עבודה לסביבת הקידוד
  3. הכנת שרת MCP מקומי ל-Confluence
  4. מבנה של קוד המקור, ההנחיה והכלים של סוכן ה-ADK לחיבור לשרת ה-MCP
  5. הסבר על השימוש בשירות Artifact והקשר בין כלים
  6. בדיקת הסוכן באמצעות ממשק משתמש מקומי לפיתוח אתרים ב-ADK
  7. ניהול משתני הסביבה והגדרת הקבצים הנדרשים לפריסת האפליקציה ב-Cloud Run
  8. פריסת האפליקציה ב-Cloud Run

סקירה כללית של הארכיטקטורה

af793beb2740233.jpeg

דרישות מוקדמות

  • ניסיון בעבודה עם Python
  • הבנה של ארכיטקטורת full-stack בסיסית באמצעות שירות HTTP

מה תלמדו

  • תכנון של ADK Agent תוך שימוש בכמה מהיכולות שלו
  • שימוש בכלי עם כלי בהתאמה אישית ו-MCP
  • הגדרת פלט קובץ לפי סוכן באמצעות ניהול Artifact Service
  • שימוש ב-BuiltInPlanner כדי לשפר את ביצוע המשימות באמצעות תכנון עם יכולות החשיבה של Gemini 2.5 Flash
  • אינטראקציה וניפוי באגים דרך ממשק האינטרנט של ADK
  • פריסת אפליקציה ב-Cloud Run באמצעות Dockerfile ומתן משתני סביבה

מה תצטרכו

  • דפדפן האינטרנט Chrome
  • חשבון Gmail
  • פרויקט ב-Cloud עם חיוב מופעל
  • (אופציונלי) מרחב ב-Confluence עם דפים של מסמכי דרישות מוצר

ב-Codelab הזה, שמיועד למפתחים בכל הרמות (כולל מתחילים), נעשה שימוש ב-Python באפליקציה לדוגמה. עם זאת, לא נדרש ידע ב-Python כדי להבין את המושגים שמוצגים. אל דאגה אם אין לכם מרחב ב-Confluence, אנחנו נספק לכם פרטי כניסה כדי לנסות את ה-codelab הזה

‫2. לפני שמתחילים

בחירת פרויקט פעיל ב-Cloud Console

ב-Codelab הזה אנחנו יוצאים מנקודת הנחה שכבר יש לכם פרויקט בענן ב-Google Cloud עם חיוב מופעל. אם עדיין אין לכם חשבון, אתם יכולים לפעול לפי ההוראות שבהמשך כדי להתחיל.

  1. ב-מסוף Google Cloud, בדף לבחירת הפרויקט, בוחרים או יוצרים פרויקט ב-Google Cloud.
  2. מוודאים שהחיוב מופעל בפרויקט ב-Cloud. כך בודקים אם החיוב מופעל בפרויקט

66ddfbce2eeba53e.png

הגדרת פרויקט בענן בטרמינל Cloud Shell

  1. תשתמשו ב-Cloud Shell, סביבת שורת פקודה שפועלת ב-Google Cloud. לוחצים על 'הפעלת Cloud Shell' בחלק העליון של מסוף Google Cloud.

26f20e837ff06119.png

  1. אחרי שמתחברים ל-Cloud Shell, בודקים שכבר בוצע אימות ושהפרויקט מוגדר למזהה הפרויקט באמצעות הפקודה הבאה:
gcloud auth list
  1. מריצים את הפקודה הבאה ב-Cloud Shell כדי לוודא שפקודת gcloud מכירה את הפרויקט.
gcloud config list project
  1. אם הפרויקט לא מוגדר, משתמשים בפקודה הבאה כדי להגדיר אותו:
gcloud config set project <YOUR_PROJECT_ID>

אפשר גם לראות את המזהה PROJECT_ID במסוף

bb98435b79995b15.jpeg

לוחצים עליו וכל הפרויקטים ומזהה הפרויקט יופיעו בצד שמאל.

ffa73dee57de5307.jpeg

  1. מפעילים את ממשקי ה-API הנדרשים באמצעות הפקודה שמוצגת למטה. זה יימשך כמה דקות, אז כדאי לחכות בסבלנות.
gcloud services enable aiplatform.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       cloudresourcemanager.googleapis.com

אם הפקודה תפעל בהצלחה, תוצג הודעה שדומה לזו שמופיעה בהמשך:

Operation "operations/..." finished successfully.

אפשר גם לחפש כל מוצר במסוף או להשתמש בקישור הזה במקום בפקודת gcloud.

אם פספסתם API כלשהו, תמיד תוכלו להפעיל אותו במהלך ההטמעה.

אפשר לעיין במאמרי העזרה בנושא פקודות gcloud ושימוש בהן.

כניסה אל Cloud Shell Editor והגדרת ספריית העבודה של האפליקציה

עכשיו אפשר להגדיר את עורך הקוד כדי לבצע פעולות שקשורות לקוד. נשתמש ב-Cloud Shell Editor לצורך זה

  1. לוחצים על הלחצן Open Editor (פתיחת העורך). כך ייפתח Cloud Shell Editor, שבו אפשר לכתוב את הקוד 168eacea651b086c.png
  2. מוודאים שהפרויקט ב-Cloud Code מוגדר בפינה הימנית התחתונה (שורת הסטטוס) של העורך ב-Cloud Shell, כפי שמודגש בתמונה שלמטה, ושהוא מוגדר לפרויקט הפעיל ב-Google Cloud שבו החיוב מופעל. אם מתבקשים, לוחצים על אישור. אם כבר ביצעתם את הפקודה הקודמת, יכול להיות שהלחצן יפנה ישירות לפרויקט שהפעלתם במקום ללחצן הכניסה.

613a4889dbcade3d.png

  1. לאחר מכן, משכפלים מ-GitHub את ספריית העבודה של התבנית בשביל ה-codelab הזה, ומריצים את הפקודה הבאה. הפעולה הזו תיצור את ספריית העבודה בספרייה qa-test-planner-agent
git clone https://github.com/alphinside/qa-test-planner-agent.git qa-test-planner-agent
  1. אחרי זה, עוברים לחלק העליון של Cloud Shell Editor ולוחצים על File->Open Folder (קובץ > פתיחת תיקייה), מאתרים את ספריית שם המשתמש ואת הספרייה qa-test-planner-agent, ואז לוחצים על הלחצן OK. הפעולה הזו תגרום לכך שהספרייה שנבחרה תהפוך לספריית העבודה הראשית. בדוגמה הזו, שם המשתמש הוא alvinprayuda, ולכן נתיב הספרייה מוצג למטה

c87d2b76896d0c59.png

efd32490d592dbb9.png

עכשיו Cloud Shell Editor אמור להיראות כך

a1a380fb20ce3308.png

הגדרת הסביבה

הכנת סביבה וירטואלית של Python

השלב הבא הוא הכנת סביבת הפיתוח. הטרמינל הפעיל הנוכחי צריך להיות בתוך ספריית העבודה qa-test-planner-agent. ב-Codelab הזה נשתמש ב-Python 3.12 וב-uv python project manager כדי לפשט את הצורך ביצירה ובניהול של גרסת Python וסביבה וירטואלית.

  1. אם עדיין לא פתחתם את הטרמינל, פותחים אותו על ידי לחיצה על Terminal (טרמינל) -> New Terminal (טרמינל חדש), או באמצעות Ctrl + Shift + C. חלון הטרמינל ייפתח בחלק התחתון של הדפדפן.

8635b60ae2f45bbc.jpeg

  1. מורידים את uv ומתקינים את Python 3.12 באמצעות הפקודה הבאה
curl -LsSf https://astral.sh/uv/0.7.19/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
  1. עכשיו מאתחלים את הסביבה הווירטואלית באמצעות uv. מריצים את הפקודה הזו
uv sync --frozen

תיקיית .venv תיצור ותתקין את יחסי התלות. תצוגה מקדימה מהירה של pyproject.toml תיתן לכם מידע על התלות שמוצגת כך

dependencies = [
    "google-adk>=1.5.0",
    "mcp-atlassian>=0.11.9",
    "pandas>=2.3.0",
    "python-dotenv>=1.1.1",
]
  1. כדי לבדוק את הסביבה הווירטואלית, יוצרים קובץ חדש בשם main.py ומעתיקים את הקוד הבא
def main():
   print("Hello from qa-test-planner-agent")

if __name__ == "__main__":
   main()
  1. לאחר מכן, מריצים את הפקודה הבאה
uv run main.py

יוצג פלט כמו זה שמופיע למטה

Using CPython 3.12
Creating virtual environment at: .venv
Hello from qa-test-planner-agent!

השורה הזו מראה שהפרויקט של Python מוגדר בצורה תקינה.

עכשיו אפשר לעבור לשלב הבא, בניית הסוכן ואז השירותים

3. יצירת הסוכן באמצעות Google ADK ו-Gemini 2.5

מבוא למבנה הספריות של ADK

נתחיל בסקירה של האפשרויות ש-ADK מציע ושל אופן בניית הסוכן. אפשר לגשת לתיעוד המלא של ADK בכתובת ה-URL הזו . חבילת ה-ADK מציעה לנו הרבה כלי עזר בביצוע פקודות ה-CLI שלה. לדוגמה :

  • הגדרת מבנה הספריות של הסוכן
  • ניסיון מהיר של אינטראקציה באמצעות קלט ופלט של CLI
  • הגדרה מהירה של ממשק משתמש מקומי לפיתוח

עכשיו ניצור את מבנה הספריות של הסוכן באמצעות פקודת ה-CLI. מריצים את הפקודה הבאה

uv run adk create qa_test_planner \
   --model gemini-2.5-flash \
   --project {your-project-id} \
   --region global

הפעולה הזו תיצור את מבנה ספריית הסוכן הבא בספריית העבודה הנוכחית

qa_test_planner/
├── __init__.py
├── .env
├── agent.py

אם בודקים את הקבצים init.py ו-agent.py, רואים את הקוד הזה

# __init__.py

from . import agent

# agent.py

from google.adk.agents import Agent

root_agent = Agent(
    model='gemini-2.5-flash',
    name='root_agent',
    description='A helpful assistant for user questions.',
    instruction='Answer user questions to the best of your knowledge',
)

יצירת סוכן לתכנון בדיקות של בקרת איכות

בואו ניצור סוכן לתכנון בדיקות בקרת איכות. פותחים את הקובץ qa_test_planner/agent.py ומעתיקים את הקוד שבהמשך, שיכיל את root_agent.

# qa_test_planner/agent.py

from google.adk.agents import Agent
from google.adk.tools.mcp_tool.mcp_toolset import (
    MCPToolset,
    StdioConnectionParams,
    StdioServerParameters,
)
from google.adk.planners import BuiltInPlanner
from google.genai import types
from dotenv import load_dotenv
import os
from pathlib import Path
from pydantic import BaseModel
from typing import Literal
import tempfile
import pandas as pd
from google.adk.tools import ToolContext


load_dotenv(dotenv_path=Path(__file__).parent / ".env")

confluence_tool = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="uvx",
            args=[
                "mcp-atlassian",
                f"--confluence-url={os.getenv('CONFLUENCE_URL')}",
                f"--confluence-username={os.getenv('CONFLUENCE_USERNAME')}",
                f"--confluence-token={os.getenv('CONFLUENCE_TOKEN')}",
                "--enabled-tools=confluence_search,confluence_get_page,confluence_get_page_children",
            ],
            env={},
        ),
        timeout=60,
    ),
)


class TestPlan(BaseModel):
    test_case_key: str
    test_type: Literal["manual", "automatic"]
    summary: str
    preconditions: str
    test_steps: str
    expected_result: str
    associated_requirements: str


async def write_test_tool(
    prd_id: str, test_cases: list[dict], tool_context: ToolContext
):
    """A tool to write the test plan into file

    Args:
        prd_id: Product requirement document ID
        test_cases: List of test case dictionaries that should conform to these fields:
            - test_case_key: str
            - test_type: Literal["manual","automatic"]
            - summary: str
            - preconditions: str
            - test_steps: str
            - expected_result: str
            - associated_requirements: str

    Returns:
        A message indicating success or failure of the validation and writing process
    """
    validated_test_cases = []
    validation_errors = []

    # Validate each test case
    for i, test_case in enumerate(test_cases):
        try:
            validated_test_case = TestPlan(**test_case)
            validated_test_cases.append(validated_test_case)
        except Exception as e:
            validation_errors.append(f"Error in test case {i + 1}: {str(e)}")

    # If validation errors exist, return error message
    if validation_errors:
        return {
            "status": "error",
            "message": "Validation failed",
            "errors": validation_errors,
        }

    # Write validated test cases to CSV
    try:
        # Convert validated test cases to a pandas DataFrame
        data = []
        for tc in validated_test_cases:
            data.append(
                {
                    "Test Case ID": tc.test_case_key,
                    "Type": tc.test_type,
                    "Summary": tc.summary,
                    "Preconditions": tc.preconditions,
                    "Test Steps": tc.test_steps,
                    "Expected Result": tc.expected_result,
                    "Associated Requirements": tc.associated_requirements,
                }
            )

        # Create DataFrame from the test case data
        df = pd.DataFrame(data)

        if not df.empty:
            # Create a temporary file with .csv extension
            with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as temp_file:
                # Write DataFrame to the temporary CSV file
                df.to_csv(temp_file.name, index=False)
                temp_file_path = temp_file.name

            # Read the file bytes from the temporary file
            with open(temp_file_path, "rb") as f:
                file_bytes = f.read()

            # Create an artifact with the file bytes
            await tool_context.save_artifact(
                filename=f"{prd_id}_test_plan.csv",
                artifact=types.Part.from_bytes(data=file_bytes, mime_type="text/csv"),
            )

            # Clean up the temporary file
            os.unlink(temp_file_path)

            return {
                "status": "success",
                "message": (
                    f"Successfully wrote {len(validated_test_cases)} test cases to "
                    f"CSV file: {prd_id}_test_plan.csv"
                ),
            }
        else:
            return {"status": "warning", "message": "No test cases to write"}
    except Exception as e:
        return {
            "status": "error",
            "message": f"An error occurred while writing to CSV: {str(e)}",
        }


root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    description="You are an expert QA Test Planner and Product Manager assistant",
    instruction=f"""
Help user search any product requirement documents on Confluence. Furthermore you also can provide the following capabilities when asked:
- evaluate product requirement documents and assess it, then give expert input on what can be improved 
- create a comprehensive test plan following Jira Xray mandatory field formatting, result showed as markdown table. Each test plan must also have explicit mapping on 
    which user stories or requirements identifier it's associated to 

Here is the Confluence space ID with it's respective document grouping:

- "{os.getenv("CONFLUENCE_PRD_SPACE_ID")}" : space to store Product Requirements Documents

Do not making things up, Always stick to the fact based on data you retrieve via tools.
""",
    tools=[confluence_tool, write_test_tool],
    planner=BuiltInPlanner(
        thinking_config=types.ThinkingConfig(
            include_thoughts=True,
            thinking_budget=2048,
        )
    ),
)

הגדרת קובצי תצורה

עכשיו צריך להוסיף הגדרות נוספות לפרויקט הזה, כי לסוכן הזה תהיה גישה ל-Confluence

פותחים את qa_test_planner/.env ומוסיפים לו את הערכים של משתני הסביבה הבאים, ומוודאים שקובץ .env שמתקבל נראה כך:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT={YOUR-CLOUD-PROJECT-ID}
GOOGLE_CLOUD_LOCATION=global
CONFLUENCE_URL={YOUR-CONFLUENCE-DOMAIN}
CONFLUENCE_USERNAME={YOUR-CONFLUENCE-USERNAME}
CONFLUENCE_TOKEN={YOUR-CONFLUENCE-API-TOKEN}
CONFLUENCE_PRD_SPACE_ID={YOUR-CONFLUENCE-SPACE-ID}

לצערנו, אי אפשר להגדיר את המרחב הזה ב-Confluence כציבורי, לכן אפשר לבדוק את הקבצים האלה כדי לקרוא את מסמכי דרישות המוצר שיהיו זמינים באמצעות פרטי הכניסה שלמעלה.

הסבר על הקוד

הסקריפט הזה מכיל את ההפעלה של הנציג, שבה אנחנו מאתחלים את הדברים הבאים:

  • הגדרת המודל שבו רוצים להשתמש ל-gemini-2.5-flash
  • מגדירים את כלי ה-MCP של Confluence, שיתקשרו באמצעות Stdio
  • הגדרת כלי מותאם אישית write_test_tool לכתיבת תוכנית בדיקה ולייצוא קובץ CSV לארטיפקט
  • הגדרת התיאור וההוראות של הסוכן
  • הפעלת תכנון לפני יצירת התשובה הסופית או הביצוע באמצעות יכולות החשיבה של Gemini 2.5 Flash

הסוכן עצמו, כשהוא מופעל על ידי מודל Gemini עם יכולות חשיבה מובנות ומוגדר עם ארגומנטים של מתכנן, יכול להציג את יכולות החשיבה שלו ולהופיע גם בממשק האינטרנט. בהמשך מופיע הקוד להגדרת האפשרות הזו

# qa-test-planner/agent.py

from google.adk.planners import BuiltInPlanner
from google.genai import types

...

# Provide the confluence tool to agent

root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    ...,
    tools=[confluence_tool, write_test_tool],
    planner=BuiltInPlanner(
        thinking_config=types.ThinkingConfig(
            include_thoughts=True,
            thinking_budget=2048,
        )
    ),

...

ולפני שמבצעים פעולות, אפשר לראות את תהליך החשיבה שלו

7de7004af0fc8c90.png

כלי ה-MCP של Confluence

כדי להתחבר לשרת MCP מ-ADK, צריך להשתמש ב-MCPToolSet שאפשר לייבא מהמודול google.adk.tools.mcp_tool.mcp_toolset. הקוד לאתחול מוצג כאן ( קוצר כדי לשפר את היעילות)

# qa-test-planner/agent.py

from google.adk.tools.mcp_tool.mcp_toolset import (
    MCPToolset,
    StdioConnectionParams,
    StdioServerParameters,
)

...

# Initialize the Confluence MCP Tool via Stdio Output

confluence_tool = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="uvx",
            args=[
                "mcp-atlassian",
                f"--confluence-url={os.getenv('CONFLUENCE_URL')}",
                f"--confluence-username={os.getenv('CONFLUENCE_USERNAME')}",
                f"--confluence-token={os.getenv('CONFLUENCE_TOKEN')}",
                "--enabled-tools=confluence_search,confluence_get_page,confluence_get_page_children",
            ],
            env={},
        ),
        timeout=60,
    ),
)


...

# Provide the confluence tool to agent

root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    ...,
    tools=[confluence_tool, write_test_tool],

...

בהגדרה הזו, הסוכן יאתחל את שרת ה-MCP של Confluence כתהליך נפרד, והוא יטפל בתקשורת עם התהליכים האלה באמצעות Studio I/O. התרשים הבא של ארכיטקטורת ה-MCP מציג את התהליך הזה בתוך התיבה האדומה.

8d8d8fd8e23d3c82.png

בנוסף, בארגומנטים של הפקודה לאתחול MCP, אנחנו מגבילים את הכלים שניתן להשתמש בהם רק לכלים הבאים: confluence_search, confluence_get_page, ו-confluence_get_page_children, שתומכים בתרחישי השימוש של סוכן בדיקת ה-QA שלנו. במדריך Codelab הזה אנחנו משתמשים בשרת Atlassian MCP שנתרם על ידי הקהילה ( פרטים נוספים זמינים במסמכי התיעוד המלאים).

כלי הכתיבה

אחרי שהסוכן מקבל הקשר מהכלי Confluence MCP, הוא יכול ליצור את תוכנית הבדיקה הדרושה למשתמש. עם זאת, אנחנו רוצים ליצור קובץ שמכיל את תוכנית הבדיקה הזו כדי שנוכל לשמור אותו ולשתף אותו עם האדם השני. כדי לתמוך בכך, אנחנו מספקים את הכלי המותאם אישית write_test_tool בהמשך

# qa-test-planner/agent.py

...

async def write_test_tool(
    prd_id: str, test_cases: list[dict], tool_context: ToolContext
):
    """A tool to write the test plan into file

    Args:
        prd_id: Product requirement document ID
        test_cases: List of test case dictionaries that should conform to these fields:
            - test_case_key: str
            - test_type: Literal["manual","automatic"]
            - summary: str
            - preconditions: str
            - test_steps: str
            - expected_result: str
            - associated_requirements: str

    Returns:
        A message indicating success or failure of the validation and writing process
    """
    validated_test_cases = []
    validation_errors = []

    # Validate each test case
    for i, test_case in enumerate(test_cases):
        try:
            validated_test_case = TestPlan(**test_case)
            validated_test_cases.append(validated_test_case)
        except Exception as e:
            validation_errors.append(f"Error in test case {i + 1}: {str(e)}")

    # If validation errors exist, return error message
    if validation_errors:
        return {
            "status": "error",
            "message": "Validation failed",
            "errors": validation_errors,
        }

    # Write validated test cases to CSV
    try:
        # Convert validated test cases to a pandas DataFrame
        data = []
        for tc in validated_test_cases:
            data.append(
                {
                    "Test Case ID": tc.test_case_key,
                    "Type": tc.test_type,
                    "Summary": tc.summary,
                    "Preconditions": tc.preconditions,
                    "Test Steps": tc.test_steps,
                    "Expected Result": tc.expected_result,
                    "Associated Requirements": tc.associated_requirements,
                }
            )

        # Create DataFrame from the test case data
        df = pd.DataFrame(data)

        if not df.empty:
            # Create a temporary file with .csv extension
            with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as temp_file:
                # Write DataFrame to the temporary CSV file
                df.to_csv(temp_file.name, index=False)
                temp_file_path = temp_file.name

            # Read the file bytes from the temporary file
            with open(temp_file_path, "rb") as f:
                file_bytes = f.read()

            # Create an artifact with the file bytes
            await tool_context.save_artifact(
                filename=f"{prd_id}_test_plan.csv",
                artifact=types.Part.from_bytes(data=file_bytes, mime_type="text/csv"),
            )

            # Clean up the temporary file
            os.unlink(temp_file_path)

            return {
                "status": "success",
                "message": (
                    f"Successfully wrote {len(validated_test_cases)} test cases to "
                    f"CSV file: {prd_id}_test_plan.csv"
                ),
            }
        else:
            return {"status": "warning", "message": "No test cases to write"}
    except Exception as e:
        return {
            "status": "error",
            "message": f"An error occurred while writing to CSV: {str(e)}",
        }


...

הפונקציה שצוינה למעלה נועדה לתמוך בפונקציות הבאות:

  1. בודקים את תוכנית הבדיקה שנוצרה כדי לוודא שהיא עומדת בדרישות של שדות החובה. אנחנו בודקים באמצעות מודל Pydantic, ואם מתרחשת שגיאה, אנחנו שולחים את הודעת השגיאה בחזרה לסוכן
  2. ייצוא התוצאה לקובץ CSV באמצעות פונקציונליות של pandas
  3. הקובץ שנוצר נשמר כארטיפקט באמצעות היכולות של Artifact Service, שאפשר לגשת אליהן באמצעות אובייקט ToolContext שאפשר לגשת אליו בכל קריאות הכלים

אם נשמור את הקבצים שנוצרו כארטיפקט, הם יסומנו כאירוע בזמן הריצה של ADK ויוצגו באינטראקציה עם הסוכן בשלב מאוחר יותר בממשק האינטרנט

adafec781450c89a.png

כך אנחנו יכולים להגדיר באופן דינמי את התשובה של הסוכן לגבי הקובץ, שתועבר למשתמש.

4. בדיקת הסוכן

עכשיו ננסה לתקשר עם הסוכן באמצעות ה-CLI. מריצים את הפקודה הבאה:

uv run adk run qa_test_planner

יוצג פלט כמו זה, שבו תוכלו להתכתב בצ'אט עם הסוכן בתורכם, אבל תוכלו לשלוח טקסט רק דרך הממשק הזה

Log setup complete: /tmp/agents_log/agent.xxxx_xxx.log
To access latest log: tail -F /tmp/agents_log/agent.latest.log
Running agent qa_test_planner_agent, type exit to exit.
user: hello
[qa_test_planner_agent]: Hello there! How can I help you today?
user:

היכולת להתכתב בצ'אט עם נציג באמצעות CLI היא שימושית. אבל הכי טוב אם ננהל איתו שיחה נעימה בצ'אט באינטרנט, ואנחנו יכולים לעשות את זה גם! ערכת ה-ADK גם מאפשרת לנו להשתמש בממשק משתמש לפיתוח כדי לבצע אינטראקציה ולבדוק מה קורה במהלך האינטראקציה. מריצים את הפקודה הבאה כדי להפעיל את שרת ממשק המשתמש של הפיתוח המקומי

uv run adk web --port 8080

יוצג פלט כמו בדוגמה הבאה, כלומר כבר יש לנו גישה לממשק האינטרנט

INFO:     Started server process [xxxx]
INFO:     Waiting for application startup.

+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8080.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

כדי לבדוק את זה, לוחצים על הלחצן Web Preview (תצוגה מקדימה של אתר) בחלק העליון של Cloud Shell Editor ובוחרים באפשרות Preview on port 8080 (תצוגה מקדימה ביציאה 8080).

edc73e971b9fc60c.png

יוצג דף האינטרנט הבא שבו תוכלו לבחור סוכנים זמינים באמצעות הלחצן הנפתח בפינה הימנית העליונה ( במקרה שלנו, הלחצן הוא qa_test_planner) ולנהל אינטראקציה עם הבוט. בחלון הימני יוצגו פרטים רבים על יומן הרישום במהלך זמן הריצה של הסוכן

940b1e0391db4871.png

אפשר לנסות כמה פעולות. אפשר להשתמש בהנחיות הבאות כדי לשוחח בצ'אט עם נציגים:

  • " Please list all available PRDs "
  • " כתוב תוכנית בדיקה עבור Snaprecipe PRD "

כשמשתמשים בכלים מסוימים, אפשר לבדוק מה קורה בממשק המשתמש של הפיתוח

35a3c928bd9dfb87.png

כדאי לראות איך הסוכן מגיב לכם, וגם לבדוק מתי אנחנו מבקשים קובץ בדיקה. הוא ייצור את תוכנית הבדיקה בקובץ CSV כארטיפקט.

183558cb042cb021.png

עכשיו אפשר לבדוק את התוכן של קובץ ה-CSV על ידי ייבוא שלו ל-Google Sheets, למשל

78933239b66eddb2.png

מעולה! עכשיו יש לכם סוכן QA Test Planner פעיל שפועל באופן מקומי. עכשיו נראה איך אפשר לפרוס אותו ב-Cloud Run כדי שאנשים אחרים יוכלו להשתמש בו.

5. פריסה ב-Cloud Run

עכשיו, ברור שאנחנו רוצים לגשת לאפליקציה המדהימה הזו מכל מקום. כדי לעשות זאת, אפשר לארוז את האפליקציה הזו ולפרוס אותה ב-Cloud Run. לצורך ההדגמה הזו, השירות הזה יהיה שירות ציבורי שאנשים אחרים יוכלו לגשת אליו. עם זאת, חשוב לזכור שזו לא השיטה המומלצת.

af793beb2740233.jpeg

בספריית העבודה הנוכחית כבר יש את כל הקבצים שנדרשים לפריסת האפליקציות ב-Cloud Run – ספריית הסוכן, Dockerfile ו-server.py (סקריפט השירות הראשי). עכשיו נבצע את הפריסה. עוברים לטרמינל Cloud Shell ומוודאים שהפרויקט הנוכחי מוגדר לפרויקט הפעיל שלכם. אם לא, צריך להשתמש בפקודה gcloud configure כדי להגדיר את מזהה הפרויקט:

gcloud config set project [PROJECT_ID]

לאחר מכן, מריצים את הפקודה הבאה כדי לפרוס אותו ב-Cloud Run.

gcloud run deploy qa-test-planner-agent \
                  --source . \
                  --port 8080 \
                  --project {YOUR_PROJECT_ID} \
                  --allow-unauthenticated \
                  --region us-central1 \
                  --update-env-vars GOOGLE_GENAI_USE_VERTEXAI=1 \
                  --update-env-vars GOOGLE_CLOUD_PROJECT={YOUR_PROJECT_ID} \
                  --update-env-vars GOOGLE_CLOUD_LOCATION=global \
                  --update-env-vars CONFLUENCE_URL={YOUR_CONFLUENCE_URL} \
                  --update-env-vars CONFLUENCE_USERNAME={YOUR_CONFLUENCE_USERNAME} \
                  --update-env-vars CONFLUENCE_TOKEN={YOUR_CONFLUENCE_TOKEN} \
                  --update-env-vars CONFLUENCE_PRD_SPACE_ID={YOUR_PRD_SPACE_ID} \
                  --memory 1G

אם מוצגת בקשה לאישור יצירה של מאגר ארטיפקטים עבור מאגר Docker, פשוט משיבים Y. שימו לב שאנחנו מאפשרים כאן גישה לא מאומתת כי זו אפליקציית הדגמה. מומלץ להשתמש באימות מתאים לאפליקציות הארגוניות ולאפליקציות הייצור.

אחרי שהפריסה תושלם, תקבלו קישור שדומה לקישור שבהמשך:

https://qa-test-planner-agent-*******.us-central1.run.app

כשתיגשו לכתובת ה-URL, תועברו לממשק המשתמש של כלי הפיתוח לאתרים, בדומה לניסיון להשתמש בו באופן מקומי. אפשר להשתמש באפליקציה מהחלון הפרטי או מהנייד. הוא כבר אמור להיות פעיל.

עכשיו ננסה שוב את ההנחיות השונות האלה – אחת אחרי השנייה, ונראה מה יקרה:

  • " Can you find PRD related to Mortgage Estimator? "
  • ‫ " Give me feedback about what can we improve on that"
  • " כתוב את תוכנית הבדיקה בשבילו"

בנוסף, מכיוון שאנחנו מריצים את הסוכן כאפליקציית FastAPI, אנחנו יכולים גם לבדוק את כל מסלולי ה-API במסלול /docs. לדוגמה, אם תיכנסו לכתובת ה-URL הזו https://qa-test-planner-agent-*******.us-central1.run.app/docs, תראו דף תיעוד של Swagger כמו שמוצג בהמשך

82d2d91fb3cb2dde.png

הסבר על הקוד

עכשיו נבדוק איזה קובץ צריך כאן לפריסה, החל מ-server.py

# server.py

import os

from fastapi import FastAPI
from google.adk.cli.fast_api import get_fast_api_app

AGENT_DIR = os.path.dirname(os.path.abspath(__file__))

app_args = {"agents_dir": AGENT_DIR, "web": True}

app: FastAPI = get_fast_api_app(**app_args)

app.title = "qa-test-planner-agent"
app.description = "API for interacting with the Agent qa-test-planner-agent"


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)

אפשר להמיר בקלות את הסוכן לאפליקציית fastapi באמצעות הפונקציה get_fast_api_app. בפונקציה הזו אפשר להגדיר פונקציות שונות, למשל הגדרת שירות סשן, שירות ארטיפקט או אפילו מעקב אחר נתונים בענן.

אם רוצים, אפשר גם להגדיר כאן את מחזור החיים של האפליקציה. אחרי זה אפשר להשתמש ב-uvicorn כדי להריץ את אפליקציית Fast API.

לאחר מכן, קובץ Dockerfile יספק לנו את השלבים הדרושים להרצת האפליקציה.

# Dockerfile

FROM python:3.12-slim

RUN pip install --no-cache-dir uv==0.7.13

WORKDIR /app

COPY . .

RUN uv sync --frozen

EXPOSE 8080

CMD ["uv", "run", "uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]

6. האתגר

עכשיו הגיע הזמן שלכם לזרוח ולשפר את מיומנויות החיפוש שלכם. האם תוכל גם ליצור כלי שיכתוב לקובץ את המשוב על בדיקת מסמך PRD?

7. הסרת המשאבים

כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם ב-Code Lab הזה:

  1. במסוף Google Cloud, עוברים לדף Manage resources.
  2. ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
  3. כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.
  4. לחלופין, אפשר לעבור אל Cloud Run במסוף, לבחור את השירות שפרסתם ולמחוק אותו.