インモビインモビ

gcloud services enable \
  aiplatform.googleapis.com \
  notebooks.googleapis.com \
  run.googleapis.com \
  cloudbuild.googleapis.com \
  cloudresourcemanager.googleapis.com

PROJECT_ID=$(gcloud config list --format 'value(core.project)')
gcloud workbench instances create agent-development \
  --project=$PROJECT_ID \
  --location=us-central1-a \
  --machine-type=e2-standard-2

%pip install --upgrade --user \
    google-adk==1.2.1 \
    google-cloud-aiplatform==1.96.0 \
    a2a-sdk==0.2.5

import IPython
app = IPython.Application.instance()
_ = app.kernel.do_shutdown(True)

import copy, json, os, pprint, uuid
import vertexai
from vertexai import agent_engines
from google.genai.types import Part, Content
from google.adk.agents.llm_agent import LlmAgent
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

[PROJECT_ID] = !gcloud config list --format 'value(core.project)'
[PROJECT_NUMBER] = !gcloud projects describe {PROJECT_ID} --format="value(projectNumber)"
LOCATION = 'us-central1'

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=f'gs://{PROJECT_ID}'
)

os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True'

class LocalApp:
    def __init__(self, agent, user_id='default_user'):
        self._agent = agent
        self._user_id = user_id
        self._runner = Runner(
            app_name=self._agent.name,
            agent=self._agent,
            artifact_service=InMemoryArtifactService(),
            session_service=InMemorySessionService(),
            memory_service=InMemoryMemoryService(),
        )
        self._session = None
        
    async def stream(self, query):
        if not self._session:
            self._session = await self._runner.session_service.create_session(
                app_name=self._agent.name,
                user_id=self._user_id,
                session_id=uuid.uuid4().hex,
            )
        content = Content(role='user', parts=[Part.from_text(text=query)])
        async_events = self._runner.run_async(
            user_id=self._user_id,
            session_id=self._session.id,
            new_message=content,
        )
        result = []
        async for event in async_events:
            if (event.content and event.content.parts):
                response = '\n'.join([p.text for p in event.content.parts if p.text])
                if response:
                    print(response)
                    result.append(response)
        return result

shopping_mall_info = '''
* 立地と外観:
  - 新宿駅南口から徒歩5分。賑やかな駅周辺から少し離れ、落ち着いた雰囲気のエリアに位置しています。
  - 緑豊かなオープンテラスが特徴的で、都会の中にありながらも自然を感じられる空間を提供しています。
  - 夜になると、間接照明が灯り、ロマンチックな雰囲気に包まれます。

* イベント:
  - 週末には、ジャズライブやアコースティックライブなどの音楽イベントがテラスで開催され、夜の雰囲気を盛り上げます。
  - 季節ごとのイルミネーションが美しく、訪れる人の目を楽しませます。
  - 地域住民向けのワークショップやマルシェなども開催され、地域との交流を深めています。

* テナント:
  - 個性的なセレクトショップ: 大手チェーン店だけでなく、オーナーのこだわりが詰まった隠れ家のようなセレクトショップが点在しています。
  - こだわりのレストランやカフェ: 「夜の帳」のように、落ち着いた雰囲気で質の高い食事や飲み物を楽しめるお店が集まっています。テラス席があるお店も多く、開放的な空間で食事を楽しめます。
  - 上質なライフスタイル雑貨店: 日常を豊かにする、デザイン性の高い雑貨や家具、オーガニックコスメなどを扱うお店があります。
  - アートギャラリーやミニシアター: 感性を刺激するアートや映画に触れることができるスペースがあります。
'''

