1. 项目概述与背景

如果你是一个经常和各类AI模型打交道的开发者或爱好者，那么对Poe.com这个平台一定不陌生。它就像一个AI模型的“聚合器”，把ChatGPT、Claude、Gemini、Code Llama等主流模型都整合到了一个统一的聊天界面里，用户无需在多个平台间切换，就能方便地使用。然而，对于开发者来说，通过网页界面交互虽然直观，但在自动化、集成和批量处理方面就显得力不从心了。这正是 snowby666/poe-api-wrapper 这个项目诞生的初衷。

简单来说， poe-api-wrapper 是一个用Python编写的、非官方的Poe.com API封装库。它的核心目标，是让你能够通过代码，以编程的方式调用Poe平台上的所有AI模型，就像调用OpenAI官方API一样方便。想象一下，你可以写一个脚本，让Claude-3.5-Sonnet分析一份PDF报告，然后让GPT-4o根据分析结果生成一份摘要，最后再用Stable Diffusion画一张配图——整个过程完全自动化，无需你手动复制粘贴。这个库就是实现这类场景的“桥梁”。

我最初接触这个项目，是因为需要将AI对话能力集成到一个内部的数据处理流水线中。官方的解决方案要么费用高昂，要么模型选择有限。Poe平台提供了丰富的免费和付费模型，但缺乏官方的API支持。 poe-api-wrapper 通过逆向工程Poe的网页接口，巧妙地解决了这个问题。它模拟了浏览器与Poe服务器之间的通信，将复杂的HTTP请求、Cookie管理、消息轮询等细节封装成简洁的Python方法，让开发者可以专注于业务逻辑。

注意：根据项目README的声明，该项目已不再积极维护，官方推荐使用Poe推出的官方API。这意味着，依赖此库进行生产级开发存在一定风险，例如接口变更导致服务中断。但对于学习、原型验证、个人自动化工具等场景，它仍然是一个极具价值的工具，能帮助你快速理解如何与这类非标准API交互。

2. 核心功能与设计思路解析

这个库的设计非常“Pythonic”，它充分考虑了不同使用场景下的开发者体验。其核心设计思路可以概括为： 通过模拟用户行为，提供一套同步与异步并存的、高层次的API抽象 。下面我们来拆解它的几个关键设计决策。

2.1 双模式客户端：同步与异步

现代Python应用开发，尤其是在涉及网络I/O的场景下，异步编程（asyncio）对于提升并发性能和资源利用率至关重要。 poe-api-wrapper 从一开始就提供了 PoeApi （同步）和 AsyncPoeApi （异步）两个客户端类。

为什么这样设计？

• 同步客户端 ( PoeApi ) ：适用于简单的脚本、快速测试、或者对并发要求不高的场景。它的代码逻辑是线性的，更易于理解和调试。例如，你写一个每天定时运行一次，向Claude询问今日新闻摘要的脚本，用同步客户端就足够了。
• 异步客户端 ( AsyncPoeApi ) ：适用于需要同时处理多个对话、构建实时聊天应用、或者作为Web服务后端的情况。异步模式可以让你在等待一个AI模型生成回复时，去处理另一个用户的请求或执行其他计算任务，极大地提高了程序的吞吐量。

在实际选择时，如果你的应用是Web框架（如FastAPI、Django Channels）、GUI应用（如Tkinter、PyQt的事件循环）或者需要高并发的爬虫/自动化工具，那么异步客户端是更优的选择。库的示例代码也展示了如何用 async for 来流式接收AI的回复，这是一种非常高效且用户体验好的方式。

2.2 认证机制：Cookie驱动的会话管理

与大多数提供API Key的官方服务不同，Poe本身没有开放官方的开发者API。因此，这个库的认证机制是基于Web用户会话的，具体来说，是依赖于浏览器Cookie。

核心认证令牌：

1. p-b 和 p-lat Cookie (必需) ：这是Poe用于标识用户身份和会话有效性的核心凭证。获取方式就是登录Poe官网后，从浏览器的开发者工具中复制。这本质上是在模拟一个已登录的浏览器会话。
2. formkey (可选) ：这是一个用于防止CSRF（跨站请求伪造）的令牌。库本身会尝试自动获取，但在某些情况下（如遇到验证挑战）可能需要手动提供以增强稳定性。

这种设计的利弊分析：

• 优点 ：实现相对简单，直接复用了现有的Web认证流程，绕开了官方API限制。
• 缺点与风险 ：
  • 账号安全 ：将 p-b 和 p-lat 这样的会话Cookie用于自动化脚本，存在一定的安全风险。如果脚本或存储凭证的环境泄露，相当于你的Poe账号完全暴露。
  • 稳定性 ：Poe可以随时更改其前端认证逻辑或Cookie机制，导致库失效。这也是项目不再维护的主要原因之一。
  • 速率限制 ：基于用户会话的请求，会受到Poe对单个用户账号的速率限制，这可能比官方API的限制更为严格和不透明。

实操心得：强烈建议为自动化脚本创建一个专用的Poe账号，而不是使用你的主账号。并且，妥善保管Cookie信息，不要将其硬编码在代码中提交到Git等版本控制系统。应该使用环境变量或配置文件来管理，并在 .gitignore 中排除这些敏感文件。

2.3 功能模块化设计

库的功能组织得非常清晰，主要围绕以下几个核心模块展开，这反映了作者对Poe平台功能的深度理解：

