ChatGPT实用插件开发指南：从零构建你的第一个生产力工具

1次阅读

没有评论

共计 3190 个字符，预计需要花费 8 分钟才能阅读完成。

当前 ChatGPT 生态中存在一个显著问题：虽然 AI 能力强大，但实际工作流程中往往需要频繁切换不同工具。比如：

写代码时要手动复制结果到 IDE
分析数据时需要导出到 Excel
生成 Markdown 后还要粘贴到渲染工具

这种割裂体验严重拖累效率。而插件 (Plugin) 机制正是为解决这一问题而生——它允许开发者将第三方服务直接接入 ChatGPT 对话流，实现无缝衔接的 AI 生产力闭环。

在集成 ChatGPT 时，开发者常见两种方式：

Webhooks
工作方式：用户触发→ChatGPT 调用预设 URL→获取外部数据
优点：实现简单，适合已有 API 的服务
缺点：响应延迟高（需完整执行才返回），无法实时交互
Plugin
工作方式：通过 OpenAPI 规范声明能力，ChatGPT 动态调用
优点：支持流式响应(Streaming)，权限控制更精细
典型场景：需要渐进式展示结果的场景（如代码生成、数据分析）

@startuml
participant User
participant ChatGPT
participant Plugin

User -> ChatGPT: 输入请求
ChatGPT -> Plugin: 调用 OpenAPI 端点
Plugin --> ChatGPT: 流式返回数据
ChatGPT -> User: 渐进式显示结果
@enduml

确保 Python≥3.8 环境，安装依赖：

pip install fastapi uvicorn python-multipart

创建基础服务结构：

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="Markdown Converter Plugin",
    description="即时将 Markdown 转换为 HTML",
    version="0.1.0"
)

@app.get("/.well-known/ai-plugin.json")
async def plugin_manifest():
    return {
        "schema_version": "v1",
        "name_for_human": "Markdown 转换器",
        "name_for_model": "markdown_converter",
        "description_for_human": "将 Markdown 文本即时转换为 HTML",
        "description_for_model": "...",
        "auth": {"type": "none"},
        "api": {"type": "openapi", "url": "/openapi.json"}
    }

对于需要用户认证的插件，建议使用 OAuth2.0 标准。以下是密码模式示例：

# auth.py
from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    if token != "mock_secret":
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

关键技巧是使用 FastAPI 的StreamingResponse：

# converter.py
import markdown
from fastapi import Response

async def convert_markdown(text: str):
    html = ""for line in text.split('\n'):
        html += markdown.markdown(line) + "\n"
        yield html  # 逐步生成结果

@app.post("/convert")
async def convert(text: str = Body(..., media_type="text/plain"),
    _: str = Depends(verify_token)
):
    return StreamingResponse(convert_markdown(text),
        media_type="text/html"
    )

使用令牌桶算法控制请求频率：

# ratelimit.py
from fastapi import Request
from fastapi.middleware import Middleware

class TokenBucket:
    def __init__(self, capacity: int, fill_rate: float):
        self.capacity = capacity
        self.tokens = capacity
        self.last_fill = time.time()
        self.fill_rate = fill_rate

    def consume(self):
        now = time.time()
        elapsed = now - self.last_fill
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.fill_rate
        )
        self.last_fill = now
        if self.tokens < 1:
            return False
        self.tokens -= 1
        return True

通过中间件拦截请求 / 响应：

# middleware.py
from fastapi import Request

async def sanitize_middleware(request: Request, call_next):
    # 请求过滤
    if "password" in await request.body():
        raise HTTPException(400, "敏感字段禁止传输")

    response = await call_next(request)

    # 响应过滤
    if "内部错误" in response.text:
        response.text = response.text.replace("内部错误", "系统异常")
    return response

创建locustfile.py：

from locust import HttpUser, task

class PluginUser(HttpUser):
    @task
    def convert_markdown(self):
        self.client.post("/convert", 
            headers={"Authorization": "Bearer mock_secret"},
            data="## Hello World"
        )

运行测试：