用Deepseek-v3.1在Trae中构建AI中继程序

用 Deepseek-v3.1 在 Trae 中构建 AI 中继程序

在本地开发 AI 工具链时，你是否也遇到过这样的困境：明明国产大模型能力强大、响应迅速，但自家的 IDE 插件、自动化代码生成器却“认不出”它？不是报错Invalid API Key，就是卡在流式传输上动弹不得。问题不在于模型不行，而在于——生态协议不兼容。

比如百度飞桨 AI Studio 星河社区提供的 ERNIE 大模型，虽然支持类 OpenAI 的 RESTful 接口（/v1/chat/completions），但它的认证逻辑和参数处理方式与标准openai>=1.0客户端存在微妙差异。很多工具如 auto-coder、moonpilot 或 VS Code 插件依赖严格的 header 校验、SSE 流控机制，稍有不符就会中断连接。

更麻烦的是，这些工具通常不允许自定义 CA 证书或跳过 HTTPS 验证，导致本地代理难以介入。

于是，我尝试了一种“轻量破局”方案：借助Trae + Deepseek-v3.1，让 AI 自己写一个中继网关，把“说不同语言”的两方拉到同一个频道上来。

整个过程令人惊讶地顺畅——从需求描述到可运行服务，仅用几分钟，且几乎无需手动调试。最终成果是一个基于 FastAPI 的本地代理层，能将任何符合 OpenAI 规范的请求无缝转发至星河社区的大模型接口，并自动完成密钥注入、格式转换与流式适配。

这个中继服务的核心价值，其实不在“转发”，而在“翻译”。

我们来看几个典型痛点：

• 星河 API 不验证客户端传入的Authorization: Bearer xxx，但大多数 openai 客户端会强制发送；
• 某些字段如stream_options被识别为非法参数，低版本 SDK 直接抛异常；
• 模型名映射混乱，gpt-3.5-turbo应该对应哪个 ERNIE 版本？
• SSE 响应头缺失text/event-stream类型，前端接收失败。

如果每个项目都去改插件源码，成本太高。不如建一层统一出口：所有请求先打到本地中继，由它来处理兼容性问题。

这就像给老房子装了个智能网关——不用拆墙换线，也能接入现代智能家居系统。

所用模型部署于飞桨 AI Studio 星河社区，例如ERNIE-4.5-21B-A3B-Paddle，基于 PaddlePaddle 框架训练，在中文理解、代码生成、文档摘要等任务上表现优异。PaddlePaddle 作为国产全场景深度学习平台，不仅提供动态图编程体验，还拥有 PaddleOCR、PaddleDetection、PaddleNLP 等工业级工具库，非常适合企业级落地。

而我们的目标很明确：让这些高性能模型，能像调用 GPT 一样简单地集成进现有工作流。

在 Trae 编辑器中输入提示词：

“请帮我写一个 FastAPI 实现的 AI 中继服务，将标准 OpenAI 格式的/v1/chat/completions请求转发到百度星河社区的大模型 API。要求支持 JSON 和 SSE 流式响应，自动注入 API Key，并允许 CORS。”

几秒后，Deepseek-v3.1 输出了完整代码。稍作调整后如下：

中继程序代码（main.py）

from fastapi import FastAPI, Request, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse import httpx import os app = FastAPI(title="PaddlePaddle 星河大模型中继服务", version="1.0.0") # 允许跨域请求（便于前端工具调用） app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 从环境变量读取星河 API 配置（安全起见不要硬编码） XINGHE_BASE_URL = os.getenv("XINGHE_BASE_URL", "https://api-r9z6e7xbe854gbi0.aistudio-app.com/v1") XINGHE_API_KEY = os.getenv("XINGHE_API_KEY", "") if not XINGHE_API_KEY: raise RuntimeError("请设置环境变量 XINGHE_API_KEY") @app.post("/v1/chat/completions") async def proxy_chat_completions(request: Request): try: # 1. 解析客户端发来的 OpenAI 格式请求 data = await request.json() messages = data.get("messages") if not messages: raise HTTPException(status_code=400, detail="Missing 'messages' in request body") # 提取关键参数 model = data.get("model", "default") temperature = data.get("temperature", 0.7) stream = data.get("stream", False) # 构造符合星河 API 的 payload payload = { "model": model, "messages": [ {"role": msg["role"], "content": msg["content"]} for msg in messages ], "temperature": temperature, "stream": stream } # 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=payload ) # 3. 处理返回结果 if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"Upstream error: {response.text}" ) if stream: # 流式响应：直接转发 SSE 数据流 return StreamingResponse( response.aiter_bytes(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive" } ) else: # 非流式：返回 JSON 响应 return response.json() except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/") async def index(): return { "service": "PaddlePaddle 星河大模型中继", "status": "running", "endpoint": "/v1/chat/completions", "docs": "/docs" } @app.get("/health") async def health_check(): return {"status": "healthy"} # 启动命令：uvicorn main:app --host 0.0.0.0 --port 8000 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

这段代码有几个关键设计点值得强调：

• 使用httpx.AsyncClient实现非阻塞转发，避免高并发下线程堆积；
• 对stream=True的请求，直接通过StreamingResponse转发字节流，保持低延迟；
• 所有敏感配置均来自环境变量，防止密钥意外提交至 Git；
• 添加健康检查端点/health，方便容器化部署时做探针检测。

特别值得一提的是，星河 API 返回的 SSE 数据结构与 OpenAI 几乎一致，因此无需逐条解析再重组 event stream，极大简化了实现复杂度。

