1. Tổng quan
Trong lớp học lập trình này, bạn sẽ xây dựng một hệ thống đa tác nhân, trong đó nhiều tác nhân ADK giao tiếp và cộng tác bằng giao thức Agent2Agent (A2A).
Kiến thức bạn sẽ học được
- Cách tạo nhiều tác nhân ADK độc lập
- Cách cấp cho mỗi tác nhân một Thẻ tác nhân và gói các thẻ này dưới dạng máy chủ A2A
- Cách tạo tác nhân lưu trữ điều phối các tác nhân từ xa
- Cách thiết lập kết nối tác nhân từ xa
- Cách kiểm thử hệ thống nhiều tác nhân cục bộ
Bạn cần có
- Một dự án trên Google Cloud đã bật tính năng thanh toán
- Một trình duyệt web như Chrome
- Python 3.12 trở lên và uv, nếu bạn chạy mã trên máy của mình thay vì Cloud Shell
Lớp học lập trình này dành cho các nhà phát triển có trình độ trung cấp và đã quen thuộc với Python và Google Cloud.
Bạn sẽ chỉ mất khoảng 15 phút để hoàn thành lớp học lập trình này.
Các tài nguyên được tạo trong lớp học lập trình này sẽ có chi phí dưới 5 USD.
2. Thiết lập môi trường
Tạo một dự án trên Google Cloud
- Trong Google Cloud Console, trên trang chọn dự án, hãy chọn hoặc tạo một dự án trên đám mây của Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trên một dự án hay không.
Khởi động Trình chỉnh sửa Cloud Shell
Để chạy một phiên Cloud Shell từ Cloud Console, hãy nhấp vào Kích hoạt Cloud Shell trong Cloud Console.
Thao tác này sẽ khởi chạy một phiên trong ngăn dưới cùng của Cloud Console.
Để khởi chạy trình chỉnh sửa, hãy nhấp vào Open Editor (Mở trình chỉnh sửa) trên thanh công cụ của cửa sổ Cloud Shell.
Bật API
Trong thiết bị đầu cuối, hãy chạy lệnh sau để bật API mà ADK dùng để gọi Gemini:
gcloud services enable aiplatform.googleapis.com
Định cấu hình môi trường
Bắt đầu bằng cách chạy lệnh sau trong thiết bị đầu cuối để tạo cấu trúc thư mục dự án cho hệ thống A2A. Để minh hoạ, chúng ta sẽ sử dụng đường dẫn tuyệt đối từ thư mục $HOME của bạn:
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
Giờ đây, khi đã có cấu trúc chung, hãy điền thông tin cấu hình môi trường. Sao chép đoạn mã sau vào tệp .env mới:
Điền mã dự án trên đám mây của Google vào tệp .env mới. Giữ GOOGLE_CLOUD_LOCATION ở chế độ global. Đây là nơi có mô hình Gemini trong lớp học lập trình này. Bạn cũng có thể nhập email mà bạn chọn vào MAIL_TO nếu muốn nhận email báo cáo từ tác nhân mà bạn sắp tạo. Bạn có thể liên tục thêm Mã truy cập cá nhân GitHub GITHUB_TOKEN để hỗ trợ tác nhân truy xuất github.
Mở tệp .env mới bằng lệnh bash sau:
cloudshell edit $HOME/app/.env
sau đó sao chép tệp sau vào tệp .env tại app/.env. LƯU Ý QUAN TRỌNG: Hãy nhớ nhập GIÁ TRỊ của RIÊNG BẠN.
# --- Google Cloud ---
GOOGLE_GENAI_USE_ENTERPRISE=True
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=global
# --- GitHub ---
# Personal Access Token with "repo" scope
# Create one at: https://github.com/settings/tokens
# Generate new token --> Fine-grained, repo-secured --> (populate) Token name --> (scroll down) Generate token
GITHUB_TOKEN=<your-github-pat>
GITHUB_MCP_URL=https://api.githubcopilot.com/mcp/
TARGET_REPOS=["google/adk-python", "langchain-ai/langchain"]
DEFAULT_ISSUE_COUNT=5
DEFAULT_PR_COUNT=5
# --- Email (Optional) ---
# Only needed if you want the emailer agent to send reports
RESEND_API_KEY=<your-resend-API-key>
RESEND_DOMAIN=<your-domain>
MAIL_TO=<insert-email-to-receive-reports>
# --- Local Development Overrides ---
RETRIEVAL_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
ORCHESTRATOR_PORT=http://localhost:8000
Giờ đây, bạn đã điền sẵn các biến môi trường, chúng ta phải định cấu hình môi trường uv. Mở tệp pyproject.toml:
cloudshell edit $HOME/app/pyproject.toml
Sao chép đoạn mã sau vào tệp pyproject.toml:
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "app"
version = "0.1.0"
description = "Multi-agent system designed to assess the newest issues and pull requests from numerous agentic development platforms"
authors = [
{ name = "Thomas Wagner" },
{ name = "Kris Overholt" }
]
license = "Apache-2.0"
requires-python = ">=3.11"
dependencies = [
"markdown",
"resend",
"uvicorn",
"a2a-sdk",
"google-adk[a2a,mcp]",
"httpx",
"python-dotenv",
]
[project.optional-dependencies]
test = [
"pytest",
"pytest-asyncio",
"pytest-mock",
"respx",
]
[tool.ruff]
target-version = "py313"
line-length = 80
[tool.ruff.lint]
select = ["E", "F", "I", "C", "PL", "B", "UP", "RUF"]
ignore = ["E501", "C901"]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
[tool.ruff.lint.isort]
known-first-party = ["src"]
[tool.hatch.build.targets.wheel]
packages=["src/"]
Cài đặt các phần phụ thuộc
Bây giờ, hãy chạy các lệnh sau trong thiết bị đầu cuối để chuyển vào thư mục app mà bạn vừa tạo và cài đặt tất cả các phần phụ thuộc cần thiết từ tệp pyproject.toml:
cd $HOME/app
uv sync
uv tạo một môi trường Python riêng biệt cho lớp học lập trình này, vì vậy bạn không cần kích hoạt bất cứ thứ gì. Thêm tiền tố uv run vào các lệnh Python.
Giờ đây, chúng ta đã có mọi thứ cần thiết để tạo hệ thống đa tác nhân!
3. Tạo Retriever Agent
Eva là một nhà phát triển phần mềm muốn nắm bắt thông tin mới nhất về các kho lưu trữ trên GitHub khi chúng được cập nhật vấn đề và yêu cầu kéo mới. Vì vậy, cô ấy tạo một tác nhân ADK để truy xuất dữ liệu trên GitHub mà cô ấy yêu cầu.
Để minh hoạ, hãy chạy lệnh sau từ gốc của dự án để tạo thư mục và tệp cần thiết cho tác nhân truy xuất:
mkdir -p $HOME/app/src/agents/retriever
touch $HOME/app/src/agents/retriever/agent.py
touch $HOME/app/src/agents/retriever/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/retriever/__init__.py
Sao chép đoạn mã ADK sau đây vào $HOME/app/src/agents/retriever/agent.py:
import logging
import os
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
from google.adk.tools.mcp_tool.mcp_session_manager import (
StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
logger = logging.getLogger(__name__)
load_dotenv()
GITHUB_MCP_URL = os.getenv(
"GITHUB_MCP_URL", "https://api.githubcopilot.com/mcp/"
)
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
if GITHUB_TOKEN is None:
raise ValueError("GITHUB_TOKEN env is not set")
TARGET_REPOS = os.getenv("TARGET_REPOS")
DEFAULT_ISSUE_COUNT = os.getenv("DEFAULT_ISSUE_COUNT")
DEFAULT_PR_COUNT = os.getenv("DEFAULT_PR_COUNT")
# --- Prompt ---
GITHUB_RETRIEVAL_INSTRUCTIONS = f"""
You are a specialized GitHub data retrieval agent.
Your only purpose is to fetch data from GitHub using the available MCP tools and format it for another agent.
**DEFAULT CONFIGURATION:**
If the orchestrator does not specify which repositories or how many items to fetch, you MUST use the following defaults:
- **Repositories:** {TARGET_REPOS}
- **PR Count:** {DEFAULT_PR_COUNT} per repository
- **Issue Count:** {DEFAULT_ISSUE_COUNT} per repository
**Your Task:**
1. Analyze the task. If no specific repo is mentioned, iterate through the Default Repositories list above.
2. Use the `GitHub` MCP tool to fetch the requested data (Pull Requests and/or Issues).
3. **CRITICAL:** After the tool has finished running, you MUST take the raw output and compile it into a single response.
The orchestrator is waiting for this raw data to pass to the Evaluator. Do not summarize it.
"""
# --- Agent Initialization ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="retriever",
instruction=GITHUB_RETRIEVAL_INSTRUCTIONS,
description="""
Connects to the GitHub MCP server to retrieve real-time development
insights, actively reading repository issues, pull requests, and commit
histories.
""",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url=GITHUB_MCP_URL,
headers={
"Authorization": f"Bearer {GITHUB_TOKEN}",
"X-MCP-Toolsets": "repos,issues,pull_requests",
"X-MCP-Readonly": "true",
},
)
)
],
)
Tác nhân này hoạt động như một công cụ độc lập. Để kiểm thử độc lập, hãy chạy các lệnh sau để trò chuyện với mô hình này bằng cách sử dụng adk run từ thư mục agents:
cd $HOME/app/src/agents
uv run adk run retriever
Nhập một tin nhắn tại dấu nhắc [user]: để dùng thử, sau đó nhập exit để dừng tác nhân.
4. Tạo Evaluator Agent
Richard thường nhận thấy rằng anh phải giải thích rất nhiều thuật ngữ kỹ thuật cho những khách hàng và đồng nghiệp không có kiến thức về kỹ thuật. Quá mệt mỏi khi phải định nghĩa cùng một thuật ngữ và giải thích cùng một dự án theo cách cô đọng, Richard đã tạo ra một tác nhân cô đọng để tóm tắt thuật ngữ kỹ thuật thành văn bản dễ hiểu.
Để minh hoạ, hãy chạy lệnh sau để tạo tệp cho tác nhân đánh giá:
mkdir -p $HOME/app/src/agents/evaluator
touch $HOME/app/src/agents/evaluator/agent.py
touch $HOME/app/src/agents/evaluator/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/evaluator/__init__.py
Sao chép đoạn mã tác nhân bên dưới vào $HOME/app/src/agents/evaluator/agent.py
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
load_dotenv()
# --- Prompt ---
EVAL_AND_SUMMARIZATION_INSTRUCTIONS = """
You are a specialized analysis and summarization agent. Your only purpose is to take raw, structured text about GitHub repositories and transform it into a concise, human-readable Markdown report.
**Your Task:**
1. You will receive a block of text containing raw data about pull requests and issues from an orchestrator.
2. Analyze the provided data. For pull requests, evaluate their significance. For issues, identify key themes and problems.
3. **CRITICAL:** You MUST generate a comprehensive summary in Markdown format based on your analysis. Your final output should be ONLY this Markdown report. Do not add any conversational text or explanations (e.g., "Here is the summary..."). The orchestrator needs to pass your clean report to another agent or directly to the user.
**Output Format Rules:**
- The report MUST be in Markdown.
- Structure the report by repository.
- For each repository, provide a concise overview of significant pull requests and important issues.
- Conclude with overall insights.
The orchestrator is waiting for this report. Ensure your final response consists of nothing but the complete Markdown summary.
"""
# --- Agent ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="evaluator",
instruction=EVAL_AND_SUMMARIZATION_INSTRUCTIONS,
description="""
Distills and summarizes complex GitHub data, breaking down pull
requests, code changes, and issue discussions into easily understandable
insights.
""",
)
Tác nhân này hoạt động như một công cụ độc lập. Để kiểm thử độc lập, hãy chạy các lệnh sau trong một cửa sổ dòng lệnh MỚI để trò chuyện với nó bằng adk run từ thư mục agents:
cd $HOME/app/src/agents
uv run adk run evaluator
Nhập một tin nhắn tại dấu nhắc [user]: để dùng thử, sau đó nhập exit để dừng tác nhân.
5. Tạo Tác nhân gửi email
Ivan cảm thấy mệt mỏi vì phải viết quá nhiều email chỉ để tóm tắt và định dạng lại một đoạn văn bản có sẵn. Vì vậy, anh ấy đã tạo một tác nhân gửi email để định dạng lại và gửi một văn bản nhất định đến một tài khoản email được cung cấp.
Để minh hoạ, hãy chạy lệnh sau trong thiết bị đầu cuối để tạo tệp cho tác nhân gửi email:
mkdir -p $HOME/app/src/agents/emailer
touch $HOME/app/src/agents/emailer/agent.py
touch $HOME/app/src/agents/emailer/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/emailer/__init__.py
Sao chép đoạn mã tác nhân sau đây vào app/src/agents/emailer/agent.py
import logging
import os
import markdown
import resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
load_dotenv()
logger = logging.getLogger(__name__)
RESEND_API_KEY = os.getenv("RESEND_API_KEY")
RESEND_DOMAIN = os.getenv("RESEND_DOMAIN")
MAIL_TO = os.getenv("MAIL_TO")
if RESEND_API_KEY:
resend.api_key = RESEND_API_KEY
# --- Tools ---
def send_report_email(recipient: str, subject: str, body: str) -> str:
"""
Sends an email of the given subject and body to the specified recipient
via Resend. If recipient is None, it defaults to the MAIL_TO environment variable.
Args:
recipient (str | None): The email address of the recipient. If None, defaults to MAIL_TO.
subject (str): The subject of the email.
body (str): The body of the email.
Returns:
str: A success or failure message.
"""
if not recipient:
recipient = MAIL_TO
if not all([RESEND_API_KEY, RESEND_DOMAIN, recipient]):
error_msg = "Error: Email tool configuration missing (API Key, Domain, or Recipient)"
logger.error(error_msg)
return error_msg
try:
html_body = markdown.markdown(body, extensions=["tables"])
params: resend.Emails.SendParams = {
"from": f"Research Agent <agent@{RESEND_DOMAIN}>",
"to": [recipient], #type: ignore
"subject": subject,
"text": body,
"html": html_body,
}
email = resend.Emails.send(params)
logger.info(f"Email sent successfully. ID: {email['id']}")
return f"Email sent! ID: {email['id']}"
except Exception as e:
error_msg = f"Failed to send email: {e}"
logger.error(error_msg)
return error_msg
# --- Prompt ---
EMAIL_INSTRUCTIONS = """
You are an emailer agent responsible for formatting and sending a research report via email.
INPUT: You will receive a Markdown-formatted string (the report) and the email recipient.
YOUR TASK:
1. Take the Markdown content and format it appropriately for an email body.
The goal is to render the Markdown effectively so it is readable and well-presented in an email client.
2. Based on the summary, generate a concise and informative subject line for the email.
The subject line should reflect the main themes or key insights from the report.
3. Use the `send_report_email` tool with the extracted recipient, the generated subject line, and the formatted email body.
If no email recipient is provided, output a message indicating that no recipient was specified and do NOT call the tool.
"""
# --- Agent ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="emailer",
instruction=EMAIL_INSTRUCTIONS,
description="""
Acts as the final delivery mechanism in the pipeline, dispatching
generated summaries and reports to designated email addresses.
""",
tools=[
send_report_email
],
)
Tác nhân này hoạt động như một công cụ độc lập. Để kiểm thử độc lập, hãy chạy các lệnh sau trong một cửa sổ dòng lệnh MỚI để trò chuyện với nó bằng adk run từ thư mục agents:
cd $HOME/app/src/agents
uv run adk run emailer
Nhập một tin nhắn tại dấu nhắc [user]: để dùng thử, sau đó nhập exit để dừng tác nhân.
6. Tạo thẻ tác nhân
Chúng ta vừa xem qua 3 câu chuyện của nhà phát triển độc lập, mỗi câu chuyện đều có những nhân vật độc đáo riêng. Tất cả các tác nhân này đều được xuất bản và đăng nhập vào Thư viện tác nhân của công ty. Một nhà phát triển khác là Diane muốn kết hợp những ý tưởng riêng lẻ này thành một hệ thống đa tác nhân duy nhất.
Mục tiêu mà cô ấy muốn đạt được là có một tác nhân nghiên cứu theo yêu cầu, giúp cô ấy nắm bắt thông tin mới nhất về các vấn đề và yêu cầu kéo từ nhiều khung phát triển tác nhân. Cô ấy cũng muốn hệ thống gửi email tóm tắt ngắn gọn về tất cả những điểm mới trong hệ sinh thái. Vì tất cả các tác nhân đều được phát triển riêng lẻ trong các môi trường khác nhau, nên A2A là một giải pháp lý tưởng để kết hợp các tác nhân hiện có này.
Bước đầu tiên trong việc tập hợp một nhóm nhân viên hỗ trợ từ xa là cấp cho mỗi nhân viên hỗ trợ từ xa cần thiết một Thẻ nhân viên hỗ trợ. Hãy coi đây là danh thiếp cho nhân viên hỗ trợ của bạn. Đây là các tệp JSON dùng để giúp tác nhân lưu trữ có khả năng xác định các tác nhân theo tên, nội dung mô tả, chức năng và giản đồ đầu vào/đầu ra. Các thẻ trong lớp học lập trình này sử dụng định dạng Thẻ tác nhân A2A phiên bản 1.0, liệt kê từng điểm cuối mà tác nhân có thể truy cập trong supportedInterfaces.
Trước tiên, hãy chạy lệnh sau từ thư mục gốc của dự án để tạo thư mục cards và tất cả các tệp JSON cần thiết cho thẻ đại lý:
mkdir -p $HOME/app/cards/
touch $HOME/app/cards/github_retrieval_agent_card.json
touch $HOME/app/cards/content_evaluator_agent_card.json
touch $HOME/app/cards/emailer_agent_card.json
Bạn chỉ có thể truy cập vào các tệp này thông qua lệnh bash sau:
cloudshell edit $HOME/app/cards/github_retrieval_agent_card.json
Sao chép nội dung JSON sau đây vào tệp app/cards/github_retrieval_agent_card.json:
{
"name": "GitHub_Retrieval_Agent",
"description": "Connects to the GitHub MCP server to retrieve real-time development insights, actively reading repository issues, pull requests, and commit histories.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8001",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"json",
"text/plain"
],
"skills": [
{
"id": "read_github_repos",
"name": "GitHub_Retriever",
"description": "Fetches and analyzes recent content updates from GitHub repositories, including open/closed issues, pull request statuses, and detailed commit logs.",
"tags": [
"Find the newest pull requests from",
"Check recent issues in",
"Summarize commits for",
"GitHub repository status"
],
"examples": [
"Find the 10 most recent pull requests from the langchain-ai/langgraph repository along with their commits.",
"List all open issues tagged with 'bug' in the current repository."
]
}
]
}
Sao chép nội dung sau vào tệp app/cards/content_evaluator_agent_card.json. Thao tác này sẽ cung cấp cho tác nhân lưu trữ nội dung mô tả về những việc mà tác nhân này có thể làm, loại dữ liệu bạn có thể cung cấp cho tác nhân và những gì tác nhân sẽ trả về.
Tương tự như trước đây, hãy dùng lệnh này để chỉnh sửa thẻ tác nhân đánh giá:
cloudshell edit $HOME/app/cards/content_evaluator_agent_card.json
{
"name": "Content_Evaluation_Agent",
"description": "Distills and summarizes complex GitHub data, breaking down pull requests, code changes, and issue discussions into easily understandable insights.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8002",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"json",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "evaluate_github_content",
"name": "Evaluate GitHub Content",
"description": "Analyzes and synthesizes provided GitHub data—including code diffs, commit histories, issue threads, and PR comments—into clear, structured summaries.",
"tags": [
"summarize pull request",
"evaluate GitHub changes",
"break down commits",
"distill issue discussion",
"code review summary"
],
"examples": [
"Make a thorough summary of the code changes and commit history from this pull request data.",
"Break down the main arguments and proposed solutions from this provided GitHub issue discussion."
]
}
]
}
Đây là Thẻ nhân viên hỗ trợ cho nhân viên hỗ trợ gửi email. Sao chép nội dung đó vào tệp app/cards/emailer_agent_card.json bằng cùng một lệnh cloudshell:
cloudshell edit $HOME/app/cards/emailer_agent_card.json
{
"name": "Email_Agent",
"description": "Acts as the final delivery mechanism in the pipeline, dispatching generated summaries and reports to designated email addresses.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8003",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "dispatch_email_report",
"name": "Dispatch_Report_via_Email",
"description": "Takes synthesized text data and securely emails it to a specified recipient or distribution list.",
"tags": [
"send email",
"email summary",
"dispatch report",
"forward to inbox"
],
"examples": [
"Email me the summarized evaluations from the retrieved pull request and issue data.",
"Take this code review breakdown and send it in an email to the dev team."
]
}
]
}
7. Tạo Remote Agent Executor
Để tác nhân từ xa có thể được tác nhân lưu trữ "gọi" hoặc "trò chuyện", mỗi tác nhân cần có một trình thực thi tác nhân mà máy chủ A2A có thể nhìn thấy, được xác định bằng Thẻ tác nhân. Trình thực thi là một lớp trừu tượng có các phương thức execute() và cancel(), sẽ gọi tác nhân từ xa và cung cấp cho tác nhân đó một tác vụ hoặc thông báo, hoặc huỷ hoàn toàn tác vụ.
Sau đây là một cách triển khai trình thực thi nhân viên hỗ trợ ADK trừu tượng, giúp gửi tin nhắn và sắp xếp các việc cần làm cho nhân viên hỗ trợ từ xa, tất cả đều trao đổi các cấu phần phần mềm văn bản Part. Do đó, chúng ta có thể sử dụng một trình thực thi tác nhân chung được chia sẻ giữa các tác nhân. Tạo một tệp thực thi mới:
touch $HOME/app/src/agents/executor.py
Sao chép đoạn mã sau vào tệp $HOME/app/src/agents/executor.py mới:
# $HOME/app/src/agents/executor.py
from a2a.helpers import new_task_from_user_message, new_text_message
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import Part, TaskState
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class BaseAgentExecutor(AgentExecutor):
"""An AgentExecutor that runs a remote ADK Agent"""
def __init__(
self,
agent,
status_message='Processing request...',
artifact_name='response',
):
"""Initialize a generic ADK agent executor.
Args:
agent: The ADK agent instance
status_message: Message to display while processing
artifact_name: Name for the response artifact
"""
self.agent = agent
self.status_message = status_message
self.artifact_name = artifact_name
self.runner = Runner(
app_name=agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
task = context.current_task or new_task_from_user_message(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
user_id = context.call_context.user.user_name or "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.TASK_STATE_WORKING,
new_text_message(
self.status_message,
context_id=task.context_id,
task_id=task.id,
),
)
# Process with ADK agent, reusing the session for this context
session = await self.runner.session_service.get_session(
app_name=self.agent.name,
user_id=user_id,
session_id=task.context_id,
)
if session is None:
session = await self.runner.session_service.create_session(
app_name=self.agent.name,
user_id=user_id,
state={},
session_id=task.context_id,
)
content = types.Content(
role="user", parts=[types.Part.from_text(text=query)]
)
response_text = ""
async for event in self.runner.run_async(
user_id=user_id, session_id=session.id, new_message=content
):
if event.is_final_response() and event.content and event.content.parts:
for part in event.content.parts:
if hasattr(part, "text") and part.text:
response_text += part.text + "\n"
elif hasattr(part, "function_call"):
# Log or handle function calls if needed
pass # Function calls are handled internally by ADK
# Add response as artifact with custom name
await updater.add_artifact(
[Part(text=response_text)],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.TASK_STATE_FAILED,
new_text_message(
f"Error: {e!s}",
context_id=task.context_id,
task_id=task.id,
),
)
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue
) -> None:
"""Cancel the execution of a specific task."""
raise NotImplementedError("Cancel not implemented for RetrieverAgentExecutor")
Giờ đây, khi đã có trình thực thi tác nhân trừu tượng đó, chúng ta có thể sử dụng tính năng kế thừa theo kiểu Python để tinh chỉnh trình thực thi cho từng chức năng và nhu cầu cụ thể của tác nhân. Trong trường hợp của chúng tôi, việc thực thi lệnh gọi tác nhân bằng tải trọng, chờ phản hồi và truy xuất tải trọng phản hồi là phổ biến đối với quy trình làm việc của chúng tôi. Do đó, chúng tôi không cần bất kỳ thay đổi cụ thể nào. Tuy nhiên, mỗi máy chủ A2A của tác nhân đều cần một trình thực thi tác nhân. Chạy lệnh sau để tạo các tệp thực thi cho cả 3 tác nhân:
touch $HOME/app/src/agents/retriever/executor.py
touch $HOME/app/src/agents/evaluator/executor.py
touch $HOME/app/src/agents/emailer/executor.py
Giờ đây, hãy thêm nội dung tương ứng vào từng tệp.
# $HOME/app/src/agents/retriever/executor.py
from ..executor import BaseAgentExecutor
class RetrieverAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Retriever Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/evaluator/executor.py
from ..executor import BaseAgentExecutor
class EvaluatorAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Evaluator Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/emailer/executor.py
from ..executor import BaseAgentExecutor
class EmailerAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Emailer Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
8. Hiển thị các Tác nhân từ xa cho Máy chủ A2A
Giờ đây, mỗi tác nhân đều có danh thiếp riêng, nên có thể được phát hiện khi tiếp xúc với một điểm cuối thông qua máy chủ A2A. Để phục vụ cho lớp học lập trình này, chúng ta sẽ giữ toàn bộ quy trình ở chế độ cục bộ bằng cách sử dụng localhost cho từng điểm cuối của tác nhân từ xa. Nhưng trên thực tế, các thông tin này có thể được hiển thị cho bất kỳ điểm cuối nào bạn muốn và sẽ có thể phát hiện được miễn là thông tin đó được tham chiếu trong url của trường supportedInterfaces trong thẻ đại lý.
Mỗi máy chủ đều phân phát Thẻ đại lý tại /.well-known/agent-card.json. Đây là nơi mà các ứng dụng A2A tìm kiếm thẻ này và xử lý các yêu cầu JSON-RPC A2A tại /.
Bước tiếp theo cho mỗi tác nhân từ xa là hiển thị chúng cho máy chủ A2A riêng. Chạy lệnh này để tạo tệp server.py cho cả 3 tác nhân từ xa:
touch $HOME/app/src/agents/retriever/server.py
touch $HOME/app/src/agents/evaluator/server.py
touch $HOME/app/src/agents/emailer/server.py
Bây giờ, bạn sẽ thêm mã máy chủ cho từng tác nhân. Bắt đầu với tác nhân truy xuất:
# $HOME/app/src/agents/retriever/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as retriever_agent
from .executor import RetrieverAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/github_retrieval_agent_card.json", "r") as f:
card_data = json.load(f)
github_retrieval_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
agent_card=github_retrieval_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(github_retrieval_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
Sau đó, hãy tạo máy chủ cho tác nhân đánh giá:
# $HOME/app/src/agents/evaluator/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as evaluator_agent
from .executor import EvaluatorAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/content_evaluator_agent_card.json", "r") as f:
card_data = json.load(f)
content_evaluator_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
agent_card=content_evaluator_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(content_evaluator_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
Và cuối cùng, đối với tác nhân gửi email:
# $HOME/app/src/agents/emailer/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as emailer_agent
from .executor import EmailerAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/emailer_agent_card.json", "r") as f:
card_data = json.load(f)
emailer_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
agent_card=emailer_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(emailer_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
9. Tạo Host Agent
Những gì chúng ta đã làm cho đến nay về cơ bản là biến mỗi tác nhân của chúng ta thành một API để tác nhân lưu trữ ping. Giờ đây, khi các tác nhân từ xa của chúng ta đã được tiếp xúc với các máy chủ A2A của riêng chúng, chúng ta cần xây dựng một tác nhân lưu trữ để khám phá và điều phối các tác nhân đó.
Chúng ta cần xây dựng một số khía cạnh khác nhau của máy chủ lưu trữ để đạt được mục tiêu này. Trước tiên, chúng ta cần tạo một cách để xây dựng từng ứng dụng cho mỗi tác nhân từ xa và máy chủ của chúng. Việc này được thực hiện bằng cách tạo một A2A Client factory để thiết lập các kết nối đến từng điểm cuối của tác nhân từ xa do máy chủ của chúng lưu trữ, trong đó máy chủ lưu trữ có thể khám phá Thẻ tác nhân. Bạn có thể khởi tạo các kết nối giữa máy chủ lưu trữ và từng tác nhân từ xa bằng một đối tượng RemoteAgentConnections.
touch $HOME/app/src/host/remote_agent_connection.py
touch $HOME/app/src/host/agent.py
touch $HOME/app/src/host/__init__.py
echo "from . import agent" >> $HOME/app/src/host/__init__.py
Sao chép quá trình triển khai sau đây của các kết nối tác nhân từ xa vào tệp src/host/remote_agent_connection.py:
# src/host/remote_agent_connection.py
import traceback
from a2a.client import (
Client,
ClientFactory,
)
from a2a.types import (
AgentCard,
Message,
SendMessageRequest,
Task,
TaskState,
)
class RemoteAgentConnection:
"""A class to hold the connections to the remote agents."""
def __init__(self, client_factory: ClientFactory, agent_card: AgentCard):
self.agent_client: Client = client_factory.create(agent_card)
self.card: AgentCard = agent_card
def get_agent(self) -> AgentCard:
return self.card
async def send_message(self, message: Message) -> Task | Message | None:
lastTask: Task | None = None
try:
async for event in self.agent_client.send_message(
SendMessageRequest(message=message)
):
if event.HasField('message'):
return event.message
# Build up the task from the streamed task and update events
if event.HasField('task'):
lastTask = event.task
elif lastTask and event.HasField('status_update'):
lastTask.status.CopyFrom(event.status_update.status)
elif lastTask and event.HasField('artifact_update'):
lastTask.artifacts.append(event.artifact_update.artifact)
if lastTask and self.is_terminal_or_interrupted(lastTask):
return lastTask
except Exception as e:
print('Exception found in send_message')
traceback.print_exc()
raise e
return lastTask
def is_terminal_or_interrupted(self, task: Task) -> bool:
return task.status.state in [
TaskState.TASK_STATE_COMPLETED,
TaskState.TASK_STATE_CANCELED,
TaskState.TASK_STATE_FAILED,
TaskState.TASK_STATE_REJECTED,
TaskState.TASK_STATE_INPUT_REQUIRED,
TaskState.TASK_STATE_AUTH_REQUIRED,
]
Giờ đây, chúng ta có thể thiết lập kết nối giữa một ứng dụng và một máy chủ thông qua một Ứng dụng và Thẻ đại lý nhất định. Đây sẽ là công cụ mà tác nhân lưu trữ sử dụng để thiết lập kết nối với các máy chủ tác nhân từ xa.
Hãy chuyển sang bước tạo chính tác nhân lưu trữ. Bắt đầu bằng cách sao chép đoạn mã sau vào tệp app/src/host/agent.py. Đây là quá trình khởi tạo lớp Pythonic yêu cầu các điểm cuối của tác nhân từ xa và một ứng dụng httpx tiêu chuẩn. Việc khởi tạo lớp tác nhân này sẽ thiết lập các kết nối thông qua các địa chỉ từ xa được cung cấp cho lớp đó.
import asyncio
import json
import os
import uuid
import httpx
from typing import Any
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import (
AgentCard,
Message,
Part,
Role,
Task,
TaskState,
)
from a2a.utils import TransportProtocol
from google.protobuf.json_format import MessageToDict
from google.adk import Agent
from google.adk.models import Gemini
from google.genai import types
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.tools.tool_context import ToolContext
from dotenv import load_dotenv
from .remote_agent_connection import RemoteAgentConnection
load_dotenv()
######################################
# --- Coordinator Agent Definition ---
######################################
class CoordinatorAgent:
"""
The Coordinator agent.
This is the agent responsible for sending tasks to agents.
"""
def __init__(
self,
remote_agent_addresses: list[str],
http_client: httpx.AsyncClient,
):
self.http_client = http_client
self.remote_agent_addresses = remote_agent_addresses
self.client_factory = None
self.remote_agent_connections: dict[str, RemoteAgentConnection] = {}
self.cards: dict[str, AgentCard] = {}
self.agents: str = ''
Khía cạnh tiếp theo của tác nhân lưu trữ là thiết lập kết nối với các tác nhân từ xa. Thay vì kết nối trong __init__ (chạy vào thời gian nhập mô-đun, trước khi vòng lặp sự kiện tồn tại), phương thức ensure_initialized sẽ hoãn công việc này cho đến khi nhân viên hỗ trợ thực sự được sử dụng. Thao tác này kiểm tra xem các kết nối đã được thiết lập hay chưa. Nếu chưa, thao tác này sẽ tạo ứng dụng A2A và kết nối song song với từng tác nhân từ xa. Sao chép đoạn mã này ngay bên dưới đoạn mã trước đó trong app/src/host/agent.py.
#####################################
# --- Remote Agent Initialization ---
#####################################
async def ensure_initialized(self):
if not self.remote_agent_connections:
config = ClientConfig(
httpx_client=self.http_client,
supported_protocol_bindings=[
TransportProtocol.JSONRPC,
TransportProtocol.HTTP_JSON,
],
)
self.client_factory = ClientFactory(config=config)
async with asyncio.TaskGroup() as task_group:
for address in self.remote_agent_addresses:
task_group.create_task(self.retrieve_card(address))
async def retrieve_card(self, address: str):
card_resolver = A2ACardResolver(self.http_client, address)
card = await card_resolver.get_agent_card()
# Use the actual address, not the hardcoded URL in the card JSON
for interface in card.supported_interfaces:
interface.url = address
remote_connection = RemoteAgentConnection(self.client_factory, card)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
agent_info = []
for card in self.cards.values():
agent_info.append(
json.dumps({"name": card.name, "description": card.description})
)
self.agents = "\n".join(agent_info)
Tiếp theo, chúng ta sẽ triển khai LLM Agent. Trong trường hợp này, chúng ta đang sử dụng một tác nhân ADK cho trình điều phối, yêu cầu các tiện ích bổ sung thông thường, chẳng hạn như hướng dẫn nhắc nhở, lệnh gọi lại và công cụ. Sao chép tất cả các thành phần này vào bên trong lớp tác nhân lưu trữ trong tệp app/src/host/agent.py. Đặt lệnh gọi lại sau vào lớp tác nhân lưu trữ. Trước khi tác nhân được kích hoạt đúng cách, lệnh gọi lại này sẽ khởi chạy tác nhân nếu tác nhân chưa được khởi chạy trong trạng thái của tác nhân.
############################
# -- Implement the ADK Agent
############################
# --- Before Model Callback ---
def before_model_callback(
self, callback_context: CallbackContext, llm_request
):
"""
A callback to set up the session state before the model processes the
request
"""
state = callback_context.state
if 'session_active' not in state or not state['session_active']:
if 'session_id' not in state:
state['session_id'] = str(uuid.uuid4())
state['session_active'] = True
Thành phần tiếp theo của tác nhân lưu trữ là hướng dẫn về câu lệnh. Vì đây là tác nhân điều phối của chúng ta, nên tác nhân này cần biết những tác nhân từ xa mà nó sẽ có quyền sử dụng, cũng như tác nhân hiện đang hoạt động để cho điều phối viên biết những gì đang diễn ra trong phiên. Thêm câu lệnh và hàm trợ giúp sau đây bên dưới lệnh gọi lại.
# --- Prompt ---
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_state(context)
return f"""
You are an expert orchestrator that can delegate user requests to the
appropriate remote agents to generate a GitHub research report.
**Your Goal:** To fulfill user requests for GitHub repository data, evaluate it, and optionally email a report.
**Workflow Steps:**
1. **Understand the User Request**:
- Identify repository names (e.g., "google/adk-python").
- Determine if the user wants Pull Requests, Issues, or both.
- Extract any specified email address for sending the report (e.g., "user@example.com").
- Note the number of PRs/issues requested per repository. If not specified, the Retrieval Agent has defaults.
2. **Retrieve Data (GitHub_Retrieval_Agent)**:
- Use the `send_message` tool to send a message to the "GitHub_Retrieval_Agent".
- Your message to the Retrieval Agent should clearly state which repositories to fetch data for, and specify if you need issues, pull requests, and the respective limits.
- Example message to Retrieval Agent: "Fetch 5 issues and 3 pull requests for google/adk-python."
3. **Evaluate and Summarize Data (Content_Evaluation_Agent)**:
- Once you receive the raw JSON data from the "GitHub_Retrieval_Agent", use the `send_message` tool to send this data to the "Content_Evaluation_Agent".
- Your message to the Evaluation Agent should include the raw JSON data you received.
- The Evaluation Agent will return a Markdown-formatted report.
4. **Email Report (Email_Agent - if email provided)**:
- If the user's initial request included an email address, use the `send_message` tool to send the Markdown report from the "Content_Evaluation_Agent" to the "Email_Agent".
- Your message to the Email Agent should include the report and the recipient's email address.
- Example message to Email Agent: "Send this report to user@example.com: [Markdown Report Content]".
5. **Respond to User**:
- Based on the outcome of the steps above, formulate a concise and informative response to the user.
- If a report was generated and emailed, confirm that. If only a report was generated, provide it directly to the user.
- If an email address was requested but the email failed to send, inform the user.
- If any step failed, inform the user about the failure.
**Available Tools:**
- `list_remote_agents()`: Use this to see what agents are available (though you already know their names for this workflow).
- `send_message(agent_name: str, message: str)`: Use this to interact with remote agents.
**Crucial Guidelines:**
- **Rely on Tools**: ALWAYS use `send_message` to interact with the remote agents. Do NOT attempt to perform the tasks yourself.
- **No Conversational Filler**: Only communicate the essential information to the remote agents or back to the user.
- **Error Handling**: If a remote agent returns an error, acknowledge it and try to provide a helpful message to the user.
Agents:
{self.agents}
Current agent: {current_agent['active_agent']}
"""
def check_state(self, context: ReadonlyContext):
state = context.state
if (
'context_id' in state
and 'session_active' in state
and state['session_active']
and 'agent' in state
):
return {'active_agent': f'{state["agent"]}'}
return {'active_agent': 'None'}
Giờ đến phần quan trọng nhất của tác nhân lưu trữ, đó là các công cụ. Máy chủ lưu trữ sẽ có quyền truy cập vào công cụ send_message() và công cụ list_remote_agents(). Công cụ list_remote_agent() đơn giản hơn, chỉ cần đọc các thẻ tác nhân của lớp và trả về một danh sách các từ điển chứa tên và nội dung mô tả của từng tác nhân. send_message() có phần nâng cao hơn. Nó có khả năng gửi tin nhắn hoặc tác vụ, truyền trực tuyến hoặc không truyền trực tuyến đến một tác nhân từ xa dựa trên tên tác nhân và chuỗi tin nhắn. Đặt đoạn mã này vào bên trong cùng một lớp python Host Agent bên dưới phần câu lệnh trong app/src/host/agent.py.
# --- Agent Tools ---
def list_remote_agents(self):
"""List the available remote agents you can use to delegate the task."""
if not self.remote_agent_connections:
return []
remote_agent_info = []
for card in self.cards.values():
remote_agent_info.append(
{'name': card.name, 'description': card.description}
)
return remote_agent_info
async def send_message(
self, agent_name: str, message: str, tool_context: ToolContext
):
"""Sends a task either streaming (if supported) or non-streaming.
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
message: The message to send to the agent for the task.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
await self.ensure_initialized()
if agent_name not in self.remote_agent_connections:
raise ValueError(f'Agent {agent_name} not found')
state = tool_context.state
state['agent'] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f'Client not available for {agent_name}')
task_id = state.get('task_id', None)
context_id = state.get('context_id', None)
message_id = state.get('message_id', None)
task: Task
if not message_id:
message_id = str(uuid.uuid4())
request_message = Message(
role=Role.ROLE_USER,
parts=[Part(text=message)],
message_id=message_id,
context_id=context_id,
task_id=task_id,
)
response = await client.send_message(request_message)
if isinstance(response, Message):
return await convert_parts(response.parts, tool_context)
task: Task = response # type: ignore
# Assume completion unless a state returns that isn't complete
state['session_active'] = not client.is_terminal_or_interrupted(task)
if task.context_id:
state['context_id'] = task.context_id
# Only reuse the task ID to continue a task that is waiting for input.
# Finished tasks can't take new messages, so the next call starts a new task.
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
state['task_id'] = task.id
else:
state['task_id'] = None
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.TASK_STATE_CANCELED:
# Open question, should we return some info for cancellation instead
raise ValueError(f'Agent {agent_name} task {task.id} is cancelled')
elif task.status.state == TaskState.TASK_STATE_FAILED:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.HasField('message'):
response.extend(
await convert_parts(task.status.message.parts, tool_context)
)
if task.artifacts:
for artifact in task.artifacts:
response.extend(
await convert_parts(artifact.parts, tool_context)
)
return response
Cuối cùng, chúng ta sẽ triển khai tác nhân ADK kết hợp tất cả các thành phần này. Đây là bộ não của tác nhân lớp điều phối, thực hiện câu lệnh được đưa ra bằng các công cụ mà tác nhân có thể sử dụng sau khi BeforeModelCallback chạy.
Đặt đoạn mã sau ngay bên dưới phần công cụ trong tệp app/src/host/agent.py bên dưới phần công cụ:
def create_agent(self) -> Agent:
"""Create an instance of the CoordinatorAgent."""
return Agent(
model=Gemini(
model='gemini-3.8-flash',
retry_options=types.HttpRetryOptions(attempts=5),
),
name='host',
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
description=(
'This coordinator agent orchestrates the retriever, evaluator, and emailer agents'
),
tools=[
self.send_message,
self.list_remote_agents,
],
)
Tác nhân cần một vài hàm trợ giúp cho công cụ send_message() để chuyển đổi các phần A2A thành văn bản và dữ liệu thô. Sao chép đoạn mã này BÊN NGOÀI lớp CoordinatorAgent:
##########################
# --- Helper Functions ---
##########################
async def convert_part(part: Part, tool_context: ToolContext):
"""Convert a part to text. Only text parts are supported."""
if part.HasField("text"):
return part.text
if part.HasField("data"):
return MessageToDict(part.data)
return f"Unknown type: {part}"
async def convert_parts(parts: list[Part], tool_context: ToolContext):
"""Convert parts to text."""
rval = []
for p in parts:
rval.append(await convert_part(p, tool_context))
return rval
Để tạo điều kiện thuận lợi cho việc kiểm thử và tương tác cục bộ thông qua adk web hoặc adk run, hãy thêm một quy trình khởi chạy root_agent ở cuối tệp app/src/host/agent.py:
root_agent = CoordinatorAgent(
remote_agent_addresses=[
os.getenv('RETRIEVAL_AGENT_URL', 'http://localhost:8001'),
os.getenv('EVALUATOR_AGENT_URL', 'http://localhost:8002'),
os.getenv('EMAILER_AGENT_URL', 'http://localhost:8003'),
],
http_client=httpx.AsyncClient(timeout=30),
).create_agent()
XIN CHÚC MỪNG! Bạn vừa tạo tác nhân lưu trữ A2A đầu tiên. Vì đã khởi tạo nó bằng biến root_agent, nên chúng ta có thể tương tác với nó thông qua adk run hoặc việc triển khai adk web cục bộ giống như các tác nhân từ xa khác. Sử dụng các lệnh sau để tương tác với người tổ chức:
cd $HOME/app/src/
uv run adk run host
Máy chủ lưu trữ chỉ có thể trả lời sau khi 3 máy chủ tác nhân từ xa đang chạy. Bạn sẽ thực hiện việc này ở bước tiếp theo. Cho đến lúc đó, người tổ chức có thể trả lời lời chào, nhưng mọi yêu cầu cần có nhân viên hỗ trợ từ xa đều sẽ thất bại với mã lỗi AgentCardResolutionError. Nhập exit để dừng máy chủ lưu trữ.
10. Kiểm thử cục bộ
Để kiểm thử toàn bộ hệ thống đa tác nhân, bạn phải khởi động các máy chủ tác nhân từ xa mà bạn muốn tác nhân lưu trữ có thể sử dụng. Tập lệnh này sẽ khởi động tất cả các máy chủ tác nhân cùng một lúc. Tạo tệp tập lệnh:
touch $HOME/app/start_agents.sh
cloudshell edit $HOME/app/start_agents.sh
Sao chép tập lệnh sau vào $HOME/app/start_agents.sh:
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to stop all background processes
cleanup() {
echo "Stopping agent servers..."
kill $(jobs -p) 2>/dev/null
wait
echo "All processes terminated."
}
# Call the cleanup function when the script exits or is interrupted
trap cleanup EXIT
# Start the agent servers in the background from the app directory
cd $HOME/app
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server &
uv run python3 -m src.agents.evaluator.server &
uv run python3 -m src.agents.emailer.server &
# Start the host agent (it reads the remote agent URLs from .env)
cd $HOME/app/src/
uv run adk run host
# Wait for all background processes to finish
wait
Sau đó, hãy chạy tập lệnh:
bash $HOME/app/start_agents.sh
Tập lệnh này sẽ khởi động cả 4 tác nhân (trình truy xuất, trình đánh giá, trình gửi email và máy chủ lưu trữ). Bạn sẽ thấy đầu ra nhật ký của từng máy chủ trong thiết bị đầu cuối. Sau khi máy chủ chạy, bạn có thể mở một cửa sổ dòng lệnh mới và khởi chạy giao diện adk web:
# In a new terminal
cd $HOME/app/src/
uv run adk web
Mở đường liên kết localhost xuất hiện trong cửa sổ dòng lệnh. Tại đây, bạn có thể tương tác trực tiếp với tác nhân lưu trữ bằng cách chọn tác nhân đó trong trình đơn thả xuống ở trên cùng bên trái.
11. Dọn dẹp
Để tránh bị tính phí liên tục, hãy xoá các tài nguyên bạn đã tạo trong lớp học lập trình này.
Dừng tất cả các máy chủ tác nhân đang chạy, sau đó xoá dự án nếu bạn đã tạo một dự án dành riêng cho lớp học lập trình này. Chuyển đến các thiết bị đầu cuối đã khởi chạy các tác nhân từ xa và chạy Ctrl+C, sau đó xoá dự án này trên Google Cloud:
gcloud projects delete <YOUR_PROJECT_ID>
12. Xin chúc mừng
Bạn đã xây dựng thành công một hệ thống đa tác nhân bằng Agent2Agent!
Kiến thức bạn học được
- Cách tạo các tác nhân ADK độc lập bằng các công cụ riêng
- Cách cung cấp danh tính có thể khám phá cho các tác nhân bằng Thẻ tác nhân
- Cách hiển thị các tác nhân thông qua máy chủ A2A
- Cách tạo tác nhân lưu trữ điều phối các tác nhân từ xa
- Cách giao thức A2A cho phép giao tiếp giữa các tác nhân được phát triển độc lập
Các bước tiếp theo
- Triển khai hệ thống lên Cloud Run để sử dụng trong thực tế
- Thêm nhiều tác nhân từ xa hơn để mở rộng khả năng của hệ thống
- Khám phá các chức năng truyền phát trực tiếp và thông báo đẩy trong A2A