coffee_shop_info = '''
* 店名: 夜の帳（よるのとばり）

* コンセプト: 一日の終わりに、静かに心と体を休ませる隠れ家のような喫茶店。落ち着いた照明と、心地よい音楽が流れる空間で、こだわりのコーヒーや軽食、デザートを提供します。

* 立地と外観:
  - 新宿スターライトテラス内の、メインフロアから少し奥まった静かな一角。3階の吹き抜けに面した見晴らしの良い場所
  - オレンジや琥珀色の暖色系間接照明が、店内から優しく漏れる。控えめな光で照らされた、筆記体のような上品な看板。

* メニュー:
  ** こだわりの珈琲:
    - 夜の帳ブレンド: 深煎りでコクがあり、ほんのりビターな大人の味わい。疲れた心に染み渡ります。
    - 月光の浅煎り: フルーティーな香りが特徴の、すっきりとした味わい。リフレッシュしたい時に。
    - カフェ・オ・レ: 丁寧に淹れたブレンドコーヒーと、温かいミルクの優しいハーモニー。
    - 水出し珈琲: じっくりと時間をかけて抽出した、まろやかで雑味のないアイスコーヒー。

  ** 軽食:
    - 厚切りトーストのたまごサンド: ふわふわの厚切りトーストに、自家製マヨネーズで和えた卵サラダをたっぷり挟みました。
    - 気まぐれキッシュ: シェフがその日の気分で作る、季節の野菜を使った焼き立てキッシュ。
    - 昔ながらのナポリタン: 喫茶店の定番メニュー。懐かしい味わいが心を満たします。
    - チーズと蜂蜜のトースト: 香ばしいトーストに、とろけるチーズと甘い蜂蜜が絶妙な組み合わせ。
'''

instruction = f'''
You are a friendly and energetic guide of the coffee shop "夜の帳".
Before giving an answer, say "とばりちゃんが答えるよ！".

[task]
Give an answer to the query based on the [shop information].

[shop information]
{coffee_shop_info}

[format instruction]
In Japanese. No markdowns.
'''

tobariChan_agent = LlmAgent(
    model='gemini-2.0-flash-001',
    name='TobariChan_agent',
    description=(
        'A friendly guide of the coffee shop "夜の帳".'
    ),
    instruction=instruction,
)

client = LocalApp(tobariChan_agent)

query = f'''
こんにちは！おすすめのコーヒーはありますか？
'''
_ = await client.stream(query)

とばりちゃんが答えるよ！おすすめのコーヒーですか？夜の帳には、こだわりのコーヒーがたくさんあるんですよ！

もし、一日のお疲れを癒やしたいなら、深煎りでコクのある「夜の帳ブレンド」がおすすめです。ほんのりビターな大人の味わいが、心に染み渡りますよ。

リフレッシュしたい気分でしたら、フルーティーな香りが特徴の「月光の浅煎り」はいかがでしょうか？すっきりとした味わいで、気分転換にぴったりです。

remote_agent = agent_engines.create(
    agent_engine=tobariChan_agent,
    display_name='tobariChan_agent',
    requirements=[
        'google-adk==1.2.1',
    ]
)

def tobari_chan_tool_vertexai(query: str) -> str:
    """
    Get an answer to a question regarding 夜の帳 from とばりちゃん

    Args:
        query: question
       
    Returns:
        str: An answer from とばりちゃん
    """
    import vertexai
    from vertexai import agent_engines

    PROJECT_ID = '[PROJECT_ID]' 
    AGENT_ID = '[AGENT_ID]'     
    LOCATION = 'us-central1'
    vertexai.init(project=PROJECT_ID, location=LOCATION)

    remote_agent = agent_engines.get(AGENT_ID)
    session = remote_agent.create_session(user_id='default_user')
    try:
        events = remote_agent.stream_query(
                    user_id='default_user',
                    session_id=session['id'],
                    message=query,
                 )
        result = []
        for event in events:
            if ('content' in event and 'parts' in event['content']):
                response = '\n'.join(
                    [p['text'] for p in event['content']['parts'] if 'text' in p]
                )
                if response:
                    result.append(response)
        return '\n'.join(result)

    finally:
        remote_agent.delete_session(
            user_id='default_user',
            session_id=session['id'],
        )