1. 消息自动化 ：这是最核心的功能。包括创建新会话、发送消息、流式接收回复、重试上一条消息、附加文件、获取建议回复、停止生成等。它覆盖了一个完整对话生命周期的所有操作。
2. 会话管理 ：获取聊天记录、删除会话、清除上下文（Chat Break）。这让你能管理历史对话，对于构建有记忆的聊天机器人或清理测试数据非常有用。
3. 机器人（Bot）管理 ：获取机器人信息、创建/编辑/删除自定义机器人。Poe允许用户基于现有模型创建具有特定指令（系统提示词）的自定义机器人，这个库让你能用代码来批量管理这些机器人。
4. 知识库定制 ：为自定义机器人上传和编辑知识库文件。这是实现“私有数据AI问答”的关键，你可以让机器人基于你提供的文档（如公司手册、产品说明书）来回答问题。
5. 探索与发现 ：浏览平台上的官方和第三方机器人、分类。可以用于发现新模型或有趣的机器人应用。
6. 群聊功能 (Beta) ：创建和管理多个机器人参与的群聊。这开启了更复杂的交互可能性，例如让多个专家模型对同一个问题进行辩论或协作。

这种模块化设计使得代码结构清晰，开发者可以根据需要只关注某一部分功能，学习成本较低。

3. 环境准备与核心配置详解

在开始写代码之前，我们需要完成一些准备工作。这个过程虽然有些繁琐，但却是保证库能正常工作的基础。

3.1 安装与依赖管理

库可以通过pip直接安装，非常方便。但这里有一些细节需要注意。

# 基础安装，适用于大多数场景
pip install -U poe-api-wrapper

# 如果你需要自动代理功能（用于应对可能的IP限制），且Python版本在3.9以上，可以安装带proxy支持的版本
pip install -U 'poe-api-wrapper[proxy]'

# 如果你计划使用其内置的OpenAI API兼容服务器功能，则需要安装llm扩展
pip install -U 'poe-api-wrapper[llm]'

依赖选择解析：

• 基础安装 ：包含了与Poe通信的所有核心逻辑。对于只需要直接调用 PoeApi 或 AsyncPoeApi 进行对话的场景，这就足够了。
• [proxy] 扩展 ：这个选项安装了一些额外的包（如 aiohttp-socks ），用于支持SOCKS5等代理协议。如果你的网络环境需要透过代理访问Poe，或者你打算运行大量请求希望轮换IP以避免被封禁，那么这个扩展是必要的。否则，基础安装会使用标准的 requests 和 aiohttp 库。
• [llm] 扩展 ：这个扩展是为了运行 PoeServer ，即那个能将Poe接口模拟成OpenAI API格式的本地服务器。它通常依赖 fastapi , uvicorn , openai 等库。只有当你希望让你那些原本使用OpenAI API的代码（例如基于 langchain 或 llama-index 的应用）无缝切换到Poe模型时，才需要安装它。

3.2 获取认证令牌（Cookie）

这是最关键的一步。我们需要从浏览器中提取 p-b 和 p-lat 这两个Cookie的值。

详细操作步骤：

1. 登录Poe ：用浏览器打开 https://poe.com 并完成登录。
2. 打开开发者工具 ：
   • Chrome/Edge ：按 F12 或右键页面选择“检查”，然后切换到 Application (应用程序) 标签页。在左侧找到 Storage -> Cookies -> https://poe.com 。
   • Firefox ：按 F12 ，切换到 Storage (存储) 标签页，然后展开 Cookies -> https://poe.com 。
   • Safari ：需先在“偏好设置-高级”中开启“在菜单栏中显示开发菜单”，然后按 Option+Command+C 打开开发者工具，切换到 Storage 标签页。
3. 查找并复制Cookie ：在Cookie列表中，找到名为 p-b 和 p-lat 的条目，将其 Value (值) 复制下来。它们通常是一长串看似随机的字母数字组合。

获取可选的 formkey ： 库通常会自己获取 formkey ，但有时会失败。手动获取有两种方法：

• 网络请求法 ：在开发者工具的 Network (网络) 标签页，刷新页面，找到一个名为 gql_POST 的请求，查看其 Headers (请求头)，在 Request Headers 部分找到 Poe-Formkey 并复制其值。
• 控制台脚本法 ：在开发者工具的 Console (控制台) 标签页，输入 allow pasting 回车（如果浏览器有安全限制），然后粘贴并执行脚本 window.ereNdsRqhp2Rd3LEW() ，控制台会输出一个字符串，这就是 formkey 。

安全存储令牌： 绝对不要像下面这样把令牌硬编码在代码里：

# ❌ 错误示范：硬编码敏感信息
tokens = {
    'p-b': 'your_p-b_cookie_here',
    'p-lat': 'your_p-lat_cookie_here',
}

正确的做法是使用环境变量：

# ✅ 正确示范：使用环境变量
import os

tokens = {
    'p-b': os.environ.get('POE_P_B_COOKIE'),
    'p-lat': os.environ.get('POE_P_LAT_COOKIE'),
    # 可选
    'formkey': os.environ.get('POE_FORMKEY', ''), # 提供默认值
}

然后在运行脚本前，在终端设置环境变量（Linux/macOS）：

export POE_P_B_COOKIE="你的p-b值"
export POE_P_LAT_COOKIE="你的p-lat值"

或者在Windows PowerShell中：

$env:POE_P_B_COOKIE="你的p-b值"
$env:POE_P_LAT_COOKIE="你的p-lat值"

对于更复杂的配置，可以使用 .env 文件配合 python-dotenv 库来管理。

3.3 客户端初始化与代理配置

初始化客户端时，除了传入令牌，还有一些重要的配置项。

from poe_api_wrapper import PoeApi
import asyncio
from poe_api_wrapper import AsyncPoeApi

# 同步客户端基础初始化
tokens = {'p-b': '...', 'p-lat': '...'}
client = PoeApi(tokens=tokens)

# 同步客户端 - 启用自动代理（需要安装[proxy]扩展）
client_with_proxy = PoeApi(tokens=tokens, auto_proxy=True)