快速部署步骤

1. 保存文件

将上述代码保存为main.py。

2. 安装依赖

pip install fastapi uvicorn httpx python-multipart

注意：不需要安装openai包，除非你要做客户端测试。

3. 设置环境变量

Linux/macOS:

export XINGHE_API_KEY="your_actual_api_key_here" export XINGHE_BASE_URL="https://api-r9z6e7xbe854gbi0.aistudio-app.com/v1"

Windows（CMD）:

set XINGHE_API_KEY=your_actual_api_key_here set XINGHE_BASE_URL=https://api-r9z6e7xbe854gbi0.aistudio-app.com/v1

API Key 可在 飞桨 AI Studio 的模型部署页面获取。

4. 启动服务

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

启动后访问http://localhost:8000/docs，即可看到自动生成的交互式文档界面，支持在线测试。

接口测试验证

非流式请求测试（curl）

curl -X POST "http://localhost:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{ "model": "default", "messages": [ {"role": "user", "content": "你好，请用中文写一首关于秋天的诗"} ], "temperature": 0.8, "stream": false }'

预期输出示例：

{ "id": "log-xxxxx", "object": "chat.completion", "created": 1712345678, "model": "default", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "秋风起兮白云飞，草木黄落兮雁南归...\n金菊绽放香满园，枫叶如火映山川。" }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 25, "completion_tokens": 38, "total_tokens": 63 } }

说明：中继已成功完成请求转换并返回完整响应。

流式请求测试（SSE）

curl -X POST "http://localhost:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{ "model": "default", "messages": [ {"role": "user", "content": "介绍一下 PaddlePaddle 的特点"} ], "stream": true }'

你会看到类似以下输出：

data: {"id":"...","choices":[{"delta":{"content":"PaddlePaddle"}}]} data: {"id":"...","choices":[{"delta":{"content":"是"}}]} data: {"id":"...","choices":[{"delta":{"content":"百度"}}]} ... data: [DONE]

这表明流式传输完全正常，适用于需要实时反馈的场景，如代码补全、对话机器人等。

使用 OpenAI 客户端测试（Python）

为了验证与主流工具链的兼容性，使用官方openai客户端进行测试：

pip install openai -U # 建议 >= 1.10

测试脚本（test_client.py）

from openai import OpenAI # 指向本地中继服务 client = OpenAI( api_key="any_key", # 中继会忽略此值 base_url="http://localhost:8000/v1" ) def test_completion(): print("👉 发起非流式请求...") resp = client.chat.completions.create( model="default", messages=[{"role": "user", "content": "什么是 PaddleOCR？"}], temperature=0.6, stream=False ) print(f"✅ 模型回复:\n{resp.choices[0].message.content}") print(f"📊 使用统计: {resp.usage}") def test_streaming(): print("\n👉 发起流式请求...") stream = client.chat.completions.create( model="default", messages=[{"role": "user", "content": "简述飞桨的三大优势"}], stream=True ) print("🔄 实时输出:") for chunk in stream: if content := chunk.choices[0].delta.content: print(content, end="", flush=True) print("\n✅ 流式响应结束") if __name__ == "__main__": test_completion() test_streaming()

运行结果表明：无论是普通请求还是流式响应，行为与调用原生 OpenAI 接口几乎无异。

这意味着，几乎所有基于openaiSDK 的工具（LangChain、LlamaIndex、AutoGen 等）都可以零改造接入。

与 auto-coder 集成实战

auto-coder 是当前热门的 LLM 辅助编程工具，支持多模型切换。我们来实测其与中继服务的协同效果。

步骤 1：启动 auto-coder

auto-coder chat

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

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

注：此处dummy_key可为任意字符串，因为中继服务会使用预设的真实密钥。

步骤 3：开始对话

输入指令：

请帮我写一个 Python 函数，使用 PaddleNLP 实现情感分析。

auto-coder 成功调用中继服务，返回了一段结构清晰、语法正确的代码片段，并附带使用说明。整个过程流畅自然，用户完全感知不到背后经过了一层代理。

这正是我们想要的效果：能力透明化接入，开发无感化升级。

进阶优化方向

当前中继已可用，但若用于生产环境，建议增加以下功能：

例如，添加一个简单的日志中间件：

from datetime import datetime @app.middleware("http") async def log_requests(request: Request, call_next): start = datetime.now() response = await call_next(request) duration = (datetime.now() - start).total_seconds() print(f"[{start}] {request.method} {request.url.path} → {response.status_code} ({duration:.2f}s)") return response

这类扩展非常灵活，可根据团队实际需求逐步叠加，不必一开始就追求大而全。

这种基于 AI 自动生成的轻量中继架构，本质上是一种“协议翻译器”，正在成为连接开放生态与私有平台的重要桥梁。

它不仅解决了当下工具链兼容性的问题，更为未来的技术演进留足空间——比如多模型 A/B 测试、成本最优调度、私有模型与公有云混合编排等。

更重要的是，整个服务由 Deepseek-v3.1 在 Trae 中自主生成并验证，展示了新一代 AI 编程助手的强大潜力：它们不仅能写函数、修 Bug，还能帮你搭建基础设施。

当你发现某个工具“不支持某模型”时，不妨试试让它自己写个适配层。有时候，最高效的解决方案，就是让 AI 来解决 AI 的问题。

💡 小贴士：如果你也在使用其他国产模型（如通义、GLM、百川），只需更改XINGHE_BASE_URL和认证方式，即可复用本模板，快速实现统一接入。