print(tobari_chan_tool_vertexai('こんにちは。あなたは誰ですか？'))

とばりちゃんが答えるよ！私は新宿スターライトテラスにある喫茶店「夜の帳」の案内人です。

instruction = f'''
You are a formal guide of the shopping mall "新宿スターライトテラス".
Before giving an answer, say "テラスガイドがお答えいたします。".

[Information]
* Name of the guide of "夜の帳" is "とばりちゃん".
* Name of the guide of "新宿スターライトテラス" is "テラスガイド".

[Tasks]
* Give an answer to the query based on the [mall information].
* For queries regarding "夜の帳", ask TobariChan_agent to get an answer
  and relay it to the user. Avoid relying on your own knowledge.

[mall information]
{shopping_mall_info}
'''

terraceGuide_agent_with_tool = LlmAgent(
    model='gemini-2.0-flash-001',
    name='TerraceGuide_agent',
    description=(
        'A formal guide of the shopping mall "新宿スターライトテラス".'
    ),
    instruction=instruction,
    tools=[
        tobari_chan_tool_vertexai,
    ],
)

client = LocalApp(terraceGuide_agent_with_tool)

query = '''
夜の帳はどんなお店ですか？
'''
_ = await client.stream(query)

テラスガイドがお答えいたします。
夜の帳についてですね。夜の帳については、とばりちゃんに聞く必要があります。少々お待ちください。

テラスガイドがお答えいたします。
とばりちゃんによると、夜の帳は、一日の終わりに心と体を休ませる隠れ家のような喫茶店です。
新宿スターライトテラスの３階にあり、落ち着いた照明と心地よい音楽が流れる空間で、こだわりのコーヒーや軽食、デザートをご用意しているとのことです。

%%bash
mkdir -p proxy_server
cat EOF > proxy_server/main.py
import json
import vertexai
from vertexai import agent_engines
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

PROJECT_ID = '[PROJECT_ID]' 
AGENT_ID = '[AGENT_ID]'     
LOCATION = 'us-central1'

vertexai.init(project=PROJECT_ID, location=LOCATION)

app = FastAPI()

class QueryItem(BaseModel):
    query: str
    
async def event_generator(request, query):
    remote_agent = agent_engines.get(AGENT_ID)
    session = remote_agent.create_session(user_id='default_user')
    try:
        events = remote_agent.stream_query(
                    user_id='default_user',
                    session_id=session['id'],
                    message=query,
                 )
        for event in events:
            if await request.is_disconnected():
                break
            yield json.dumps(event)
    finally:
        remote_agent.delete_session(
            user_id='default_user',
            session_id=session['id'],
        )

@app.get("https://zenn.dev/")
def read_root():
    return {'message': 'Tobari-chan agent.'}

@app.post('/stream')
async def stream_events(item: QueryItem, request: Request):
    return StreamingResponse(event_generator(request, item.query),
                             media_type='text/event-stream')
EOF


cat EOF > proxy_server/requirements.txt
fastapi
uvicorn
google-adk==1.2.1
google-cloud-aiplatform==1.96.0
EOF


cat EOF > proxy_server/Dockerfile
FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
EOF

REPO_NAME = 'cloud-run-source-deploy'
REPO = f'{LOCATION}-docker.pkg.dev/{PROJECT_ID}/{REPO_NAME}'
!gcloud artifacts repositories create {REPO_NAME} \
   --repository-format docker --location {LOCATION}

SERVICE_NAME = 'tobari-chan-proxy-server'

!cd proxy_server && gcloud builds submit --tag {REPO}/{SERVICE_NAME}
!gcloud run deploy {SERVICE_NAME} \
   --image {REPO}/{SERVICE_NAME} \
   --platform managed \
   --region {LOCATION} \
   --no-allow-unauthenticated 2>&1 | cat

Creating temporary archive of 3 file(s) totalling 1.6 KiB before compression.
...(中略)...
Service URL: https://tobari-chan-proxy-server-879055303739.us-central1.run.app