# 同步客户端 - 手动指定代理列表
proxy_list = [
    {"http": "http://user:pass@proxy1:port", "https": "http://user:pass@proxy1:port"},
    {"http": "socks5://user:pass@proxy2:port", "https": "socks5://user:pass@proxy2:port"},
]
client_manual_proxy = PoeApi(tokens=tokens, proxy=proxy_list)

# 异步客户端初始化
async def main():
    async_client = await AsyncPoeApi(tokens=tokens).create()
    # ... 使用 async_client
asyncio.run(main())

配置参数解析：

• auto_proxy=True ：这个选项会让库尝试从一些公共源获取免费的代理IP并使用。 慎用 ，因为公共代理往往速度慢、不稳定且不安全，可能泄露你的请求数据。仅建议在无法直接连接Poe且没有可靠代理时临时使用。
• proxy= ：手动传入代理列表。列表中的每个元素是一个字典，格式与 requests 或 aiohttp 库的代理参数兼容。库会在发送请求时随机或按顺序使用列表中的代理。这是处理IP限制或进行分布式请求的更可靠方式。
• 异步客户端的 .create() ：注意 AsyncPoeApi 是一个异步类，其初始化需要 await 。这是因为它内部可能有一些异步的准备工作（如建立连接池）。务必使用 await AsyncPoeApi(...).create() ，而不是直接 AsyncPoeApi(...) 。

4. 基础功能实战：从对话到管理

现在，让我们进入实战环节，看看如何用这个库完成最常见的任务。我会结合代码示例和实际经验，讲解其中的细节和坑点。

4.1 与AI机器人对话

发送消息并接收回复是最核心的功能。库支持流式和非流式两种方式。

from poe_api_wrapper import PoeApi

client = PoeApi(tokens=tokens)
bot = "a2"  # 这是Claude-instant的模型代号
message = "用简单的语言解释一下量子计算的基本概念。"

# 方式1：流式输出（推荐用于长回复，体验好）
print("AI正在回复（流式）: ")
for chunk in client.send_message(bot, message):
    # chunk是一个字典，包含多种信息，其中'response'是当前片段的文本
    print(chunk["response"], end='', flush=True)
print("\n---流式结束---")

# 方式2：非流式输出（获取完整回复）
full_response = ""
for chunk in client.send_message(bot, message):
    # 循环会持续到消息生成完毕，最后一个chunk包含完整文本
    pass
# 最后一个chunk的‘text’字段包含了完整的回复
full_text = chunk["text"]
print(f"完整回复:\n{full_text}")

# 从回复中提取有用的元信息
chat_code = chunk["chatCode"]  # 本次对话的唯一标识码，用于继续此会话
chat_id = chunk["chatId"]      # 本次对话的ID
message_price = chunk.get("msgPrice")  # 本次消息消耗的点数（如有）
print(f"聊天代码: {chat_code}, 聊天ID: {chat_id}")

关键参数与技巧：

• bot 参数 ：这里需要传入的是模型的“代号”（codename），而不是显示名称。例如， "a2" 对应Claude-instant， "chinchilla" 对应GPT-3.5-Turbo。项目README中提供了一个非常详细的表格，列出了所有官方机器人的代号。对于自定义机器人，直接使用你在Poe上创建时设定的显示名称即可。
• chunk 字典的内容 ：在流式输出中，每一次迭代得到的 chunk 都包含 "response" 字段，即最新生成的一小段文本。除此之外， chunk 还可能包含 "text" （当前累计的完整文本）、 "suggestedReplies" （建议回复）、 "citations" （引用来源）等字段，但它们通常在流式传输的最后一个 chunk 或非流式模式下才完整填充。
• flush=True ：在打印流式输出时设置 flush=True 可以确保字符立即显示，而不是缓存在缓冲区，从而获得更实时的效果。

4.2 管理聊天历史与上下文

自动化场景下，我们经常需要管理对话的历史记录和状态。

# 1. 获取聊天历史
# 获取所有机器人的最近50个对话线程（默认）
all_history = client.get_chat_history()
print(f"历史记录数据: {all_history['data']}")
print(f"用于分页的光标: {all_history.get('cursor')}")

# 获取特定机器人（如Claude-instant）的聊天历史
claude_history = client.get_chat_history(bot="a2")
for chat_thread in claude_history['data'].get('a2', []):
    print(f"标题: {chat_thread['title']}, 聊天ID: {chat_thread['chatId']}, 代码: {chat_thread['chatCode']}")

# 获取最近的10个对话（所有机器人）
recent_chats = client.get_chat_history(count=10)

# 2. 继续一个已有的对话
# 使用 chatCode 或 chatId 来指定对话线程
continue_message = "那么，量子纠缠在实际的量子计算机中是如何应用的呢？"
for chunk in client.send_message(bot="a2", 
                                 message=continue_message, 
                                 chatCode=chat_code):  # 或 chatId=chat_id
    print(chunk["response"], end='', flush=True)

# 3. 清除对话上下文（Chat Break）
# 这相当于在网页上点击“New Chat”，机器人会忘记之前的对话内容，但会话记录本身还在。
client.chat_break(bot="a2", chatCode=chat_code)
print("上下文已清除。接下来发送的消息将开启一个全新的对话。")

# 4. 删除聊天会话
# 删除单个会话
client.delete_chat(bot="a2", chatCode=chat_code)
# 删除多个会话（传入列表）
client.delete_chat(bot="a2", chatId=[chat_id_1, chat_id_2])
# 删除该机器人的所有会话（谨慎操作！）
# client.delete_chat(bot="a2", del_all=True)

分页获取历史记录： 当对话数量很多时， get_chat_history 支持分页。返回的字典中有一个 'cursor' 字段，可以用于获取下一页。

