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 chúng dưới dạng máy chủ A2A
- Cách tạo một 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
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ẽ 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 Google Cloud.
- Đảm bảo rằng 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ừ bảng điều khiển Google Cloud, hãy nhấp vào Kích hoạt Cloud Shell trong bảng điều khiển Google Cloud.
Thao tác này sẽ khởi chạy một phiên trong ngăn dưới cùng của bảng điều khiển Cloud của bạn.
Để 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.
Định cấu hình môi trường
Bắt đầu bằng cách chạy lệnh sau trong cửa sổ dòng lệnh để 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:
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 GCP và khu vực GCP vào tệp .env mới. 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. Liên tục, bạn có thể 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 .env
rồi 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_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=<your-gcp-project-region>
# --- 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>
MCP_SERVER_HOST=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
Bây giờ, bạn đã điền các biến môi trường, chúng ta phải định cấu hình môi trường uv. 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,<3.14"
dependencies = [
"resend>=2.21.0",
"uvicorn>=0.40.0",
"a2a-sdk>=0.3.26,<0.4.0",
"google-adk[a2a]>=1.27.0,<1.30.0",
"google-generativeai>=0.8.4",
"httpx>=0.28.1",
"pydantic>=2.12.5, <3.0.0",
"python-dotenv>=1.2.0",
"nest-asyncio>=1.6.0",
]
[project.optional-dependencies]
test = [
"pytest>=8.3.2",
"pytest-asyncio>=0.23.8",
"pytest-mock>=3.14.0",
"respx>=0.21.0",
]
[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/"]
Tạo môi trường ảo
Bây giờ, trong thư mục app mà bạn vừa tạo, hãy chạy tập lệnh bash sau trong thiết bị đầu cuối. Thao tác này sẽ thiết lập môi trường ảo Python và cài đặt tất cả các phần phụ thuộc cần thiết từ tệp pyproject.toml.
# Ensure you are in the 'app' directory before running this
# Exit immediately if a command exits with a non-zero status.
set -e
# 1. Install uv, the Python package manager used for this project
echo "Installing uv..."
if ! command -v uv &> /dev/null
then
pip install uv
else
echo "uv is already installed."
fi
# 2. Create and activate virtual environment
echo "Setting up virtual environment..."
# --clear ensures you start with a fresh environment
uv venv --clear
# 3. Install dependencies
echo "Installing Python dependencies..."
uv sync
echo "Installation and setup complete."
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 các 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.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-2.5-flash",
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 để chạy uv run adk web TRONG THƯ MỤC **/agents/:
cd $HOME/app/src/agents
uv run adk run retriever
Mở đường liên kết localhost trong thiết bị đầu cuối "uv run adk web" rồi chọn tác nhân để dùng thử.
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 google.adk.agents.llm_agent import Agent
# --- 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-2.5-flash",
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 để chạy uv run adk web TRONG THƯ MỤC **/agents/:
cd $HOME/app/src/agents
uv run adk run evaluator
Mở đường liên kết localhost trong thiết bị đầu cuối và chọn tác nhân đánh giá để dùng thử.
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 email chứa 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 resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
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 = f"<div style='font-family: sans-serif; white-space: pre-wrap;'>{body}</div>"
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-2.5-flash",
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 để chạy uv run adk web TRONG THƯ MỤC **/agents/:
cd $HOME/app/src/agents
uv run adk run emailer
Mở đường liên kết localhost trong thiết bị đầu cuối và chọn tác nhân để dùng thử.
6. Tạo thẻ tác nhân
Chúng ta vừa xem qua 3 câu chuyện về 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 trợ lý này đều được xuất bản và đăng nhập vào Thư viện trợ lý 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 nhận được 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 loạt các tác nhân từ xa là cung cấp cho mỗi tác nhân từ xa cần thiết một Thẻ tác nhân. Hãy coi đây là danh thiếp cho trợ lý của bạn. Đây là các tệp JSON dùng để giúp tác nhân lưu trữ có thể 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.
Trước tiên, hãy chạy lệnh sau từ gốc 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 thư mục app/cards/github_retrieval_agent_card.json. Bạn có thể truy cập vào tệp này:
{
"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.",
"url": "http://localhost:8001",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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, hãy dùng lệnh sau để 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.",
"url": "http://localhost:8002",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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 tệp đó 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.",
"url": "http://localhost:8003",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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 trình thực thi tác nhân từ xa
Để nhân viên hỗ trợ từ xa có thể được nhân viên hỗ trợ chính "gọi" hoặc "trò chuyện", mỗi nhân viên hỗ trợ từ xa cần có một trình thực thi nhân viên hỗ trợ mà máy chủ A2A có thể nhìn thấy, được xác định bằng Thẻ nhân viên hỗ trợ. 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 nhiệm vụ hoặc thông báo, hoặc huỷ hoàn toàn nhiệm 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 tác vụ 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 TextPart. 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.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, Part
from a2a.utils import new_agent_text_message, new_task
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(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
if context.call_context:
user_id = context.call_context.user.user_name
else:
user_id = "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.working,
new_agent_text_message(
self.status_message,
task.context_id,
task.id
),
)
# Process with ADK agent
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(root=TextPart(text=response_text))],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.failed,
new_agent_text_message(f"Error: {e!s}", task.context_id, task.id),
final=True,
)
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 nhu cầu và khả năng cụ thể của tác nhân. Trong trường hợp này, 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 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 trường url của thẻ đại lý.
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 a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=github_retrieval_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server.build(), 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 a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=content_evaluator_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server.build(), 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 a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=emailer_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
9. Tạo Host Agent
Những gì chúng tôi đã làm cho đến nay về cơ bản là biến mỗi tác nhân của chúng tôi 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 đó.
Có một số khía cạnh khác nhau của máy chủ lưu trữ mà chúng ta cần xây dựng để đạ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 nhà máy A2A Client để 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ữ, nơi máy chủ 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,
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(message):
if isinstance(event, Message):
return event
if self.is_terminal_or_interrupted(event[0]):
return event[0]
lastTask = event[0]
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.completed,
TaskState.canceled,
TaskState.failed,
TaskState.input_required,
TaskState.unknown,
]
Giờ đây, chúng tôi có thể thiết lập kết nối giữa ứng dụng và máy chủ thông qua một Khách hàng và Thẻ nhân viên hỗ trợ 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,
TextPart,
TransportProtocol,
)
from google.adk import Agent
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 tại thời điểm nhập mô-đun, trước khi có vòng lặp sự kiện), 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_transports=[
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()
card.url = address # Use the actual address, not the hardcoded URL in the card JSON
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 tạo tác nhân nếu tác nhân chưa được khởi tạo 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'}
Bây giờ là 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.user,
parts=[Part(root=TextPart(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
state['task_id'] = task.id
if task.status.state == TaskState.input_required:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.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.failed:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.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 lời nhắc đượ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-2.5-flash',
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 isinstance(part.root, TextPart):
return part.root.text
if part.root.kind == "data":
return part.root.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 uv run 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 việc triển khai web adk chạy uv 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
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. Thao tác này sẽ khởi động tất cả các máy chủ tác nhân cùng một lúc.
# Make sure you have activated your virtual environment first!
# source .venv/bin/activate
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to kill all background processes
cleanup() {
echo "Caught signal, terminating background processes..."
# The negative PID kills the entire process group
kill -TERM -$$
wait
echo "All processes terminated."
exit
}
# Trap TERM, INT, and EXIT signals and call the cleanup function
trap cleanup TERM INT EXIT
# Activate virtual environment
uv sync
# Start the agent servers in the background
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server --port 8001 &
uv run python3 -m src.agents.evaluator.server --port 8002 &
uv run python3 -m src.agents.emailer.server --port 8003 &
cd $HOME/app/src/
RETRIEVER_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
uv run adk run host
# Wait for all background processes to finish
wait
Lệnh bash này sẽ khởi động cả 4 máy chủ 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ý từ mỗi máy chủ trong thiết bị đầu cuối. Sau khi các máy chủ đang chạy, bạn có thể mở một cửa sổ dòng lệnh mới (đảm bảo bạn đang ở trong cùng một thư mục app và môi trường ảo đã được kích hoạt) rồi khởi chạy giao diện web adk chạy uv:
# In a new terminal, from the 'app' directory
source .venv/bin/activate
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 đã 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 trên đám mây này của Google:
gcloud projects delete ${GOOGLE_CLOUD_PROJECT}
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 một 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 các chức năng của hệ thống
- Khám phá các tính năng phát trực tuyến và thông báo đẩy trong A2A