def tobari_chan_tool_proxy(query: str) -> str:
    """
    Get an answer to a question regarding 夜の帳 from とばりちゃん

    Args:
        query: question
       
    Returns:
        str: An answer from とばりちゃん
    """
    import requests
    import json
    import google.auth.transport.requests
    import google.oauth2.id_token

    PROXY_URL = '[SERVICE_URL]' 
    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, PROXY_URL)
    headers = {
        'Authorization': f'Bearer {id_token}',
        'Content-Type': 'application/json'
    }
    payload = {'query': query}

    result = []
    with requests.post(f'{PROXY_URL}/stream',
                       headers=headers,
                       json=payload,
                       stream=True) as server_response:
        server_response.raise_for_status()
        for line in server_response.iter_lines(decode_unicode=True):
            event = json.loads(line)
            if ('content' in event and 'parts' in event['content']):
                response = '\n'.join(
                    [p['text'] for p in event['content']['parts'] if 'text' in p]
                )
                if response:
                    result.append(response)
        return '\n'.join(result)

print(tobari_chan_tool_proxy('おすすめのコーヒーは？'))

とばりちゃんが答えるよ！夜の帳ブレンドは、深煎りでコクがあって、ほんのりビターな大人の味わいだよ。
疲れた心に染み渡るから、ぜひ試してみてね！

%%bash
mkdir -p a2a_server
cat EOF >a2a_server/main.py
import os
import vertexai
from vertexai import agent_engines

from a2a.server.agent_execution import AgentExecutor
from a2a.server.tasks import TaskUpdater, InMemoryTaskStore
from a2a.types import (
    Task, TextPart, UnsupportedOperationError,
    AgentCapabilities, AgentCard, AgentSkill,
)
from a2a.utils.errors import ServerError
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler

PROJECT_ID = '[PROJECT_ID]'
AGENT_ID = '[AGENT_ID]'
LOCATION = 'us-central1'

vertexai.init(project=PROJECT_ID, location=LOCATION)


class AdkAgent:
    SUPPORTED_CONTENT_TYPES = ['text', 'text/plain']

    def __init__(self, remote_agent):
        self._remote_agent = remote_agent

    async def stream(self, updater, query):
        session = self._remote_agent.create_session(user_id='default_user')
        try:
            events = self._remote_agent.stream_query(
                user_id='default_user',
                session_id=session['id'],
                message=query,
            )
            text_parts = []        
            for event in events:
                if text_parts: 
                    updater.new_agent_message(text_parts)
                text_parts = []
                if 'content' in event and 'parts' in event['content']:
                    for part in event['content']['parts']:
                        if 'text' in part:
                            text_parts.append(TextPart(text=part['text']))
            
            updater.add_artifact(text_parts)
            updater.complete()
        finally:
            self._remote_agent.delete_session(
                user_id='default_user',
                session_id=session['id'],
            )


class AdkAgentExecutor(AgentExecutor):
    def __init__(self, adk_agent: AdkAgent):
        self.adk_agent = adk_agent

    async def cancel(self, request, event_queue):
        raise ServerError(error=UnsupportedOperationError('Cancel operation is not supported.'))

    async def execute(self, context, event_queue):
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        if not context.current_task:
            updater.submit()
        updater.start_work()
        query = context.get_user_input()
        await self.adk_agent.stream(updater, query)


def create_agent_card(
    base_url, supported_content_types, agent_name, agent_description):
    
    capabilities = AgentCapabilities(streaming=True)
    skill = AgentSkill(
        id=f"{agent_name.lower().replace(' ', '_')}_skill",
        name=f'{agent_name} Skill',
        description=agent_description,
        tags=[tag.strip() for tag in agent_name.lower().split()],
        examples=[f'Interact with {agent_name}'],
    )

    return AgentCard(
        name=agent_name,
        description=agent_description,
        url=base_url,
        version='1.0.0',
        defaultInputModes=supported_content_types,
        defaultOutputModes=supported_content_types,
        capabilities=capabilities,
        skills=[skill],
    )