def get_all_chats(bot_name=None, max_pages=5):
    """获取所有聊天记录（分页示例）"""
    all_chats = []
    cursor = None
    page = 0
    
    while page < max_pages:
        if bot_name:
            history = client.get_chat_history(bot=bot_name, count=50, cursor=cursor)
        else:
            history = client.get_chat_history(count=50, cursor=cursor)
            
        if not history['data']:
            break
            
        all_chats.append(history['data'])
        cursor = history.get('cursor')
        
        if cursor is None:
            break
            
        page += 1
        # 建议在分页请求间添加短暂延迟，避免请求过快
        time.sleep(0.5)
        
    return all_chats

4.3 高级消息功能

除了发送纯文本，库还支持一些增强功能。

发送文件附件： Poe的许多模型支持上传文件（如图片、PDF、Word、TXT等）并进行内容分析。

# 支持网络URL和本地文件路径
file_urls = [
    "https://arxiv.org/pdf/2303.08774.pdf",  # 一篇AI论文的URL
    "/Users/me/Documents/my_notes.txt"       # 本地文本文件路径
]

query = "请总结一下这两个文件的核心内容，并指出它们之间的关联。"
try:
    for chunk in client.send_message(bot="claude_3_5_sonnet",  # 使用支持长上下文和文件的Claude 3.5 Sonnet
                                     message=query,
                                     file_path=file_urls):
        print(chunk["response"], end='', flush=True)
except Exception as e:
    print(f"发送文件时出错: {e}")
    # 常见错误：文件大小超限、文件类型不支持、网络问题等

注意事项：不同模型对文件的支持程度和大小限制不同。例如，GPT-4o可能更擅长图像分析，而Claude-3.5-Sonnet-200k在处理长文本PDF方面更有优势。如果遇到文件相关错误，可以尝试换一个模型，或者检查文件是否过大（通常免费模型有更严格的限制）。

获取建议回复： Poe网页端在AI回复后经常会给出几个建议的后续问题。这个功能也可以通过API获取。

for chunk in client.send_message(bot="gpt4_o", 
                                 message="给我制定一个为期一周的Python学习计划",
                                 suggest_replies=True):  # 关键参数
    print(chunk["response"], end='', flush=True)

# 循环结束后，最后一个chunk中会包含'suggestedReplies'列表
if "suggestedReplies" in chunk:
    print("\n\n建议的后续问题：")
    for i, reply in enumerate(chunk["suggestedReplies"], 1):
        print(f"{i}. {reply}")

这个功能在构建聊天机器人时非常有用，你可以把这些建议回复作为按钮选项提供给用户，提升交互体验。

停止消息生成： 如果AI的回复太长或者方向不对，你可以中途停止。

import threading
import time

def ask_and_stop():
    bot = "chinchilla"  # GPT-3.5-Turbo
    question = "请详细描述古希腊哲学的发展史，从泰勒斯开始。"
    
    print("开始生成（将在3秒后停止）...")
    stop_event = threading.Event()
    
    def stop_generation():
        time.sleep(3)
        stop_event.set()
        print("\n[用户主动停止了生成]")
    
    # 启动一个线程在3秒后触发停止
    stopper = threading.Thread(target=stop_generation)
    stopper.start()
    
    for chunk in client.send_message(bot, question):
        if stop_event.is_set():
            # 调用cancel_message来中断生成
            client.cancel_message(chunk)
            break
        print(chunk["response"], end='', flush=True)

ask_and_stop()

重试上一条消息： 如果对AI的回复不满意，可以要求它重新生成。

# 首先发送一条消息并获取chatCode
for chunk in client.send_message(bot="a2", message="写一个关于友谊的简短寓言故事。"):
    print(chunk["response"], end='', flush=True)
    chat_code = chunk["chatCode"]

print("\n---第一次回复结束---\n")

# 不满意，重试一次
print("重试生成...")
for chunk in client.retry_message(chatCode=chat_code):
    print(chunk["response"], end='', flush=True)

retry_message 函数会使用相同的对话上下文和用户上一条消息，让AI重新生成回复。

5. 深入高级功能与集成方案

掌握了基础对话后，我们可以探索这个库更强大的能力，包括管理自定义机器人、使用知识库，以及最重要的——通过OpenAI兼容服务器将其集成到现有生态中。

5.1 创建与管理自定义机器人

Poe允许你基于现有模型创建具有自定义指令的机器人。 poe-api-wrapper 让你能用代码批量完成这个操作。

# 1. 首先，获取可用的基础模型（即你可以基于哪些模型创建机器人）
available_models = client.get_available_bots()
print("可用的创建模型:", available_models)
# 输出可能包含: ['chinchilla', 'a2', 'claude_3_5_sonnet', ...]

# 2. 创建一个自定义机器人
bot_data = client.create_bot(
    handle="my_python_tutor",  # 机器人的唯一标识符（URL的一部分）
    display_name="Python编程助手",
    prompt="你是一个耐心、专业的Python编程导师。你的回答应该清晰、有条理，并包含实际的代码示例。如果用户的问题不明确，你会主动询问细节。",  # 系统提示词
    base_model="chinchilla",  # 基于GPT-3.5-Turbo创建
    description="专注于解答Python从入门到进阶的所有问题。",
    intro_message="你好！我是你的Python专属助手。有什么编程问题需要我帮忙解决吗？",
    # 还可以设置图片、是否公开等参数
)

if bot_data.get("status") == "success":
    new_bot_id = bot_data["data"]["botId"]
    new_bot_codename = bot_data["data"]["botCode"]
    print(f"机器人创建成功！ID: {new_bot_id}, 代号: {new_bot_codename}")
    
    # 现在你可以像使用官方机器人一样使用它
    for chunk in client.send_message(bot=new_bot_codename, 
                                     message="如何用Python读取CSV文件？"):
        print(chunk["response"], end='', flush=True)
else:
    print(f"创建失败: {bot_data}")

