moon签名代理平台推荐🏀TG:@yyffvip.tmd

用 Deepseek-v3.1 在 Trae 中构建 AI 中继服务：打通国产大模型与 OpenAI 生态的桥梁

在本地开发中，我们常遇到这样一个问题：许多优秀的国产大模型（如百度飞桨星河社区部署的 ERNIE 系列）虽然性能强劲、中文理解出色，但其 API 接口规范和认证机制与主流工具链不兼容。比如 Auto-Coder、Moon Pilot 这类基于 OpenAI 协议设计的智能编程助手，无法直接调用星河平台的模型接口——哪怕你把 base_url 改了，也会因为签名方式不同而失败。

于是，一个轻量级 AI 中继服务 就成了理想的解决方案。它像一座翻译桥，接收标准 OpenAI 格式的请求，将其“转译”成百度星河能理解的形式，转发过去后再把响应还原回 OpenAI 兼容格式返回。这样一来，所有遵循 OpenAI 接口规范的客户端都能无缝接入国产大模型，无需修改任何逻辑代码。

✅ 当前对接模型：ERNIE-4.5-21B-A3B-Paddle
📍 部署地址：飞桨AI Studio星河社区

为什么需要这个中继？

有人可能会问：“现在不是有些工具已经支持直连了吗？” 确实，较新版本的 auto-coder 或更新后的 openai SDK 已经可以绕过部分限制。但这类中继依然有不可替代的价值：

统一管理访问入口：团队内部可通过一个中继集中控制 API Key、做权限审计。
协议隔离与适配：不只是百度，未来想换通义千问、百川、GLM，只需改后端，前端完全无感。
增强能力扩展性：你可以轻松加入缓存、限流、日志记录、敏感词过滤等中间件功能。
安全隔离：避免将真实密钥暴露给前端应用或第三方插件。
调试友好：方便抓包分析请求结构，快速定位兼容性问题。

换句话说，中继不是“绕路”，而是为国产大模型落地提供了一种更工程化、可运维的接入路径。

快速实现：FastAPI + Deepseek-v3.1 自动生成核心逻辑

整个中继服务使用 FastAPI 编写，简洁高效，仅需几十行代码即可完成核心代理逻辑。以下是由 Deepseek-v3.1 模型在 Trae 环境中自动生成 的完整服务代码：

from fastapi import FastAPIRequestHTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
import httpx
import on

app = FastAPI(title="PaddlePaddle ERNIE 大模型中继服务")

# 启用CORS以允许跨域请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 星河平台 API 配置
XINGHE_BASE_URL = "https://api-r9z6e7xbe854gbi0.aistudio-app.com/v1"
XINGHE_API_KEY = "6cac673af748cec3440270xxxx"  # 替换为你自己的Key

@app.post("/v1/chat/completions")
async def proxy_to_ernie(request: Request):
    try:
        # 1. 解析来自OpenAI客户端的标准请求
        data = await request.on()

        messages = data.get("messages"[])
        if not messages:
            raise HTTPException(status_code=400detail="Missing 'messages' field in request body")

        # 构建适配星河平台的消息格式
        ernie_messages = []
        for msg in messages:
            role = msg.get("role""user")
            content = msg.get("content""")
            ernie_messages.append({"role": role"content": content})

        # 获取其他参数
        model_name = data.get("model""default")
        temperature = data.get("temperature"0.6)
        stream = data.get("stream"False)

        # 2. 转发至星河平台
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{XINGHE_BASE_URL}/chat/completions",
                headers={
                    "Content-Type": "application/on",
                    "Authorization": f"Bearer {XINGHE_API_KEY}"
                },
                on={
                    "model": model_name,
                    "messages": ernie_messages,
                    "temperature": temperature,
                    "stream": stream
                }
            )

            if response.status_code != 200:
                raise HTTPException(status_code=response.status_codedetail=response.text)

            # 3. 返回响应（区分流式与非流式）
            if stream:
                return StreamingResponse(
                    response.aiter_bytes(),
                    media_type="text/event-stream",
                    headers={
                        "Cache-Control": "no-cache",
                        "Connection": "keep-alive"
                    }
                )
            else:
                return response.on()

    except Exception as e:
        raise HTTPException(status_code=500detail=f"Server error: {str(e)}")

