Poe Server Bots を使用すると、開発者は、Server-Sent Events (SSE) に基づく標準化された HTTP プロトコルを使用して、外部のコンピューティングインフラストラクチャ、カスタム機械学習モデル、および独自のロジックを直接 Quora の Poe プラットフォームに接続できます。カスタムシステムプロンプトを使用して Poe がホストする商用モデルに依存する標準のプロンプトボットとは異なり、サーバーボットは、バックエンドの推論エンジン、プライベートな重み、検索拡張生成 (RAG) パイプライン、および外部 API 統合に対する完全な制御をエンジニアリングチームに提供します。軽量な fastapi_poe Python ライブラリにカスタムモデルをラップし、公開アクセス可能なエンドポイントにサービスをデプロイすることで、開発者は、ウェブ、iOS、Android、デスクトップアプリを横断して Poe のグローバルユーザーベースに独自の AI ツールを即座に配信できます。
アーキテクチャと Poe Server Bot プロトコル
Poe とカスタムバックエンド間の通信は、HTTPS 経由の非同期クライアントサーバーモデルに従います。ユーザーが Poe 上のサーバーボットにメッセージを送信すると、Poe インフラストラクチャはメッセージコンテキストを HTTP POST リクエストとして指定されたサーバーエンドポイントに転送します。サーバーは受信したペイロードを処理し、Server-Sent Events (SSE) を使用して生成された応答を Poe にストリーミングして送り返します。このアーキテクチャにより、低遅延が確保され、エンドユーザーへのリアルタイムのトークンストリーミングが提供されます。
プロトコル仕様では、いくつかのイベントとリクエストタイプが標準化されています。
QueryRequest: スレッド内の過去のメッセージ (query)、一意の会話識別子 (conversation_id)、ユーザー識別子 (user_id)、メッセージのメタデータ、およびクライアント機能を含む、完全な会話コンテキストが含まれます。PartialResponse: サーバーによって発行されるアトミックなストリーミングチャンク。各部分応答は、クライアントに増分テキスト、データペイロード、または状態の更新を提供します。SettingsRequest/SettingsResponse: サーバーが Poe との間で、紹介の挨拶、ファイル添付の許可、おすすめの返信、ボット間のクエリの依存関係の宣言などのボットレベルの構成をネゴシエートできるようにします。ReportFeedbackRequestおよびReportErrorRequest: ユーザーがメッセージを評価したとき、またはクライアント側で送信エラーが発生したときに、Poe からサーバーへのテレメトリおよびフィードバックコールバックを提供します。
リクエストを認証し、バックエンドを不正なトラフィックから保護するために、Poe は Access Key を発行します。サーバーは、着信リクエストごとに送信される Authorization: Bearer <access_key> ヘッダーに対してこのキーを検証します。
| Component | Role in Server Bot Workflow | Key Configuration / Method |
|---|---|---|
fastapi_poe | FastAPI と Starlette の上に Poe プロトコルを実装する Python SDK | fp.make_app(), fp.run() |
PoeBot | ボットの動作とイベントハンドラーを定義するベースクラス | get_response(), get_settings() |
| Poe Gateway | ユーザー入力をリレーし、エンドポイントの応答性を検証する | https://poe.com/create_bot?server=1 |
| Inference Server | モデルの重み(vLLM、TGI、Ollama、カスタム PyTorch など)をホストする | 非同期ストリーミングジェネレーター |
| Access Key | Poe からの受信 HTTP POST 呼び出しを認証する | make_app() に渡されるシークレットトークン |
開発環境のセットアップ
公式の fastapi_poe ライブラリは、準拠したサーバーボットを実装するための最短ルートを提供します。HTTP パース、イベントのシリアライズ、認証を拡張可能な非同期インターフェースに抽象化します。
まず、Uvicorn などの ASGI サーバーとともに fastapi-poe をインストールします。
pip install fastapi-poe uvicorn
最小限のサーバーボットは PoeBot をサブクラス化し、get_response ジェネレーターを実装します。
import fastapi_poe as fp
from typing import AsyncIterable
class EchoBot(fp.PoeBot):
async def get_response(
self, request: fp.QueryRequest
) -> AsyncIterable[fp.PartialResponse]:
last_message = request.query[-1].content
yield fp.PartialResponse(text=f"Received: {last_message}")
async def get_settings(
self, setting: fp.SettingsRequest
) -> fp.SettingsResponse:
return fp.SettingsResponse(
introduction_message="Welcome! This bot is running on a custom backend.",
allow_attachments=False
)
bot = EchoBot()
app = fp.make_app(bot, access_key="YOUR_SECRET_ACCESS_KEY")
if __name__ == "__main__":
fp.run(bot, access_key="YOUR_SECRET_ACCESS_KEY", port=8080)
ローカルでのプロトタイピング中は、ngrok や Cloudflare Tunnels などのトンネリングツールを使用して、ローカルポート(例:8080)をパブリックインターネットに公開できます。
ngrok http 8080
これにより、一時的なパブリック HTTPS URL(例:https://abcdef123.ngrok-free.app)が生成され、すぐにテストするために Poe のボット登録インターフェースに入力できます。
カスタムモデルと推論パイプラインのデプロイ
サーバーボットをカスタムモデルに接続するには、受信した QueryRequest メッセージを推論エンジンにブリッジし、生成されたトークンを PartialResponse オブジェクトとしてストリーミング出力する必要があります。
vLLM または Hugging Face 推論バックエンドとの統合
専用の GPU ハードウェア上でオープンウェイトモデル(Llama 3、Mistral、Qwen など)を実行する場合、vLLM や Hugging Face Text Generation Inference (TGI) などの最適化されたエンジンを通じてそれらを提供することで、高いスループットと継続的なバッチ処理(continuous batching)が実現します。
次の例は、OpenAI 互換の vLLM 推論サーバーにリクエストを転送する Poe サーバーボットを示しています。
import os
import fastapi_poe as fp
from openai import AsyncOpenAI
from typing import AsyncIterable
# Connect to local or remote vLLM endpoint
client = AsyncOpenAI(
base_url=os.environ.get("VLLM_BASE_URL", "http://localhost:8000/v1"),
api_key=os.environ.get("VLLM_API_KEY", "EMPTY"),
)
class CustomModelBot(fp.PoeBot):
async def get_response(
self, request: fp.QueryRequest
) -> AsyncIterable[fp.PartialResponse]:
# Convert Poe conversation history to chat messages
messages = []
for message in request.query:
role = "user" if message.role == "user" else "assistant"
messages.append({"role": role, "content": message.content})
try:
stream = await client.chat.completions.create(
model="meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
stream=True,
temperature=0.7,
max_tokens=1024,
)
async for chunk in stream:
token = chunk.choices[0].delta.content
if token:
yield fp.PartialResponse(text=token)
except Exception as e:
yield fp.ErrorResponse(
text=f"An error occurred during inference: {str(e)}",
allow_retry=True
)
async def get_settings(
self, setting: fp.SettingsRequest
) -> fp.SettingsResponse:
return fp.SettingsResponse(
introduction_message="Powered by a custom open-source model running on dedicated infrastructure.",
enable_image_comprehension=False,
allow_attachments=False
)
bot = CustomModelBot()
app = fp.make_app(bot, access_key=os.environ.get("POE_ACCESS_KEY"))
マルチターンコンテキストと添付ファイルの処理
Poe は、request.query 配列で完全なダイアログ履歴を時系列順に送信します。カスタムモデルのコンテキストウィンドウが限られている場合、サーバーボットのロジックはトークン化の前に以前のターンを削減または要約する必要があります。
ボットが get_settings で allow_attachments=True を有効にすると、ファイルアップロード(画像、PDF、テキストドキュメント)が request.query[-1].attachments に表示されます。各添付ファイルオブジェクトにはダウンロード URL と MIME タイプが含まれているため、バックエンドはアセットをダウンロードしたり、OCR を実行したり、ビジョンエンコーディングを実行したり、応答を生成する前にドキュメントをベクターストアに埋め込んだりできます。
サーバーボットのホスティングオプション
Poe では、SSE 応答をストリーミングできる中断のない HTTPS エンドポイントが必要なため、適切なホスティングプラットフォームの選択は、モデルのサイズ、計算要件、予算によって異なります。
Modal によるサーバーレス GPU デプロイ
Modal は Poe サーバーボットのファーストクラスのサポートを提供し、開発者が GPU インフラストラクチャをコードとして定義し、実際の実行時間に対してのみ支払うことができるようにします。Modal はコンテナインスタンスをゼロから自動的にスケーリングし、HTTPS URL を提供します。
本番環境に対応した Modal デプロイスクリプト:
import os
import modal
import fastapi_poe as fp
app = modal.App("poe-custom-model-bot")
# Define container environment with dependencies
image = (
modal.Image.debian_slim()
.pip_install("fastapi-poe", "openai")
)
@app.function(
image=image,
secrets=[modal.Secret.from_name("poe-custom-model-secrets")],
keep_warm=1, # Keeps one instance ready to prevent cold starts
)
@modal.asgi_app()
def fastapi_app():
from bot_implementation import CustomModelBot
bot = CustomModelBot()
return fp.make_app(bot, access_key=os.environ["POE_ACCESS_KEY"])
modal deploy app.py を使用してこのサービスをデプロイすると、次のような永続的な URL が出力されます。
https://<organization>--poe-custom-model-bot-fastapi-app.modal.run
クラウドプロバイダー(AWS、GCP、Render、Fly.io)でのコンテナ化されたホスティング
独自の推論クラスターを実行している、またはマネージドコンテナプラットフォームを利用しているチームの場合:
- アプリケーションの Docker 化:
uvicorn app:app --host 0.0.0.0 --port 8080 --workers 4を実行する軽量の Dockerfile を作成します。 - リバースプロキシと SSL: 長寿命の HTTP 接続とチャンク化された応答のストリーミングをサポートするイングレスコントローラー(AWS ALB、Cloudflare、Nginx など)の背後にデプロイします。ストリームの途中で切断されないように、プロキシのタイムアウト(例:
proxy_read_timeout)が少なくとも 300 秒に構成されていることを確認してください。 - 環境シークレット: コンテナ環境変数を通じて、
POE_ACCESS_KEYと推論資格情報を安全に挿入します。
Poe へのボットの登録と構成
サーバーがデプロイされ、パブリック HTTPS 経由でアクセスできるようになったら、Poe プラットフォームでの登録を完了します。
- poe.com/create_bot?server=1 に移動します。
- ボットハンドル: 一意の英数字の識別子(例:
@MyCustomModel)を選択します。これがボットの公開リンクになります。 - ボットタイプ: 「Server bot」が選択されていることを確認します。
- サーバー URL: 完全なパブリックエンドポイント(例:
https://your-domain.com/または Modal のデプロイ URL)を入力します。 - アクセスキー (Access Key): サーバーの
fp.make_app()呼び出しで構成されたシークレットトークンを貼り付けます。Poe は、各テストリクエストでこのトークンを検証します。 - プロファイルの詳細: ユーザーに表示される表示名、説明、アバター、およびシステム指示を設定します。
- 公開設定: ボットを非公開(自分または指定されたチームメンバーのみがアクセス可能)のままにするか、公開(Poe Explore ディレクトリで検索可能)にするかを選択します。
Create Bot をクリックします。Poe は、エンドポイントに SettingsRequest を発行することで接続性を直ちに検証します。サーバーが有効な応答を返した場合、ボットが有効になります。
収益化とコンピュートポイント(Compute Point)の経済性
Poe はサーバーボットにネイティブのクリエイター収益化機能を提供しており、ユーザーがツールを操作するにつれて開発者が収益を得ることができます。
メッセージあたりの価格設定とボット収益化 API
クリエイターは、ボット設定でメッセージあたりの固定コンピュートポイントコストを設定するか、ボット収益化 API を利用して使用量に基づく動的な課金を行うことができます。
- サブスクリプション収益の分配: ボットと操作した後に新しいユーザーが Poe にサブスクライブすると、アフィリエイトの支払いを受け取ります。
- コンピュートポイントの消費: 既存の Poe サブスクライバーまたは無料ユーザーがボットにコンピュートポイントを費やすと、消費されたポイントに比例した収益を獲得します。
- 変動価格設定: モデルが予測不可能な計算コストを伴うタスク(例:高解像度の画像レンダリングや長いコンテキストのドキュメント分析)を実行する場合、
authorize_costおよびcapture_costエンドポイントを使用して変動価格設定を構成できます。コストのかかる操作が実行される前に、明確なrate_cardまたはcost_labelがユーザーに提示されます。
# Setting dynamic rate cards in SettingsResponse
async def get_settings(self, setting: fp.SettingsRequest) -> fp.SettingsResponse:
return fp.SettingsResponse(
introduction_message="Professional analysis bot with dynamic compute points.",
cost_label="50-200 pts",
rate_card="* Base Query: 50 points\n* Extended Analysis: 150-200 points"
)
本番環境のベストプラクティス
信頼性の高いサーバーボットをデプロイするには、ネットワークの変動、遅延バジェット、および操作テレメトリに対処する必要があります。
- 即時ストリーミングの実装: ユーザーは、最初のトークンがすぐに表示されると低遅延であると認識します。モデルの生成全体を出力する前にメモリにバッファリングすることは避けてください。推論バックエンドから最初のチャンクが到着したらすぐにトークンの発行を開始します。
- グレースフルなエラーリカバリ: モデルの呼び出しを構造化された例外ハンドラーでラップします。上流の GPU ワーカーが失敗した場合は、明確な説明とともに
fp.ErrorResponseを返し、allow_retry=Trueを設定して、ユーザーがアプリをリロードせずにクエリを再トリガーできるようにします。 - おすすめのフォローアップ: 応答の最後に
fp.MetaResponse(suggested_replies=["Explain further", "Provide code sample"])を介しておすすめの返信を追加することで、エンゲージメントを強化します。 - テレメトリの監視:
PoeBotクラスにon_feedbackメソッドを実装して、肯定的な評価(いいね)と否定的な評価(よくないね)をオブザーバビリティプラットフォーム(Datadog、Prometheus、Grafana など)に直接記録します。