# 3. 编辑已有的自定义机器人
edit_result = client.edit_bot(
    bot_code=new_bot_codename,
    prompt="你是一个专注于数据分析和可视化的Python专家。特别擅长使用pandas, numpy和matplotlib。",  # 更新提示词
    display_name="Python数据分析助手",  # 更新显示名
    base_model="gpt4_o",  # 甚至可以切换基础模型（如果订阅支持）
)
print(f"编辑结果: {edit_result}")

# 4. 删除自定义机器人
# delete_result = client.delete_bot(bot_code=new_bot_codename)
# print(f"删除结果: {delete_result}")

创建自定义机器人的核心参数解析：

• handle ：机器人在Poe平台URL中的唯一标识。例如， handle="my_bot" 对应的URL可能是 poe.com/my_bot 。需要确保唯一性。
• prompt ：这是机器人的“灵魂”，即系统提示词。它定义了机器人的角色、行为规范和回答风格。精心设计提示词是获得高质量回复的关键。
• base_model ：选择底层模型。免费用户通常只能选择免费的基模型（如 chinchilla , a2 , capybara 等）。订阅用户可以选择付费模型（如 gpt4_o , claude_3_opus 等）。
• intro_message ：用户开始新对话时，机器人发送的第一条问候消息。

5.2 为自定义机器人上传知识库

知识库功能允许你上传文档（TXT, PDF, DOCX等），让机器人在回答问题时参考这些文档的内容，实现基于私有资料的问答。

# 假设我们已经有一个自定义机器人的代号 `my_data_bot`

# 1. 获取该机器人当前的知识库列表（如果有的话）
kb_list = client.get_knowledge_bases(bot_code="my_data_bot")
print(f"现有知识库: {kb_list}")

# 2. 上传一个新的知识库文件
upload_result = client.upload_knowledge_base(
    bot_code="my_data_bot",
    file_path="/path/to/your/company_handbook.pdf",  # 本地文件路径
    # 或者使用URL
    # file_path="https://example.com/document.pdf",
    name="公司员工手册2024",  # 知识库显示名称
    description="包含公司政策、福利和规章制度。"
)

if upload_result.get("status") == "success":
    kb_id = upload_result["data"]["knowledgeBaseId"]
    print(f"知识库上传成功！ID: {kb_id}")
    
    # 3. 现在，向机器人提问，它会参考知识库内容
    question = "我们公司的年假政策是怎样的？"
    for chunk in client.send_message(bot="my_data_bot", message=question):
        print(chunk["response"], end='', flush=True)
        # 注意：回复中可能会包含引用标记，指向知识库的具体部分
else:
    print(f"上传失败: {upload_result}")

# 4. 编辑知识库信息（例如更新名称或描述）
# edit_kb_result = client.edit_knowledge_base(
#     knowledge_base_id=kb_id,
#     name="更新后的手册名称",
#     description="2024年最新修订版"
# )

# 5. 注意：库似乎没有提供直接删除知识库的方法，可能需要通过Poe网页端操作。

知识库使用心得：

• 文件格式与大小 ：支持常见格式，但解析效果可能因模型而异。文本文件（TXT）和PDF通常效果最好。注意免费账号有文件大小和数量的限制。
• 回复质量 ：机器人并不会100%严格遵循知识库，有时会结合自身知识。对于关键事实性问题，最好在提示词中强调“请严格根据提供的知识库内容回答”。
• 多文件管理 ：你可以为同一个机器人上传多个知识库文件。机器人会综合所有文件的内容来回答问题。

5.3 使用OpenAI兼容服务器进行集成

这是 poe-api-wrapper 最强大的功能之一。它启动一个本地HTTP服务器，这个服务器模拟了OpenAI API的接口。这意味着，任何使用OpenAI Python库 ( openai package) 的代码，只需修改API的基地址 ( base_url )，就能无缝切换到使用Poe背后的模型。

为什么这个功能如此重要？ 许多流行的AI应用框架和库（如LangChain、LlamaIndex、AutoGPT等）原生支持OpenAI API。通过这个兼容层，你可以让这些工具直接调用Claude、Gemini等模型，而无需重写任何适配代码。

部署与使用步骤：

1. 安装必要扩展并启动服务器 ：

   pip install -U 'poe-api-wrapper[llm]'
   

   # server.py
   from poe_api_wrapper import PoeServer
   
   # 可以配置多个账号令牌，服务器会轮询使用以平衡负载或绕过限制
   tokens = [
       {"p-b": "cookie1_p-b", "p-lat": "cookie1_p-lat"},
       {"p-b": "cookie2_p-b", "p-lat": "cookie2_p-lat"},
   ]
   
   # 启动服务器，默认监听 127.0.0.1:8000
   # 你可以指定地址和端口，例如让它在所有网络接口上监听
   server = PoeServer(tokens=tokens, address="0.0.0.0", port=8080)
   # 服务器会持续运行，直到你中断程序 (Ctrl+C)
   

2. 像使用OpenAI一样调用Poe模型 ：

   # client_example.py
   import openai
   
   # 关键：将base_url指向本地运行的PoeServer
   client = openai.OpenAI(
       api_key="anything",  # API Key可以任意填写，因为PoeServer不验证这个
       base_url="http://127.0.0.1:8000/v1/",  # 注意是 /v1/ 端点
       default_headers={"Authorization": "Bearer anything"}  # 同样，令牌任意
   )
   
   # 1. 列出可用模型（这里列出的是PoeServer映射的模型名）
   models = client.models.list()
   print("可用模型:", [model.id for model in models.data])
   
   # 2. 非流式聊天
   response = client.chat.completions.create(
       model="gpt-3.5-turbo",  # 映射到Poe的'chinchilla'
       messages=[
           {"role": "system", "content": "你是一个有用的助手。"},
           {"role": "user", "content": "你好，世界！"}
       ]
   )
   print("回复:", response.choices[0].message.content)
   
   # 3. 流式聊天
   stream = client.chat.completions.create(
       model="claude-instant",  # 映射到Poe的'a2'
       messages=[{"role": "user", "content": "讲一个笑话"}],
       stream=True
   )
   for chunk in stream:
       if chunk.choices[0].delta.content is not None:
           print(chunk.choices[0].delta.content, end='', flush=True)
   
   # 4. 图像生成（调用Poe的Stable Diffusion等图像模型）
   image_response = client.images.generate(
       model="playground-v2.5",  # 对应Poe的图像生成模型
       prompt="A serene landscape with mountains and a lake, digital art",
       n=1,
       size="1024x1024"
   )
   print("生成的图片URL:", image_response.data[0].url)
   

