Poe Server Bots 允許開發者使用基於 Server-Sent Events (SSE) 的標準 HTTP 協議,將外部計算基礎設施、自訂機器學習模型和專有邏輯直接連接到 Quora 的 Poe 平台。與依賴 Poe 託管商業模型及自訂系統提示的標準提示機器人不同,伺服器機器人使工程團隊能夠完全控制後端推理引擎、私有權重、檢索增強生成 (RAG) 管道和外部 API 整合。透過在輕量級的 fastapi_poe Python 庫中封裝自訂模型並將服務部署到可公開存取的端點,開發者可以立即將專有 AI 工具分發給 Poe 跨網路、iOS、Android 和桌面應用的全球用戶群。
架構與 Poe Server Bot 協議
Poe 與您的自訂後端之間的通訊遵循透過 HTTPS 進行的非同步主從式架構。當用戶在 Poe 上向您的伺服器機器人發送訊息時,Poe 基礎設施會將訊息上下文作為 HTTP POST 請求轉發到您指定的伺服器端點。您的伺服器處理傳入的有效負載,並使用 Server-Sent Events (SSE) 將生成的響應串流傳輸回 Poe。這種架構確保了低延遲,並為終端用戶提供即時的 token 流。
協議規範標準化了以下幾種事件和請求類型:
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() 中的秘密 token |
設置開發環境
官方的 fastapi_poe 庫提供了實現兼容伺服器機器人的最快途徑。它將 HTTP 解析、事件序列化和身份驗證抽象為一個可擴展的非同步介面。
首先安裝 fastapi-poe 以及像 Uvicorn 這樣的 ASGI 伺服器:
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 訊息橋接到推理引擎,並將生成的 token 作為 PartialResponse 對象串流輸出。
與 vLLM 或 Hugging Face 推理後端整合
當在專用 GPU 硬體上運行開放權重模型(例如 Llama 3、Mistral 或 Qwen)時,透過像 vLLM 或 Hugging Face Text Generation Inference (TGI) 這樣的優化引擎為它們提供服務,可以提供高吞吐量和連續批次處理。
以下示例演示了一個將請求轉發到兼容 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)上的容器化託管
對於運行專有推理集群或利用託管容器平台的團隊:
- 容器化應用:創建一個運行
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。
- 機器人句柄 (Bot Handle):選擇一個唯一的字母數字識別碼(例如
@MyCustomModel)。這構成機器人的公共連結。 - 機器人類型 (Bot Type):確保選擇「伺服器機器人 (Server bot)」。
- 伺服器 URL (Server URL):輸入完整的公共端點(例如
https://your-domain.com/或您的 Modal 部署 URL)。 - 訪問金鑰 (Access Key):貼上配置在伺服器
fp.make_app()調用中的秘密 token。Poe 將透過每個測試請求來驗證此 token。 - 個人資料詳情 (Profile Details):設置對用戶可見的顯示名稱、描述、頭像和系統指令。
- 可見性 (Visibility):選擇機器人應保持私有(僅您或指定的團隊成員可存取)還是公開(在 Poe 探索目錄中可搜尋)。
點擊 Create Bot(創建機器人)。Poe 立即透過向您的端點發出 SettingsRequest 來驗證連接。如果伺服器返回有效響應,則機器人將被激活。
貨幣化和計算點數經濟學
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"
)
生產環境最佳實踐
部署可靠的伺服器機器人需要解決網路波動、延遲預算和操作遙測問題:
- 實施即時串流傳輸:當第一個 token 快速出現時,用戶會感覺到低延遲。避免在產生輸出前將整個模型生成內容緩存在記憶體中。在從您的推理後端到達第一個塊後立即開始發出 token。
- 優雅的錯誤恢復:將模型調用封裝在結構化異常處理程序中。如果上游 GPU 工作節點失敗,請返回具有清晰解釋的
fp.ErrorResponse並設置allow_retry=True,以便用戶可以重新觸發查詢而無需重新加載應用。 - 建議跟進:透過在響應結束時附加
fp.MetaResponse(suggested_replies=["Explain further", "Provide code sample"])的建議回覆來提升參與度。 - 監控遙測:在您的
PoeBot類上實現on_feedback方法,以將正面(點讚)和負面(點踩)評分直接記錄到您的可觀測性平台(例如 Datadog、Prometheus 或 Grafana)中。