Agent Development Kit (ADK) của Google: Hướng dẫn kèm dự án demo

Tổng quan dự án: Trình lập kế hoạch du lịch dùng AI với ADK và A2A

Dự án này xây dựng một trình lập kế hoạch du lịch có thể:

• Nhận điểm đến, ngày tháng và ngân sách làm đầu vào.
• Gọi ba tác tử riêng biệt:
• flight_agent:  Đề xuất lựa chọn chuyến bay.
• stay_agent: Tìm khách sạn trong phạm vi ngân sách.
• activities_agent: Gợi ý các hoạt động thú vị.
• Sau đó, dùng host_agent trung tâm để điều phối tất cả yêu cầu.
• Cuối cùng, dùng giao diện Streamlit để người dùng tương tác.

Tất cả các tác tử được lưu trữ dưới dạng các máy chủ FastAPI riêng và cung cấp endpoint /run. Việc giao tiếp được thực hiện thông qua client tương thích A2A dùng chung.

Lưu ý: Dự án này chạy hoàn toàn trên máy cục bộ của bạn để đơn giản, nhưng bạn có thể dễ dàng triển khai từng tác tử và UI lên các nền tảng đám mây như Render, Railway hoặc Google Cloud Run để truy cập mở rộng.

Bước 1: Yêu cầu tiên quyết

Hãy bắt đầu bằng cách cài đặt các thư viện sau:

pip install google-adk litellm fastapi uvicorn httpx pydantic openai streamlit

Tiếp theo, thiết lập khóa API OpenAI của bạn—bạn có thể thoải mái dùng nhà cung cấp mô hình khác. Để tìm hiểu cách thiết lập khóa API của OpenAI, tôi khuyến nghị hướng dẫn nhập môn về GPT-4o API.

export OPENAI_API_KEY="your_key_here"

Bước 2: Lược đồ và tiện ích dùng chung

Trước khi xây dựng các tác tử thông minh, chúng ta cần định nghĩa một ngôn ngữ chung để chúng giao tiếp. Trong thiết lập của chúng ta, điều này được thực hiện bằng cách:

• Một lược đồ dùng chung cho đầu vào (định nghĩa qua Pydantic)
• Một tiện ích REST client để gọi các tác tử
• Một trình bao REST server để tiêu chuẩn hóa endpoint /run trên tất cả tác tử

Những phần này được đặt trong thư mục shared/ và common/ để giữ cho mã mô-đun. Hãy xem từng phần.

Tạo tệp shared/schemas.py

Chúng ta định nghĩa lược đồ TravelRequest bằng Pydantic. Điều này đảm bảo tất cả tác tử thống nhất về cấu trúc của yêu cầu đến, bao gồm điểm đến, ngày đi và ngân sách người dùng.

from pydantic import BaseModel
class TravelRequest(BaseModel):
    destination: str
    start_date: str
    end_date: str
    budget: float

Lớp này giúp:

• Giữ đầu vào nhất quán cho mọi tác tử.
• Thêm xác thực tự động với FastAPI.
• Đơn giản hóa tái sử dụng mã.

Tạo tệp common/a2a_client.py

Tiện ích async nhẹ này cho phép bất kỳ tác tử nào (đặc biệt là host) gọi một tác tử khác bằng giao thức A2A thông qua endpoint /run.

import httpx
async def call_agent(url, payload):
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, timeout=60.0)
        response.raise_for_status()
        return response.json()

Tiện ích này gửi bất đồng bộ một yêu cầu POST tới endpoint /run của tác tử khác bằng httpx. Nó trả về phản hồi JSON đã phân tích và sẽ báo lỗi nếu yêu cầu thất bại.

Chúng ta sẽ dùng tiện ích này trong host agent để gọi flight_agent, stay_agent và activities_agent.

Tạo tệp common/a2a_server.py

Thay vì viết route FastAPI riêng cho từng tác tử, chúng ta tổng quát hóa bằng hàm create_app(agent), hàm này xử lý:

• Phục vụ tác tử tại /run
• Nhận yêu cầu du lịch
• Trả về phản hồi có cấu trúc

from fastapi import FastAPI
import uvicorn
def create_app(agent):
    app = FastAPI()
    @app.post("/run")
    async def run(payload: dict):
        return await agent.execute(payload)
    return app

Tiện ích này tạo một ứng dụng FastAPI với route /run tiêu chuẩn, ủy quyền thực thi cho tác tử được cung cấp.  Nó đảm bảo giao diện tác tử-đến-tác tử (A2A) nhất quán cho mọi dịch vụ bằng đầu vào JSON có cấu trúc.

Kết hợp lại, các thành phần dùng chung này giúp hệ thống đa tác tử của chúng ta dễ bảo trì, tái sử dụng và phù hợp với triết lý A2A của Google về nhắn tin liên tác tử đơn giản, có cấu trúc.

Bước 3: Xây dựng hệ thống đa tác tử với ADK và A2A

Giờ khi đã có hợp đồng và tiện ích dùng chung, hãy bắt đầu xây dựng từng tác tử. Để biến hệ thống thực sự mô-đun và đa tác tử, chúng ta sẽ dùng giao thức A2A của Google—một giao diện dựa trên HTTP đơn giản cho phép tác tử giao tiếp nhất quán và tương tác được.

A2A (Agent-to-Agent) cho phép phối hợp dạng cắm-là-chạy giữa các tác tử, dù đó là hàm Python cục bộ hay được lưu trữ qua mạng. Mỗi tác tử cung cấp endpoint /run với lược đồ chung và hoạt động như một dịch vụ.

Trong bản demo, chúng ta có bốn tác tử:

• host_agent: Điều phối mọi tác tử khác
• flight_agent: Tìm chuyến bay phù hợp
• stay_agent: Gợi ý chỗ ở
• activities_agent: Gợi ý các hoạt động địa phương hấp dẫn

Tất cả tác tử có cấu trúc tương tự, gồm 3 tệp và một thư mục con:

agents/
├── host_agent/
│   │   ├── agent.py              # Tùy chọn nếu logic host tối thiểu
│   │   ├── task_manager.py       # Gọi các tác tử khác và tổng hợp phản hồi
│   │   ├── __main__.py           # Khởi động FastAPI thông qua common/a2a_server.py
│   │   └── .well-known/
│   │       └── agent.json        # Metadata A2A Agent Card
├── flight_agent/
├── stay_agent/
└── activities_agent/

Mỗi tác tử dùng google.adk.agents.Agent, bộ bao mô hình LiteLlm và Runner để thực thi. Bắt đầu tạo các tệp sau trong thư mục activities_agent và lặp lại tương tự cho flight_agent và stay_agent.

Tạo tệp agent.py

Giờ hãy định nghĩa logic cho activities_agent, chịu trách nhiệm tạo ra các trải nghiệm địa phương hấp dẫn dựa trên lịch trình du lịch của người dùng. 

Bước 1: Import

Bắt đầu bằng cách import các mô-đun thiết yếu để cấu hình và chạy tác tử.

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import json

Tác tử này dùng các thành phần của Google ADK như Agent, ' Runner, ' và LiteLlm và quản lý trạng thái bằng InMemorySessionService. Thư viện Types được dùng để xây dựng prompt có cấu trúc.

Bước 2: Tác tử activities 

Giờ, chúng ta khởi tạo chính tác tử bằng lớp Agent của ADK.

activities_agent = Agent(
    name="activities_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Suggests interesting activities for the user at a destination.",
    instruction=(
        "Given a destination, dates, and budget, suggest 2-3 engaging tourist or cultural activities. "
        "For each activity, provide a name, a short description, price estimate, and duration in hours. "
        "Respond in plain English. Keep it concise and well-formatted."
    )
)