模型名称映射： PoeServer内部维护了一个从OpenAI风格模型名到Poe模型代号的映射。例如：

• gpt-3.5-turbo -> chinchilla (GPT-3.5-Turbo)
• gpt-4 -> beaver (GPT-4-Turbo)
• claude-instant -> a2
• claude-3.5-sonnet -> claude_3_igloo
• dall-e-3 -> playground-v2.5 (或其他Poe图像模型)

你可以在服务器日志或源代码中查看完整的映射关系。这个设计使得现有代码几乎无需修改。

高级用法示例：函数调用与视觉模型

# 函数调用（Function Calling）示例
# 许多Agent框架依赖此功能
import json

def get_weather(location):
    """模拟获取天气的函数"""
    return json.dumps({"location": location, "temperature": "22", "condition": "晴朗"})

client = openai.OpenAI(api_key="x", base_url="http://127.0.0.1:8000/v1/")

response = client.chat.completions.create(
    model="gpt-4o",  # 使用支持函数调用的模型
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "城市名"}
                },
                "required": ["location"]
            }
        }
    }]
)

# 处理函数调用请求...
# （此处省略具体的函数调用处理循环，与标准OpenAI流程一致）

# 视觉模型示例（上传图片）
import base64

def encode_image(image_path):
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

base64_image = encode_image("path/to/your/image.jpg")

vision_response = client.chat.completions.create(
    model="gpt-4o",  # 或 "claude-3.5-sonnet"
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "描述这张图片里有什么。"},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
                }
            ]
        }
    ]
)
print(vision_response.choices[0].message.content)

通过这个兼容服务器，你可以轻松地将Poe的强大模型集成到基于OpenAI生态构建的复杂应用中，极大地扩展了可能性。

6. 实战技巧、问题排查与性能优化

在实际使用中，你肯定会遇到各种问题。下面分享一些我踩过坑后总结的经验和解决方案。

6.1 常见错误与排查方法

错误现象	可能原因	解决方案
Authentication failed 或 Invalid token	1. p-b 或 p-lat Cookie 过期或无效。 2. Cookie 格式错误（包含了引号或多余空格）。 3. 账号被限制或封禁。	1. 重新登录Poe官网，获取新的Cookie。 2. 检查Cookie字符串，确保是纯文本值。 3. 检查账号状态，避免短时间内发送过多请求。
Rate limit exceeded	请求频率过高，触发Poe的速率限制。	1. 在请求间添加随机延迟（如 time.sleep(random.uniform(1, 3)) ）。 2. 使用多个账号令牌轮询（通过 PoeServer 的令牌列表或手动管理多个客户端）。 3. 降低并发请求数。
Bot not found 或 Invalid bot	1. 机器人代号拼写错误。 2. 该机器人对你不可用（如订阅专属模型但未订阅）。 3. 自定义机器人已被删除。	1. 对照README中的表格检查代号。 2. 确认你的账号是否有权限使用该模型。 3. 对于自定义机器人，确认其是否还存在。
连接超时或网络错误	1. 网络问题。 2. Poe服务器暂时不可用。 3. IP被临时限制。	1. 检查网络连接。 2. 重试几次，或等待一段时间。 3. 考虑使用代理（配置 proxy 参数）。
文件上传失败	1. 文件过大，超出模型限制。 2. 文件格式不支持。 3. 网络问题导致上传中断。	1. 尝试压缩文件或拆分内容。 2. 转换为支持的格式（如PDF、TXT）。 3. 重试上传，或检查文件URL是否可公开访问。
流式输出中断或不完整	1. 网络连接不稳定。 2. 服务器端生成中断。 3. 客户端处理超时。	1. 实现重试逻辑，从最后一个收到的chunk ID继续。 2. 检查是否触发了 cancel_message 。 3. 增加客户端超时设置（如果库支持）。
OpenAI兼容服务器返回 404 或 模型不支持	1. 服务器未正确启动。 2. 请求的模型名未在映射表中。 3. 请求的端点路径错误。	1. 检查服务器日志，确认PoeServer已成功启动并加载令牌。 2. 使用 client.models.list() 查看服务器支持的模型列表。 3. 确保请求的URL是 http://地址:端口/v1/chat/completions 。

6.2 性能优化与最佳实践

1. 连接复用与会话保持 ：避免为每条消息都创建新的 PoeApi 客户端实例。初始化一次，然后重复使用。客户端内部会管理会话和连接池。

2. 异步客户端用于高并发 ：如果你需要同时处理多个对话请求，务必使用 AsyncPoeApi 。结合 asyncio.gather 可以大幅提升效率。

   import asyncio
   from poe_api_wrapper import AsyncPoeApi
   
   async def ask_bot_async(bot_name, question):
       client = await AsyncPoeApi(tokens=tokens).create()
       full_text = ""
       async for chunk in client.send_message(bot_name, question):
           full_text += chunk.get("response", "")
       return full_text
   
   async def main():
       questions = [
           ("a2", "什么是机器学习？"),
           ("chinchilla", "用Python写一个快速排序函数。"),
           ("capybara", "讲一个励志小故事。")
       ]
       tasks = [ask_bot_async(bot, q) for bot, q in questions]
       results = await asyncio.gather(*tasks, return_exceptions=True)
       for i, (bot, _) in enumerate(questions):
           if isinstance(results[i], Exception):
               print(f"{bot} 出错: {results[i]}")
           else:
               print(f"{bot} 的回答摘要: {results[i][:100]}...")
   
   asyncio.run(main())
   

