构建分布式 AI 智能体

# researcher/app/agent.py
from google.adk.agents import Agent
from google.adk.tools import google_search

researcher = Agent(
    name="researcher",
    model="gemini-2.5-flash",
    description="Gathers information on a topic using Google Search.",
    instruction="""
    You are an expert researcher. Your goal is to find comprehensive information.
    Use the `google_search` tool to find relevant information.
    Summarize your findings clearly.
    """,
    tools=[google_search],
)

# judge/app/agent.py
from pydantic import BaseModel, Field
from typing import Literal

class JudgeFeedback(BaseModel):
    status: Literal["pass", "fail"] = Field(
        description="Whether the research is sufficient ('pass') or needs more work ('fail')."
    )
    feedback: str = Field(
        description="Detailed feedback on what is missing."
    )

judge = Agent(
    name="judge",
    model="gemini-2.5-flash",
    description="Evaluates research findings.",
    instruction="""
    You are a strict editor. Evaluate the findings.
    If they are missing key info, output status='fail' and provide feedback.
    """,
    output_schema=JudgeFeedback, # Enforce the contract!
)

# researcher/app/server.py
from fastapi import FastAPI
from a2a.server.apps import A2AFastAPIApplication
from app.agent import app as adk_app

# ... setup runner ...

# Create the A2A App wrapper
a2a_app = A2AFastAPIApplication(agent_card=agent_card, http_handler=request_handler)

app = FastAPI(lifespan=lifespan)

# Register routes: /.well-known/agent.json and /rpc
a2a_app.add_routes_to_app(app)

# orchestrator/app/agent.py
from google.adk.agents import LoopAgent, SequentialAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent

# Connect to the remote Researcher service
researcher = RemoteA2aAgent(
    name="researcher",
    agent_card="http://researcher-service:8000/.well-known/agent.json",
    description="Gathers information on a topic."
)

# Connect to the remote Judge service
judge = RemoteA2aAgent(
    name="judge",
    agent_card="http://judge-service:8000/.well-known/agent.json",
    description="Evaluates research findings."
)

# The Orchestrator manages the loop
research_loop = LoopAgent(
    name="research_loop",
    sub_agents=[researcher, judge, escalation_checker],
    max_iterations=3,
)

# The full pipeline
root_agent = SequentialAgent(
    name="course_creation_pipeline",
    sub_agents=[research_loop, content_builder],
)