@app.get("/")
async def root():
    return {
        "message": "PaddlePaddle ERNIE 中继服务",
        "status": "running",
        "endpoint": "/v1/chat/completions"
    }

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(apphost="0.0.0.0"port=8000)

这段代码实现了几个关键点：

完全兼容 OpenAI /v1/chat/completions 接口；
支持 stream=True 的流式输出，用户体验无损；
使用异步客户端 httpx 实现高效代理，避免阻塞；
自动透传 temperature、model 等常用参数；
错误处理完善，状态码和错误信息原样传递。

值得一提的是，该代码由 Deepseek-v3.1 在 Trae 中一次性生成并通过测试，展现了当前国产大模型在实际工程任务中的强大辅助能力。

如何启动服务？

保存上述代码为 main.py 后，有两种方式运行：

方法一：推荐使用 Uvicorn（适合开发）

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

--reload 参数会在文件修改时自动重启服务，非常适合调试阶段。

方法二：直接运行脚本

python main.py

服务启动后，默认监听 http://0.0.0.0:8000，你可以通过浏览器访问：

http://localhost:8000 —— 查看服务基本信息
http://localhost:8000/health —— 健康检查接口

只要返回 {"status": "healthy"}，说明服务已就绪。

使用 curl 测试中继是否正常工作

确保服务运行后，执行如下命令发起一次非流式请求：

curl 'http://127.0.0.1:8000/v1/chat/completions' \
  --header 'Content-Type: application/on' \
  --data '{
    "model": "default",
    "messages": [
      {"role": "user""content": "你好，请介绍一下你自己"}
    ],
    "temperature": 0.6,
    "stream": false
  }'

如果一切正常，你会收到类似下面的 JSON 响应：

{
  "id": "asst-xxx",
  "object": "chat.completion",
  "created": 1718765432,
  "model": "default",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "我是由百度研发的ERNIE大模型……"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 12,
    "completion_tokens": 105,
    "total_tokens": 117
  }
}

这说明中继成功完成了协议转换和代理转发。

使用 Python 客户端验证：OpenAI SDK 直接调用

由于中继完全兼容 OpenAI 接口，我们可以直接使用官方 openai 包进行集成测试。

安装依赖

pip install openai -U

⚠️ 注意：必须升级到最新版（建议 ≥1.105.0），否则会报错 TypeError: Completions.create() got an unexpected keyword argument 'stream_options'。

创建测试脚本 `test_client.py`

from openai import OpenAI

client = OpenAI(
    api_key="any_key" # 实际不使用，由中继服务内部配置
    base_url="http://localhost:8000/v1"
)

def test_completion():
    print("👉 正在测试非流式请求...")
    try:
        resp = client.chat.completions.create(
            model="default",
            messages=[{"role": "user""content": "请用中文写一段关于春天的描述"}],
            temperature=0.7,
            stream=False
        )
        print(f"✅ 成功收到回复：\n{resp.choices[0].message.content}")
        print(f"📊 使用统计：{resp.usage}")
    except Exception as e:
        print(f"❌ 请求失败：{e}")

def test_streaming():
    print("\n👉 正在测试流式请求...")
    try:
        stream = client.chat.completions.create(
            model="default",
            messages=[{"role": "user""content": "讲一个程序员的幽默故事"}],
            temperature=0.8,
            stream=True
        )
        print("💬 流式输出开始：")
        for chunk in stream:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.contentend=""flush=True)
        print("\n✅ 流式响应完成")
    except Exception as e:
        print(f"❌ 流式请求失败：{e}")

if __name__ == "__main__":
    test_completion()
    test_streaming()

运行脚本：

python test_client.py

