用 Deepseek-v3.1 在 Trae 中构建 AI 中继服务

在本地开发 AI 应用时，我们常常会遇到这样一个问题：某些工具链或 SDK 只支持 OpenAI 的接口规范，但实际想调用的却是国产大模型平台（如百度飞桨星河社区）提供的服务。由于协议不兼容、认证方式不同，直接集成变得异常困难。

最近我就碰到了类似情况——使用 auto-coder 这类自动化编程助手时，它默认依赖 OpenAI 格式的 API 接口，而星河大模型虽然功能强大，其 API 却并非完全遵循 OpenAI 规范。更麻烦的是，部分客户端（比如早期版本的 moon pilot）压根无法识别非标准响应结构。

于是，我决定搭建一个轻量级 AI 中继服务，作为协议转换的“翻译官”：前端按 OpenAI 标准发请求，后端转发给星河 API，并将返回结果重新封装成兼容格式。整个过程就像搭了一座桥，让原本互不相通的两个系统顺利对话。

最终实现方案基于 Trae 平台 + Deepseek-v3.1 模型辅助编码，仅用几十行 Python 代码就完成了核心逻辑。经过多轮测试，该中继服务不仅能稳定处理普通请求，还完整支持流式输出（SSE），与主流工具生态无缝对接。

更重要的是，这种架构的价值远不止于解决临时兼容性问题。它可以统一接口规范、集中管理密钥、为后续添加缓存、限流、日志审计等企业级能力预留空间。哪怕未来某些工具可以直接连接星河 API，这层抽象依然值得保留。

---

为什么选择 PaddlePaddle 星河生态？

百度飞桨（PaddlePaddle）是国内最早推出的全栈式深度学习平台之一，具备从训练到部署的一整套工具链。其旗下的 AI Studio 星河社区 提供了大量预训练大模型和在线推理服务，特别适合中文场景下的 NLP 任务。

尽管本项目没有直接使用 PaddlePaddle 框架进行模型训练或推理，但我们所对接的星河大模型本身就是基于 PaddlePaddle 构建的。这意味着：

• 对中文语义理解更精准
• 响应速度较快，延迟可控
• 支持动态扩缩容，在线服务稳定性高

换句话说，我们的中继服务其实是在扩展 PaddlePaddle 生态的应用边界——通过标准化接入层，让更多不符合原生协议的客户端也能享受其强大的 AI 能力。

这也反映出当前国产大模型落地的一个现实：技术先进 ≠ 易于集成。许多优秀模型受限于封闭接口或私有协议，难以融入现有开发流程。而中继服务正是打破这一壁垒的关键一环。

---

设计目标：不只是简单的代理

这个中继服务不是简单地做 HTTP 转发，而是要解决几个关键痛点：

目标	实现意义
✅ 兼容 OpenAI 接口格式	让 openai 客户端库无需修改即可使用
✅ 支持 stream=True 流式响应	实现聊天界面逐字输出，提升交互体验
✅ 启用 CORS 跨域访问	允许前端页面直连中继服务
✅ 统一认证管理	客户端无需配置真实 API Key，由中继内部处理
✅ 提供健康检查接口	方便监控和服务发现

这些设计使得开发者可以像调用本地 OpenAI 服务一样使用星河大模型，极大降低了集成成本，也提升了系统的可维护性。

---

核心中继代码实现

以下是使用 Deepseek-v3.1 辅助生成并在 Trae 上运行的核心代码，保存为 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星河大模型中继服务")

# 允许跨域请求（便于前端调用）
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 = os.getenv("XINGHE_API_KEY", "6cac673af748cec3440270xxxx")  # 建议通过环境变量注入

@app.post("/v1/chat/completions")
async def proxy_to_paddle(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")

        # 构建符合星河API要求的消息结构
        formatted_messages = [
            {"role": msg["role"], "content": msg["content"]}
            for msg in messages
        ]

        # 提取关键参数
        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": formatted_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 {
        "service": "PaddlePaddle星河大模型中继服务",
        "status": "running",
        "endpoint": "/v1/chat/completions",
        "documentation": "/docs"
    }

@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)

整个服务基于 FastAPI + httpx 构建，异步非阻塞，性能优异。最关键的部分是 /v1/chat/completions 接口的实现：

1. 接收标准 OpenAI 格式的 JSON 请求；
2. 提取 messages, temperature, stream 等字段并重组为星河 API 所需格式；
3. 使用 httpx.AsyncClient 发起带认证头的异步 POST 请求；
4. 若为流式请求，则通过 StreamingResponse 透传 SSE 数据流；否则直接返回 JSON 结果。

值得一提的是，流式支持非常关键。很多现代 AI 工具（如 VS Code 插件、网页聊天框）都依赖 text/event-stream 实现“打字机效果”。如果中继不能正确透传事件流，用户体验将大打折扣。好在 FastAPI 和 httpx 都提供了原生支持，只需几行代码就能完成转发。