3. 实现简单的重试机制 ：网络请求难免失败，增加重试逻辑可以提高鲁棒性。

   import time
   from functools import wraps
   
   def retry_on_failure(max_retries=3, delay=2):
       def decorator(func):
           @wraps(func)
           def wrapper(*args, **kwargs):
               last_exception = None
               for attempt in range(max_retries):
                   try:
                       return func(*args, **kwargs)
                   except Exception as e:
                       last_exception = e
                       print(f"尝试 {attempt+1}/{max_retries} 失败: {e}")
                       if attempt < max_retries - 1:
                           time.sleep(delay * (attempt + 1))  # 指数退避
               raise last_exception
           return wrapper
       return decorator
   
   # 使用装饰器
   @retry_on_failure(max_retries=3, delay=1)
   def send_message_safe(client, bot, message):
       for chunk in client.send_message(bot, message):
           print(chunk["response"], end='', flush=True)
       return chunk["text"]
   

4. 合理设置超时与代理 ：对于不稳定的网络环境，在初始化客户端时配置合理的超时时间和代理是必要的。虽然库可能没有直接暴露所有 aiohttp 或 requests 的参数，但你可以通过修改库的底层会话配置或使用代理中间件来实现。

5. 监控使用量与成本 ：即使是免费模型，Poe也可能有每日限额。付费模型则直接消耗订阅点数。定期通过 client.get_settings() 检查剩余点数，并在代码中实现用量监控和预警。

   def check_usage(client):
       settings = client.get_settings()
       if 'data' in settings and 'remaining_limits' in settings['data']:
           limits = settings['data']['remaining_limits']
           for model, info in limits.items():
               if 'remaining' in info:
                   print(f"模型 {model} 剩余次数: {info['remaining']}")
       # 也可以检查订阅状态
       if 'data' in settings and 'subscription' in settings['data']:
           sub = settings['data']['subscription']
           print(f"订阅状态: {sub.get('status', 'N/A')}")
   

6.3 构建一个简单的自动化问答机器人示例

最后，我们结合以上所有知识，构建一个简单的控制台问答机器人，它可以选择不同的模型，并保持对话历史。

import os
import json
from poe_api_wrapper import PoeApi

class SimplePoeChatbot:
    def __init__(self, tokens_path="tokens.json"):
        """初始化机器人，从文件加载令牌"""
        self.load_tokens(tokens_path)
        self.client = PoeApi(tokens=self.tokens)
        self.current_bot = "chinchilla"  # 默认模型
        self.chat_history = {}  # 以 {bot: {chatCode: [messages]}} 格式存储
        self.active_chat_code = None
        
    def load_tokens(self, path):
        """从JSON文件加载令牌"""
        if os.path.exists(path):
            with open(path, 'r') as f:
                self.tokens = json.load(f)
            print("令牌已从文件加载。")
        else:
            # 提示用户输入并保存
            print("未找到令牌文件，请手动输入：")
            p_b = input("请输入 p-b cookie 值: ").strip()
            p_lat = input("请输入 p-lat cookie 值: ").strip()
            self.tokens = {'p-b': p_b, 'p-lat': p_lat}
            with open(path, 'w') as f:
                json.dump(self.tokens, f)
            print(f"令牌已保存至 {path}")
    
    def list_available_bots(self):
        """列出一些常用的官方机器人"""
        common_bots = {
            "1": {"code": "chinchilla", "name": "GPT-3.5-Turbo (免费)"},
            "2": {"code": "gpt4_o", "name": "GPT-4o (免费/订阅)"},
            "3": {"code": "a2", "name": "Claude-instant (免费)"},
            "4": {"code": "claude_3_5_sonnet", "name": "Claude-3.5-Sonnet (免费)"},
            "5": {"code": "capybara", "name": "Assistant (免费)"},
        }
        print("\n可用机器人:")
        for key, bot in common_bots.items():
            print(f"  [{key}] {bot['name']} ({bot['code']})")
        return common_bots
    
    def select_bot(self):
        """让用户选择机器人"""
        bots = self.list_available_bots()
        choice = input("\n请选择机器人编号 (直接回车使用当前机器人): ").strip()
        if choice and choice in bots:
            self.current_bot = bots[choice]["code"]
            print(f"已切换到机器人: {bots[choice]['name']}")
            # 切换机器人时，可以开启新会话或沿用历史（这里开启新会话）
            self.active_chat_code = None
        else:
            print(f"继续使用当前机器人: {self.current_bot}")
    
    def send_message(self, user_input):
        """发送消息并处理回复"""
        try:
            full_reply = ""
            print(f"\n[{self.current_bot}] 正在思考...", end='', flush=True)
            
            # 发送消息，如果存在活跃会话则继续，否则新建
            for chunk in self.client.send_message(
                bot=self.current_bot,
                message=user_input,
                chatCode=self.active_chat_code
            ):
                text_chunk = chunk.get("response", "")
                print(text_chunk, end='', flush=True)
                full_reply += text_chunk
                
                # 如果是新对话，保存chatCode
                if self.active_chat_code is None and "chatCode" in chunk:
                    self.active_chat_code = chunk["chatCode"]
                    # 初始化该机器人的历史记录
                    if self.current_bot not in self.chat_history:
                        self.chat_history[self.current_bot] = {}
                    if self.active_chat_code not in self.chat_history[self.current_bot]:
                        self.chat_history[self.current_bot][self.active_chat_code] = []
            
            # 保存对话历史（简化版，只保存最近几条）
            if self.active_chat_code:
                history = self.chat_history[self.current_bot][self.active_chat_code]
                history.append({"role": "user", "content": user_input})
                history.append({"role": "assistant", "content": full_reply})
                # 保持历史记录不超过10轮
                if len(history) > 20:
                    self.chat_history[self.current_bot][self.active_chat_code] = history[-20:]
            
            print()  # 换行
            return full_reply
            
        except Exception as e:
            print(f"\n发送消息时出错: {e}")
            return None
    
    def clear_context(self):
        """清除当前对话的上下文（Chat Break）"""
        if self.active_chat_code:
            try:
                self.client.chat_break(self.current_bot, chatCode=self.active_chat_code)
                print("上下文已清除。")
                # 不清除本地历史，但新消息将开始新上下文
            except Exception as e:
                print(f"清除上下文时出错: {e}")
        else:
            print("没有活跃的对话。")
    
    def run(self):
        """运行主交互循环"""
        print("=== Poe API 简单聊天机器人 ===")
        print("命令: /bot 切换模型, /clear 清除上下文, /history 查看历史, /exit 退出")
        
        while True:
            try:
                user_input = input("\n你: ").strip()
                if not user_input:
                    continue
                    
                if user_input.lower() == '/exit':
                    print("再见！")
                    break
                elif user_input.lower() == '/bot':
                    self.select_bot()
                elif user_input.lower() == '/clear':
                    self.clear_context()
                elif user_input.lower() == '/history':
                    self.show_history()
                else:
                    self.send_message(user_input)
                    
            except KeyboardInterrupt:
                print("\n\n程序被中断。")
                break
            except Exception as e:
                print(f"发生错误: {e}")
    
    def show_history(self):
        """显示当前对话的历史记录"""
        if self.active_chat_code and self.current_bot in self.chat_history:
            history = self.chat_history[self.current_bot][self.active_chat_code]
            print(f"\n当前对话历史 (共{len(history)//2}轮):")
            for i, msg in enumerate(history):
                role = "用户" if msg["role"] == "user" else "AI"
                # 只显示前100个字符预览
                preview = msg["content"][:100] + ("..." if len(msg["content"]) > 100 else "")
                print(f"  {i+1}. [{role}] {preview}")
        else:
            print("没有可显示的历史记录。")

