[ネタ] Agent Engine で genkit を動かす
こんにちは サントリーこと大橋です。
先日GCPUG Tokyo Next 2025 Extended があって、AI Agentの話をしました。
その中で、Agent Engineでgenkitを動かせる！(やらないほうがいい)と言ったのですが、
やり方を細かく説明しなかったので、今回はAgent Engineで genkit を動かしてみたいと思います。
やらないほうがいいです。
事前知識
Agent Engineは2025/05/02現在、pythonでしか動かすことはできません。
ただ環境としてはCloud Run上で動いており、ある程度自由にpythonパッケージを入れることができるので、pypi経由で入れられるものは大抵導入できます。
つまり、pypi経由で、Node.jsが入れられれば、python経由で、Node.jsを実行し、genkitも動かせるはずです。それではやってみましょう。
genkit周りの作成
genkit環境準備
まずgenkitのコードを書くためのディレクトリを用意して、Node.jsの開発環境を用意します。
今回はgenkitをpythonからCLI経由で呼ぶことを想定するため、Node.jsのcli環境を作成します。
そのため、Node.js CLIを作るためのツール oclif を利用したいと思います。
npx -y oclif generate genkit-sample
cd genkit-sample
次にgenkitと、vertex-ai経由でgeminiを呼ぶために@genkit-ai/vertexaiをインストールします。
npm i genkit @genkit-ai/vertexai
コード作成
ではコードを書いていきます。今回は最低限のコードで済ませたいと思います。
まずgenkitのセットアップ用コードを作成します。
import {gemini25FlashPreview0417, vertexAI} from '@genkit-ai/vertexai';
import {genkit} from "genkit";
import {logger} from 'genkit/logging';
export const ai = genkit({
model: gemini25FlashPreview0417,
plugins: [vertexAI({
location: 'us-central1'
})]
});
logger.setLogLevel('info');
次にコマンド用コードを作成します。
genkitのインスタンスを呼ぶのみの最小限コードです。
import {Args, Command} from '@oclif/core';
import {ai} from '../ai/ai-instance';
export default class Message extends Command {
static override args = {
message: Args.string({description: 'message to send llm', required: true}),
}
static override description = 'Send the message to llm';
static override examples = [
'<%= config.bin %> <%= command.id %>',
]
static override flags = {
}
public async run(): Promise<void> {
const {args} = await this.parse(Message);
const {text} = await ai.generate(args.message);
console.log(text);
}
}
ビルド
Agent Engineへデプロイした後利用する為にビルドしておきます。
npm run build
python周りのコード
環境準備
次にpythonの準備します。
とりあえずローカル環境で言えば、python経由で、Agent Engineがデプロイ & 実行できればいいので google-cloud-aiplatform[agent-engines] だけインストールします。
uv add "google-cloud-aiplatform[agent-engines]"
Agent Engine用のカスタムAgentClass
Agent EngineにAgentをデプロイするためには、Agent Classを作成する必要があります。
Agent Classと便宜上言っていますが、特になにかのクラスを継承する必要があるわけではなく、以下のドキュメントに作成方法は書いています。
基本的には、setupとquery または stream_query メソッドがあり、pickel化で記載すれば、どのようなクラスでも大丈夫です。
setupはエージェントの初期化処理を書きます。 例えばSession保存用のDBへの接続処理などを行います。
queryは完全なレスポンスを1つの結果返すためのメソッドです。
stream_queryはyieldなどを利用してGeneratorを返却することで、streaming形式で返却できるようになります。
なお、register_operations メソッドを用意し、以下のように呼び出し可能なメソッドを返却することで、Agent Engineで呼び出し可能なAPIを増やすことができます。
def register_operations(self) -> Dict[str, List[str]]:
"""Registers the operations of the ADK application."""
return {
"": [
"get_session",
"list_sessions",
"create_session",
"delete_session",
],
"stream": ["stream_query", "streaming_agent_run_with_events"],
}
※ちなみに↑のはADKのAgent Engine用のAppのregister_operationsです。
今回は超雑に、ユーザーからのメッセージをそのままexecで実行するストロングスタイルで行きましょう。
セキュリティ上だめすぎるので、皆さんはデプロイしないでくださいね
import vertexai
from vertexai import agent_engines
PROJECT_ID = "project_id"
LOCATION = "location"
STAGING_BUCKET = "gs://bucket_name"
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
class TestApp:
def set_up(self):
import vertexai
vertexai.init(project=PROJECT_ID, location=LOCATION)
def query(self, message=None, **kwargs):
return exec(message)
デプロイ & 実行
次にこれらをデプロイするコードを書きます。
またデプロイと一緒に実行までやったほうが楽なので、実行するコードまで書いてしまいましょう。
# さっきのコードの下に書く
def main():
agent_engine = vertexai.agent_engines.create(
agent_engine=TestApp(),
display_name="Test App",
description="Test App",
requirements=[
"google-cloud-aiplatform[agent-engines]>=1.88.0",
"nodejs-wheel"
],
extra_packages=['genkit-sample']
)
print(agent_engine.query(message="""
import subprocess
import glob
from nodejs_wheel import (
node,
npm,
npx,
)
cp = node(["genkit-sample/bin/run.js", "message", "hello"], return_completed_process=True, capture_output=True, text=True)
print("------- node genkit --------")
print("returncode", cp.returncode)
print("stdout", cp.stdout)
print("stderr", cp.stderr)
"""
)
)
if __name__ == "__main__":
main()
上記のコードの肝は、
-
requirementsで Node.jsをインストールするためのnodejs-wheelパッケージをインストールしている -
genkit-sampleのコードをデプロイするためにextra_packagesで対象ディレクトリを設定している -
agent_engine.query(message=部分でデプロイしたagent_engineを実行しています。この部分で文字列で、作成したgenkitのCLIを呼び出しています。
実行
では実行してみましょう
uv run main.py
コードだけ見ると、コンソールのログにメッセージが表示されそうですが、表示されません。
ログはCloud Loggingに出力されます。見てみましょう。
無事実行できたようです！
まとめ
ネタですが、Agent Engineでgenkitを実行しました。
大切なポイントとしては、Agent Engineは決して、決まり切ったAI Frameworkを実行する環境ではなく、Agent Classでラップさえすればどのようなものでもデプロイが可能になる点です。
genkitを無理やり動かしましたが、それはやらないほうが良さそうですけどね
Discussion