remote_agent = agent_engines.get(AGENT_ID)
adk_agent = AdkAgent(remote_agent=remote_agent)

agent_card = create_agent_card(
    base_url=os.environ['SERVICE_URL'],
    supported_content_types=AdkAgent.SUPPORTED_CONTENT_TYPES,
    agent_name=remote_agent.name,
    agent_description=remote_agent.name,
)
    
request_handler = DefaultRequestHandler(
    agent_executor=AdkAgentExecutor(adk_agent),
    task_store=InMemoryTaskStore(),
)

server_app_builder = A2AStarletteApplication(
    agent_card=agent_card, http_handler=request_handler
)

app = server_app_builder.build()
EOF


cat EOF > a2a_server/requirements.txt
uvicorn
starlette
google-adk==1.2.1
google-cloud-aiplatform==1.96.0
a2a-sdk==0.2.5
EOF


cat EOF > a2a_server/Dockerfile
FROM python:3.11-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
EOF

SERVICE_NAME = 'tobari-chan-a2a-server'

!cd a2a_server && gcloud builds submit --tag {REPO}/{SERVICE_NAME}
!gcloud run deploy {SERVICE_NAME} \
   --image {REPO}/{SERVICE_NAME} \
   --platform managed \
   --region {LOCATION} \
   --update-env-vars "SERVICE_URL=https://{SERVICE_NAME}-{PROJECT_NUMBER}.{LOCATION}.run.app" \
   --no-allow-unauthenticated 2>&1 | cat

Creating temporary archive of 4 file(s) totalling 7.8 KiB before compression.
...(中略)...
Service URL: https://tobari-chan-a2a-server-879055303739.us-central1.run.app

async def tobari_chan_tool_a2a(query: str) -> str:
    """
    Get an answer to a question regarding 夜の帳 from とばりちゃん

    Args:
        query: question
       
    Returns:
        str: An answer from とばりちゃん
    """
    import json
    import httpx
    from uuid import uuid4
    from a2a.client import A2AClient
    from a2a.types import MessageSendParams, SendStreamingMessageRequest
    import google.auth.transport.requests
    import google.oauth2.id_token 

    A2A_SERVER_URL = '[SERVICE_URL]' 

    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, A2A_SERVER_URL)
    headers = {
        'Authorization': f'Bearer {id_token}',
        'Content-Type': 'application/json'
    }
    payload = {
        'message': {
            'role': 'user',
            'parts': [{'kind': 'text', 'text': query}],
            'messageId': uuid4().hex,
        },
    }
    request = SendStreamingMessageRequest(params=MessageSendParams(**payload))

    async with httpx.AsyncClient(headers=headers, timeout=100) as httpx_client:
        client = await A2AClient.get_client_from_agent_card_url(
            httpx_client, A2A_SERVER_URL
        )
        stream_response = client.send_message_streaming(request)
        result = []
        async for chunk in stream_response:
            chunk = json.loads(chunk.root.model_dump_json(exclude_none=True))
            if 'DEBUG' in globals() and DEBUG:
                pprint.pp(chunk)
                print('====')
            if chunk['result']['kind'] == 'artifact-update':
                text_messages = []
                for part in chunk['result']['artifact']['parts']:
                    if 'text' in part:
                        text_messages.append(part['text'])
                result.append('\n'.join(text_messages))
    
    return '\n'.join(result)

print(await tobari_chan_tool_a2a('コーヒー以外のおすすめはあるの？'))

とばりちゃんが答えるよ！

はい、ございます！夜の帳では、軽食とデザートもご用意しております。
軽食では、ふわふわの「厚切りトーストのたまごサンド」や、喫茶店の定番「昔ながらのナポリタン」が人気です。
また、デザートもございますので、ぜひお試しくださいね。