if __name__ == "__main__":
    bot = SimplePoeChatbot()
    bot.run()

这个示例展示了如何构建一个结构清晰、功能实用的交互程序。你可以在此基础上扩展更多功能，比如支持文件上传、多轮对话记忆、将历史记录保存到数据库等。

7. 项目现状、替代方案与未来展望

正如项目README开头明确警告的， snowby666/poe-api-wrapper 目前处于 “不再积极维护” 的状态。这对于考虑将其用于生产环境的开发者来说，是一个重要的风险点。

主要风险包括：

1. 接口失效风险 ：Poe.com 的前端或后端API一旦发生变更，这个逆向工程出来的封装库就可能完全无法工作，且无人修复。
2. 安全风险 ：依赖Cookie进行认证的方式本身较为脆弱，且账号有因自动化行为被限制的风险。
3. 功能缺失 ：Poe平台新增的功能（如新的模型、特性）可能不会及时加入到这个库中。

官方替代方案：Poe官方API Quora（Poe的母公司）已经推出了官方的Poe API。与这个封装库相比，官方API：

• 优点 ：稳定、可靠、有官方支持、文档完善、使用标准的API Key认证（更安全）、明确的使用条款和价格。
• 缺点 ：通常是付费服务，可能有调用频率或功能限制，且不一定能完全覆盖网页版的所有功能（特别是免费模型的使用方式可能不同）。

其他替代思路：

1. 直接使用各模型提供商的官方API ：例如OpenAI API、Anthropic Claude API、Google Gemini API等。这提供了最大的控制权和稳定性，但成本可能更高，且需要管理多个API密钥和不同的调用方式。
2. 使用其他聚合平台 ：一些云服务或开源项目提供了统一的AI模型网关，可以聚合多个后端的API。
3. 继续使用本项目，但做好风险隔离 ：如果你确实需要Poe上特定的免费模型或功能，可以继续使用这个库，但建议：
   • 将其用于非核心、可降级的业务逻辑。
   • 实现完善的错误处理和回退机制（例如，当此库失败时，切换到官方API或另一个模型）。
   • 密切关注Poe平台的更新，并做好随时重写相关代码的准备。

个人建议与总结 从我个人的使用经验来看， poe-api-wrapper 在其活跃期是一个非常出色且强大的工具。它极大地降低了开发者探索和利用Poe平台能力的门槛。对于以下场景，它依然有很高的价值：

• 快速原型验证 ：在决定使用哪个AI模型或设计提示词时，快速进行测试和比较。
• 个人自动化脚本 ：用于处理个人事务的自动化，如自动总结文章、分类邮件、生成内容草稿等。
• 学习和研究 ：学习如何与复杂的Web应用进行逆向工程交互，理解API封装的设计模式。

然而，对于需要 7x24小时稳定运行 的商业项目或关键业务，我强烈建议评估并迁移到 Poe官方API 或各模型的 官方API 。稳定性、安全性和长期支持是生产环境不可或缺的要素。

这个项目的代码本身也是一个很好的学习资源，它展示了如何处理复杂的Web会话、管理状态、实现流式响应和构建兼容层。即使未来不再使用它，从中汲取的设计思路和问题解决方案，对任何从事API集成或自动化工作的开发者来说，都是一笔宝贵的财富。技术的世界就是这样，旧工具的落幕常常为新工具的诞生铺平道路，而理解其原理的人，总能更快地适应变化。