Tham số instruction định nghĩa system prompt để dẫn dắt hành vi của LLM. Dù ví dụ này dùng tiếng Anh đơn giản, bạn có thể điều chỉnh instruction để trả về JSON có cấu trúc cho việc phân tích dễ hơn.

Bước 3: Quản lý phiên

Tiếp theo, để theo dõi tương tác người dùng, chúng ta cấu hình Runner cùng thông tin phiên.

session_service = InMemorySessionService()
runner = Runner(
    agent=activities_agent,
    app_name="activities_app",
    session_service=session_service
)
USER_ID = "user_activities"
SESSION_ID = "session_activities"

Runner quản lý việc thực thi tác tử cho một phiên ứng dụng cụ thể. InMemorySessionService lưu ngữ cảnh trong bộ nhớ. Sau đó, chúng ta định nghĩa user và session ID. Trong môi trường sản xuất, các giá trị này có thể động hoặc theo người dùng. Cách này đảm bảo có một phiên ADK mới trước khi gửi bất kỳ prompt nào tới tác tử LLM.

Bước 4: Thực thi logic của tác tử

Hàm execute() xử lý yêu cầu đến, xây dựng prompt, gọi mô hình và phân tích đầu ra.

async def execute(request):
    session_service.create_session(
        app_name="activities_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"User is flying to {request['destination']} from {request['start_date']} to {request['end_date']}, "
        f"with a budget of {request['budget']}. Suggest 2-3 activities, each with name, description, price estimate, and duration. "
        f"Respond in JSON format using the key 'activities' with a list of activity objects."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            response_text = event.content.parts[0].text
            try:
                parsed = json.loads(response_text)
                if "activities" in parsed and isinstance(parsed["activities"], list):
                    return {"activities": parsed["activities"]}
                else:
                    print("'activities' key missing or not a list in response JSON")
                    return {"activities": response_text}  # fallback to raw text
            except json.JSONDecodeError as e:
                print("JSON parsing failed:", e)
                print("Response content:", response_text)
                return {"activities": response_text}  # fallback to raw text

Hàm execute() xây dựng prompt một cách động sử dụng các tham số trong yêu cầu đến như điểm đến, ngày tháng, ngân sách. Hoạt động bên trong gồm:

• Prompt hướng dẫn mô hình trả về một đối tượng JSON có cấu trúc chứa danh sách activities.
• Tạo đối tượng Content và truyền cho ADK Runner, chờ bất đồng bộ phản hồi cuối cùng qua generator streaming.
• Khi nhận phản hồi cuối, tác tử trích xuất văn bản thô và cố gắng phân tích thành JSON.
• Nếu phân tích thành công và có khóa activities như mong đợi, dữ liệu có cấu trúc sẽ được trả về.
• Nếu khóa thiếu hoặc sai cấu trúc, phương án dự phòng là trả về văn bản thô để UI vẫn có nội dung hiển thị.
• Cách xử lý hai hướng này đảm bảo suy giảm duyên dáng khi LLM trả về văn bản thuần thay vì JSON có cấu trúc.

Chiến lược này cải thiện độ bền và trải nghiệm người dùng, đặc biệt khi đầu ra mô hình thay đổi nhẹ do nhiệt độ hoặc cách hiểu prompt.

Tạo tệp task_manager.py

Sau khi định nghĩa logic execute() trong agent.py, giờ chúng ta nối nó vào thiết lập máy chủ tương thích ADK bằng task_manager.py.

from .agent import execute
async def run(payload):
    return await execute(payload)

Tệp này hoạt động như một lớp mỏng bao quanh hàm execute() đã định nghĩa trước đó. Nó giúp phương thức run() sẵn sàng cho các mô-đun bên ngoài, đặc biệt là script máy chủ trong __main__.py.

Tạo tệp __main__.py

Tệp __main__.py khởi chạy máy chủ FastAPI trên cổng 8003, phục vụ tác tử tại endpoint /run.

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8003)

Cụ thể đang diễn ra:

• create_app() (từ common/a2a_server.py) bọc tác tử của chúng ta trong một giao diện FastAPI tương thích A2A tiêu chuẩn.
• Chúng ta xây dựng động một đối tượng có phương thức execute() để ADK có thể gọi run() đúng cách.
• Phân tách này cho phép duy trì giao diện API không trạng thái trong khi tái sử dụng logic cốt lõi của tác tử.

Tạo tệp .well-known/agent.json

Chúng ta dùng tệp JSON này để mô tả danh tính và mục đích của tác tử theo giao thức A2A (Agent-to-Agent).

{
    "name": "activity_agent",
    "description": "Agent providing activity details."
  }

Lưu ý: Mặc dù tệp .well-known/agent.json không được các tác tử trực tiếp sử dụng trong dự án này, nó tuân thủ đặc tả A2A và quan trọng cho việc khám phá, tự kiểm tra và khả năng tương thích trong tương lai với các bộ điều phối như LangGraph, CrewAI hoặc registry tác tử của Google.

Logic tương tự cũng được dùng cho flight_agent và stay_agent.

Bước 4: Điều phối với host_agent

host_agent đóng vai trò là bộ lập kế hoạch trung tâm cho bản demo. host_agent minh họa mô hình controller trong hệ thống đa tác tử. Nó tách biệt việc ra quyết định và thực thi, cho phép mỗi tác tử phía dưới tập trung vào lĩnh vực của mình trong khi tập trung hóa logic điều phối. Cách này không chỉ đơn giản hóa kiểm thử và mở rộng, mà còn phản chiếu kiến trúc microservice thực tế trong hệ thống phân tán.

Tác tử này gửi cùng một payload tới cả ba tác tử thông qua API /run mà chúng cung cấp và trộn kết quả. Hãy thêm các tệp sau vào thư mục host_agent.

Tạo tệp agent.py

Bắt đầu với các import cơ bản.

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

Khối import này mang vào tất cả các khối xây dựng cốt lõi cần thiết để định nghĩa và chạy tác tử dựa trên LLM bằng Google ADK: lớp Agent, bộ bao LLM nhẹ, Runner để xử lý thực thi và quản lý phiên trong bộ nhớ.

host_agent = Agent(
    name="host_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Coordinates travel planning by calling flight, stay, and activity agents.",
    instruction="You are the host agent responsible for orchestrating trip planning tasks. "
                "You call external agents to gather flights, stays, and activities, then return a final result."
)
session_service = InMemorySessionService()
runner = Runner(
    agent=host_agent,
    app_name="host_app",
    session_service=session_service
)
USER_ID = "user_host"
SESSION_ID = "session_host"

Đoạn mã trên định nghĩa một tác tử ADK cấp cao chịu trách nhiệm điều phối toàn bộ kế hoạch chuyến đi. Dù chúng ta chưa gọi các tác tử con từ LLM trong hiện thực này, system prompt thiết lập vai trò cho lần mở rộng sau, nơi LLM có thể xử lý tool use và meta-reasoning.