另外，安全方面建议将 XINGHE_API_KEY 存入环境变量，避免硬编码泄露。生产环境中还可以进一步限制 allow_origins 到具体域名，而不是开放 "*"。

---

启动服务与依赖安装

确保已安装以下依赖：

pip install fastapi uvicorn httpx

启动方式有两种：

方式一：使用 Uvicorn 命令行（推荐）

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

其中 --reload 开启热重载，适合开发调试；生产环境应移除。

方式二：直接运行脚本

python main.py

服务启动后，访问 http://localhost:8000/docs 即可查看自动生成的 Swagger UI 文档，方便测试和分享接口说明。

---

使用 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": "cmpl-xxx",
  "object": "chat.completion",
  "created": 1718923456,
  "model": "default",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "我是由星河社区提供的大语言模型..."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 15,
    "completion_tokens": 48,
    "total_tokens": 63
  }
}

流式请求测试

curl 'http://127.0.0.1:8000/v1/chat/completions' \
  --header 'Content-Type: application/json' \
  --data '{
    "model": "default",
    "messages": [
      {
        "role": "user",
        "content": "请用中文写一段关于春天的散文诗"
      }
    ],
    "temperature": 0.8,
    "stream": true
  }'

你会看到一系列 data: {...} 事件逐块返回，可用于前端实时渲染。

---

Python 客户端兼容性测试

为了验证是否真正实现了“无缝切换”，我们可以直接使用官方 openai SDK 来调用中继服务：

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": "解释一下PaddlePaddle的特点"}],
            temperature=0.7,
            stream=False
        )
        print("✅ 成功获取响应：")
        print(resp.choices[0].message.content)
    except Exception as e:
        print(f"❌ 请求失败：{e}")

def test_stream():
    print("\n开始测试流式请求...")
    try:
        stream = client.chat.completions.create(
            model="default",
            messages=[{"role": "user", "content": "讲一个程序员的笑话"}],
            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_stream()

实测结果显示，无论是同步响应还是流式输出，都能正常工作。这意味着你现有的基于 openai 的业务代码，几乎不需要任何改动就可以迁移到星河大模型上。

---

与 auto-coder 的集成实践

auto-coder 是一个流行的自动化代码生成工具，支持自定义模型源。我们来试试让它通过中继服务调用星河模型。

步骤 1：启动 auto-coder

auto-coder.chat

步骤 2：配置中继模型

/models /add_model name=paddle-relay model_name=default base_url=http://127.0.0.1:8000/v1
/models /add paddle-relay any_key
/conf model:paddle-relay

注意：此处 any_key 可填任意字符串，因为真实认证已在中继内部完成。

步骤 3：提交任务

输入指令：

请帮我生成一个Flask REST API，提供用户注册和登录功能。

观察控制台输出。如果能看到清晰的思考路径、分步拆解、代码生成全过程，并最终输出完整的路由和数据库模型代码，那就说明中继服务完全可用。

实测中，整个任务消耗约 2 万 token，平均响应延迟在 800ms 左右，属于可接受范围。更重要的是，整个流程无需修改 auto-coder 源码，仅靠配置即可完成切换。

---

可演进的方向：从中继到智能网关

当前版本的中继服务已经能满足基本需求，但如果要在生产环境长期使用，还有不少优化空间：

🔐 安全增强

• 启用 HTTPS 加密通信
• 添加 JWT 或 API Key 验证机制，防止滥用
• 设置 IP 白名单或速率限制（可用 slowapi 实现）

⚙️ 性能优化

• 引入 Redis 缓存高频问答对，降低重复请求开销
• 实现后端负载均衡，支持多个星河实例或多模型并行

📊 可观测性建设

• 集成 Prometheus + Grafana，监控 QPS、延迟、错误率
• 使用 ELK 或 Loki 记录完整请求日志，便于排查问题

🔄 多模型路由

• 根据 model 字段自动路由到不同后端（ERNIE、Qwen、GLM 等）
• 支持灰度发布、A/B 测试、故障转移等高级策略

当这些能力逐步完善后，这个“中继服务”也就进化成了真正的 AI 网关，成为企业级 AI 平台的核心组件之一。

---

这种基于国产大模型 + 自建接入层的模式，正在成为越来越多企业的选择。一方面，我们可以充分利用国内先进的 AI 技术；另一方面，又能保持与国际主流开发工具的兼容性。

而 PaddlePaddle 作为国产深度学习平台的代表，不仅提供了高质量的模型能力，其开放性和生态成熟度也为这类创新应用提供了坚实支撑。未来，在私有化部署、混合云调度、多模型编排等复杂场景下，类似的中继架构必将发挥更大作用。