用Deepseek-v3.1在Trae中编写AI中继程序

用 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或更新后的openaiSDK 已经可以绕过部分限制。但这类中继依然有不可替代的价值：

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

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

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

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

from fastapi import FastAPI, Request, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse import httpx import json 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.json() messages = data.get("messages", []) if not messages: raise HTTPException(status_code=400, detail="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/json", "Authorization": f"Bearer {XINGHE_API_KEY}" }, json={ "model": model_name, "messages": ernie_messages, "temperature": temperature, "stream": stream } ) if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail=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.json() except Exception as e: raise HTTPException(status_code=500, detail=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(app, host="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/json' \ --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.content, end="", 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: Request, credentials: HTTPAuthorizationCredentials = Depends(security)): if not verify_token(credentials.credentials): raise HTTPException(status_code=401, detail="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框架

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