async def execute(request):
    # Ensure session exists
    session_service.create_session(
        app_name="host_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"Plan a trip to {request['destination']} from {request['start_date']} to {request['end_date']} "
        f"within a total budget of {request['budget']}. Call the flights, stays, and activities agents for results."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            return {"summary": event.content.parts[0].text}

Hàm execute() này đóng vai trò điểm vào chính của LLM trong host agent. Nó:

• Khởi tạo phiên (hỗ trợ bộ nhớ nếu cần)
• Xây dựng động prompt người dùng
• Gửi tới mô hình bằng phương thức runner.run_async() của ADK
• Cuối cùng, chờ và trích xuất phản hồi cuối cùng

Tạo tệp task_manager.py

Trình quản lý tác vụ thực thi logic điều phối bằng cách gọi các tác tử từ xa và xử lý toàn bộ quy trình lập kế hoạch chuyến đi. Về mặt triển khai, chúng ta định nghĩa URL dịch vụ cho từng tác tử con. Các endpoint này tuân theo giao thức A2A /run và mong đợi lược đồ JSON TravelRequest` dùng chung.

from common.a2a_client import call_agent
FLIGHT_URL = "http://localhost:8001/run"
STAY_URL = "http://localhost:8002/run"
ACTIVITIES_URL = "http://localhost:8003/run"

Giờ, chúng ta định nghĩa payload.

async def run(payload):
    #Print what the host agent is sending
    print("Incoming payload:", payload)
    flights = await call_agent(FLIGHT_URL, payload)
    stay = await call_agent(STAY_URL, payload)
    activities = await call_agent(ACTIVITIES_URL, payload)
    # Log outputs
    print("flights:", flights)
    print("stay:", stay)
    print("activities:", activities)
    # Ensure all are dicts before access
    flights = flights if isinstance(flights, dict) else {}
    stay = stay if isinstance(stay, dict) else {}
    activities = activities if isinstance(activities, dict) else {}
    return {
        "flights": flights.get("flights", "No flights returned."),
        "stay": stay.get("stays", "No stay options returned."),
        "activities": activities.get("activities", "No activities found.")
    }

Hàm này sử dụng helper call_agent() để gửi payload tới từng dịch vụ phía dưới và ghi log đầu vào/ra nhằm tăng khả năng quan sát trong giai đoạn phát triển. Đây về bản chất là nơi chứa logic điều phối thực sự.

Tạo tệp __main__.py

Tệp __main__.py đóng vai trò điểm vào cho máy chủ FastAPI bọc host agent. 

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8000)

Tệp chính thực hiện:

• Dùng create_app() từ common/a2a_server.py để tạo ứng dụng FastAPI với endpoint /run tiêu chuẩn.
• Sau đó truyền vào một đối tượng giống tác tử với phương thức execute() bên trong ủy quyền cho hàm task_manager.run().
• Cuối cùng, khởi chạy máy chủ FastAPI bằng uvicorn trên cổng chỉ định (thường là 8000).

Cách này giúp giao diện host agent nhất quán với các tác tử phía dưới khác, duy trì tính thống nhất trên toàn hệ thống.

Tạo tệp .well-known/agent.json

Tệp này thể hiện mẫu đa tác tử kinh điển, nơi một nút trung tâm ủy thác và kết hợp tác vụ. 

{
  "name": "host_agent",
  "description": "Coordinates travel planning among specialized agents."
}

Dù là tùy chọn, việc bao gồm tệp này trong mọi thư mục tác tử là một thực hành tốt, như đã giải thích trước đó.

Bước 5: Xây dựng UI với Streamlit

Cuối cùng, hãy xây dựng một ứng dụng đơn giản nơi người dùng có thể nhập tùy chọn và nhận lịch trình có cấu trúc. Bắt đầu bằng cách tạo tệp travel_ui.py ở thư mục gốc và thêm đoạn mã sau.

import streamlit as st
import requests

Chúng ta import các thư viện cơ bản như Streamlit và requests để hỗ trợ UI.

st.set_page_config(page_title="ADK-Powered Travel Planner", page_icon="✈️")
st.title("🌍 ADK-Powered Travel Planner")
origin = st.text_input("Where are you flying from?", placeholder="e.g., New York")
destination = st.text_input("Destination", placeholder="e.g., Paris")
start_date = st.date_input("Start Date")
end_date = st.date_input("End Date")
budget = st.number_input("Budget (in USD)", min_value=100, step=50)
if st.button("Plan My Trip ✨"):
    if not all([origin, destination, start_date, end_date, budget]):
        st.warning("Please fill in all the details.")
    else:
        payload = {
            "origin": origin,
            "destination": destination,
            "start_date": str(start_date),
            "end_date": str(end_date),
            "budget": budget
        }
        response = requests.post("http://localhost:8000/run", json=payload)
        if response.ok:
            data = response.json()
            st.subheader("✈️ Flights")
            st.markdown(data["flights"])
            st.subheader("🏨 Stays")
            st.markdown(data["stay"])
            st.subheader("🗺️ Activities")
            st.markdown(data["activities"])
        else:
            st.error("Failed to fetch travel plan. Please try again.")

Ứng dụng Streamlit cung cấp một UI thân thiện để tương tác với trình lập kế hoạch du lịch đa tác tử được xây dựng bằng ADK. Một vài điểm trong đoạn mã trên:

• Dùng text_input, date_input và number_input để thu thập điểm đi, điểm đến, ngày tháng và ngân sách.
• Khi nhấp “Plan My Trip”, ứng dụng kiểm tra tính hợp lệ để đảm bảo không trường nào bị bỏ trống.
• Nếu hợp lệ, ứng dụng tạo payload JSON và gửi yêu cầu POST tới host_agent tại http://localhost:8000/run.
• host_agent gọi tất cả tác tử con (chuyến bay, chỗ ở, hoạt động), tổng hợp phản hồi và trả về kế hoạch du lịch thống nhất.
• Phản hồi được phân tích và hiển thị bằng st.markdown() dưới các tiêu đề riêng cho chuyến bay, chỗ ở và hoạt động.
• Nếu backend gặp lỗi, hiển thị thông báo dự phòng bằng st.error().

Giờ hãy chạy các lệnh sau trên terminal máy bạn:

uvicorn agents.host_agent.__main__:app --port 8000 &
uvicorn agents.flight_agent.__main__:app --port 8001 &
uvicorn agents.stay_agent.__main__:app --port 8002 &      
uvicorn agents.activities_agent.__main__:app --port 8003 &
streamlit run travel_ui.py

Khi người dùng nhấp “Plan My Trip,” host agent sẽ tiếp quản, kích hoạt các tác tử và hiển thị kết quả trên UI:

Cấu trúc tệp tổng thể của bạn sẽ trông như sau:

ADK_demo/
├── agents/
│   ├── host_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py      
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json   
│   │
│   ├── flight_agent/
│   │   ├── agent.py        
│   │   ├── task_manager.py    
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json        
│   │
│   ├── stay_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py       
│   │   ├── __main__.py           
│   │   └── .well-known/
│   │       └── agent.json     
│   │
│   └── activities_agent/
│       ├── agent.py              
│       ├── task_manager.py       
│       ├── __main__.py           
│       └── .well-known/
│           └── agent.json       
│
├── common/
│   ├── a2a_client.py             # Tiện ích gửi yêu cầu tới các tác tử khác
│   └── a2a_server.py             # Mẫu máy chủ FastAPI tương thích A2A dùng chung
│
├── shared/
│   └── schemas.py                # Lược đồ Pydantic dùng chung
│
├── streamlit_app.py             # Giao diện người dùng để nhập dữ liệu và hiển thị phản hồi
├── requirements.txt           
└── README.md                    

Và thế là xong! Tôi đã tổng hợp mọi thứ chúng ta đã xây dựng trong dự án GitHub này.

Kết luận

Chỉ với vài ứng dụng FastAPI và các tác tử ADK, chúng ta đã xây dựng một trình lập kế hoạch du lịch cộng tác có thể:

• Giao tiếp qua giao thức A2A
• Dùng các tác tử LLM cho chuyến bay, chỗ ở và hoạt động
• Tổng hợp và hiển thị kết quả trong UI Streamlit gọn gàng

Dù tất cả tác tử dùng cùng một mô hình bên dưới, hệ thống này vẫn mang hành vi đa tác tử, tức là tác tử có vai trò riêng, trách nhiệm tách biệt và giao tiếp có cấu trúc.

Dưới đây là một vài tài nguyên để bắt đầu:

• A2A GitHub
• Xây dựng bot thời tiết với ADK
• Tài liệu kỹ thuật A2A