开发者如何在 Poe 上构建和托管 API 服务机器人以部署自定义模型

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> 标头来验证此密钥。

设置开发环境

官方的 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）上的容器化托管

对于运行专有推理集群或利用托管容器平台的团队：

1. 容器化应用：创建一个运行 uvicorn app:app --host 0.0.0.0 --port 8080 --workers 4 的轻量级 Dockerfile。
2. 反向代理和 SSL：在支持长寿命 HTTP 连接和流式分块响应的入口控制器（例如 AWS ALB、Cloudflare 或 Nginx）后面部署。确保代理超时（例如 proxy_read_timeout）配置为至少 300 秒，以防止中途断开连接。
3. 环境机密：通过容器环境变量安全地注入 POE_ACCESS_KEY 和推理凭据。

在 Poe 上注册和配置机器人

一旦您的服务器部署好并通过公共 HTTPS 访问后，即可在 Poe 平台上完成注册：

1. 导航到 poe.com/create_bot?server=1。
2. 机器人句柄 (Bot Handle)：选择一个唯一的字母数字标识符（例如 @MyCustomModel）。这构成机器人的公共链接。
3. 机器人类型 (Bot Type)：确保选择“服务器机器人 (Server bot)”。
4. 服务器 URL (Server URL)：输入完整的公共端点（例如 https://your-domain.com/ 或您的 Modal 部署 URL）。
5. 访问密钥 (Access Key)：粘贴配置在服务器 fp.make_app() 调用中的秘密 token。Poe 将通过每个测试请求来验证此 token。
6. 个人资料详情 (Profile Details)：设置对用户可见的显示名称、描述、头像和系统指令。
7. 可见性 (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）中。