如果看到文本逐步打印出来，并且 usage 数据完整返回，恭喜你！你的中继服务已经具备生产级可用性。

在 auto-coder 中集成中继模型

auto-coder 是一款强大的自动化编程工具，支持自定义模型接入。通过我们的中继服务，可以让它使用 PaddlePaddle 上的 ERNIE 模型。

步骤一：启动 auto-coder

auto-coder.chat

步骤二：添加中继模型配置

/models /add_model name=paddle_ernie model_name=default base_url=http://127.0.0.1:8000/v1
/models /add paddle_ernie your_fake_api_key
/conf model:paddle_ernie

🔐 注意：这里的 your_fake_api_key 可以是任意字符串，因为中继服务不会验证客户端密钥。

步骤三：提交任务测试

输入需求，例如：

请帮我设计一个简单的Flask API，提供用户注册和登录功能。

观察是否能顺利生成代码。若无认证错误且响应流畅，则说明集成成功。

常见问题与排查建议

❌ 报错：`Connection refused`

原因：中继服务未启动或端口被占用
解决方法：
- 检查 uvicorn 是否正在运行
- 更换端口：uvicorn main:app --port 8080
- 确认防火墙设置是否放行对应端口

❌ 报错：`TypeError: Completions.create() got an unexpected keyword argument 'stream_options'`

原因：openai 库版本过低（<1.0）
解决方法：

pip install openai -U

❌ 返回空响应或超时

原因：星河平台响应慢或网络不稳定
解决方法：
- 提高客户端超时时间（如设为 timeout=60.0）
- 检查 XINGHE_BASE_URL 和 XINGHE_API_KEY 是否正确
- 登录星河社区查看模型部署状态和调用日志

中继服务的演进方向

虽然目前这个中继已能满足基本需求，但它只是一个起点。接下来可以考虑以下几个增强方向：

1. 容器化部署（Docker）

便于在服务器、Kubernetes 或 CI/CD 流程中标准化部署。

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn""main:app""--host""0.0.0.0""--port""8000"]

2. 添加 JWT 鉴权

支持多用户访问控制，防止滥用。

from fastapi.security import HTTPBearer
security = HTTPBearer()

@app.post("/v1/chat/completions")
async def proxy_to_ernie(request: Requestcredentials: HTTPAuthorizationCredentials = Depends(security)):
    if not verify_token(credentials.credentials):
        raise HTTPException(status_code=401detail="Invalid token")
    # ...

3. 暴露 Prometheus 指标

用于监控调用量、延迟、成功率等关键指标。

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

4. 多模型路由 + fallback 机制

根据 model 字段分发到不同后端，甚至结合 PaddleHub 提供本地模型降级兜底。

if model_name == "ernie":
    backend_url = XINGHE_BASE_URL
elif model_name == "qwen":
    backend_url = QWEN_BASE_URL
else:
    backend_url = LOCAL_PADDLE_MODEL_URL

总结：从一个小中继看国产AI生态的连接力

这个看似简单的中继程序，背后承载的是国产大模型走向实用化的关键一步。它不仅解决了“能不能用”的问题，更打开了“如何更好用”的可能性。

通过这样一个轻量级服务，我们实现了：

✅ 国产大模型（ERNIE）替代 GPT 系列
✅ 零成本迁移现有 OpenAI 应用
✅ 支持流式输出，体验无损
✅ 本地部署，保障数据隐私
✅ 代码清晰，易于维护和二次开发

更重要的是，这种模式具有高度可复制性——无论是对接通义千问、百川、GLM 还是自研私有模型，都可以沿用相同的架构思路。

🌱 技术栈组合建议：

开发：Trae + Deepseek-v3.1（智能编码）
模型：PaddlePaddle / ERNIE（中文强项）
服务：FastAPI + Uvicorn + Docker（现代Python生态）
应用：auto-coder / Moon Pilot / 自研Agent框架

让国产深度学习平台真正“跑起来”，往往不需要惊天动地的大工程。有时候，一个小小的中继，就是连接创新与落地的关键纽带。