星河社区aistudio大模型部署非常简单方便，但是每次使用，都需要专门在程序里写入token等，比较麻烦，于是想着做一个AI agent转发程序，这样只要转发程序里写token，客户端可以直接用http免密码使用，非常方便啊！

让Trae帮着写的代码。

AI agent代码：

#!/usr/bin/env python3
"""
星河社区部署模型中继服务
基于星河社区API提供OpenAI兼容的API接口
"""

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json
import os
import time
import asyncio
import aiohttp

app = FastAPI(title="星河社区部署模型中继服务")

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

# 星河社区API配置
XINGHE_API_KEY = "xxxx"
XINGHE_BASE_URL = "https://api-rakbyb46k9xcc5t0.aistudio-app.com/v1"

# 初始化OpenAI客户端
client = OpenAI(
    api_key=XINGHE_API_KEY,
    base_url=XINGHE_BASE_URL
)

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    """
    OpenAI兼容的聊天完成端点
    支持流式和非流式响应
    """
    try:
        # 1. 解析请求数据
        data = await request.json()
        
        # 调试日志控制
        DEBUG = os.environ.get("XINGHE_DEBUG", "false").lower() == "true"
        if DEBUG:
            print(f"==== 收到星河社区请求: {data}")
        
        # 提取消息内容
        messages = data.get("messages", [])
        if not messages:
            raise HTTPException(status_code=400, detail="No messages provided")
        
        # 2. 构建星河社区API调用参数
        stream = data.get("stream", False)
        
        # 构建调用参数
        completion_params = {
            "model": data.get("model", "default"),
            "temperature": data.get("temperature", 0.6),
            "messages": messages,
            "stream": stream
        }
        
        # 可选参数
        if "max_tokens" in data:
            completion_params["max_tokens"] = data["max_tokens"]
        if "top_p" in data:
            completion_params["top_p"] = data["top_p"]
        
        # 3. 调用星河社区API
        if stream:
            # 流式响应处理
            async def generate_stream():
                try:
                    # 使用同步客户端在异步环境中调用
                    completion = client.chat.completions.create(**completion_params)
                    
                    for chunk in completion:
                        if hasattr(chunk.choices[0].delta, "reasoning_content") and chunk.choices[0].delta.reasoning_content:
                            content = chunk.choices[0].delta.reasoning_content
                        else:
                            content = chunk.choices[0].delta.content
                        
                        # 只有当内容不为None时才发送
                        if content is not None:
                            yield f"data: {json.dumps({'choices': [{'delta': {'content': content}}]})}\n\n"
                    
                    # 流结束时发送[DONE]信号
                    yield "data: [DONE]\n\n"
                    
                except Exception as e:
                    yield f"data: {json.dumps({'error': str(e)})}\n\n"
            
            return StreamingResponse(
                generate_stream(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                }
            )
        else:
            # 非流式响应
            completion = client.chat.completions.create(**completion_params)
            
            # 获取响应内容
            content = completion.choices[0].message.content
            
            # 估算token使用量
            estimated_prompt_tokens = len(str(messages)) // 4
            estimated_completion_tokens = len(content) // 4
            
            # 转换为OpenAI兼容格式
            return {
                "id": f"chatcmpl-xinghe-{int(time.time())}",
                "object": "chat.completion",
                "created": int(time.time()),
                "model": data.get("model", "default"),
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": content
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": estimated_prompt_tokens,
                    "completion_tokens": estimated_completion_tokens,
                    "total_tokens": estimated_prompt_tokens + estimated_completion_tokens
                }
            }
        
    except Exception as e:
        error_msg = str(e)
        
        # 处理特定错误类型
        if "429" in error_msg:
            raise HTTPException(
                status_code=429, 
                detail="星河社区API请求频率限制，请稍后重试"
            )
        elif "401" in error_msg:
            raise HTTPException(
                status_code=401,
                detail="星河社区API认证失败，请检查API密钥"
            )
        elif "403" in error_msg:
            raise HTTPException(
                status_code=403,
                detail="星河社区API认证失败，请检查API密钥配置"
            )
        elif "404" in error_msg:
            raise HTTPException(
                status_code=404,
                detail="星河社区API端点不存在"
            )
        else:
            raise HTTPException(status_code=500, detail=f"星河社区API错误: {error_msg}")

@app.get("/")
async def root():
    """根端点"""
    return {
        "message": "星河社区部署模型中继服务",
        "status": "running",
        "endpoint": "/v1/chat/completions",
        "base_url": XINGHE_BASE_URL
    }

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {
        "status": "healthy",
        "service": "星河社区中继",
        "timestamp": int(time.time())
    }

@app.get("/models")
async def list_models():
    """列出可用模型"""
    return {
        "object": "list",
        "data": [
            {
                "id": "default",
                "object": "model",
                "created": 0,
                "owned_by": "星河社区"
            }
        ]
    }

if __name__ == "__main__":
    import uvicorn
    
    # 启动服务器
    print("🚀 启动星河社区部署模型中继服务...")
    print(f"📡 服务地址: http://127.0.0.1:1337")
    print(f"🔗 后端API: {XINGHE_BASE_URL}")
    print("=" * 60)
    
    uvicorn.run(app, host="127.0.0.1", port=1337)

测试代码

#!/usr/bin/env python3
"""
智能测试星河社区中继服务
自动处理频率限制，提供更好的用户体验
"""

import requests
import time
import random

def smart_test():
    """智能测试函数"""
    base_url = "http://127.0.0.1:1337"
    
    print("🧠 智能测试星河社区中继服务")
    print("📌 自动处理频率限制，优化测试体验")
    print("=" * 60)
    
    # 1. 基础服务检查
    print("\n🔍 基础服务检查...")
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code == 200:
            print("✅ 服务运行正常")
        else:
            print(f"❌ 服务异常: HTTP {response.status_code}")
            return False
    except:
        print("❌ 无法连接到服务")
        return False
    
    # 2. 智能API测试（带重试机制）
    print("\n🧪 智能API测试（带重试机制）...")
    
    test_cases = [
        {
            "name": "简短问候测试",
            "messages": [{"role": "user", "content": "你好"}],
            "max_tokens": 5
        },
        {
            "name": "简单问答测试", 
            "messages": [{"role": "user", "content": "什么是AI"}],
            "max_tokens": 20
        }
    ]
    
    success_count = 0
    
    for i, test_case in enumerate(test_cases):
        print(f"\n📋 测试 {i+1}/{len(test_cases)}: {test_case['name']}")
        
        payload = {
            "model": "default",
            "messages": test_case["messages"],
            "max_tokens": test_case["max_tokens"],
            "temperature": 0.7
        }
        
        # 带重试的测试
        max_retries = 2
        for attempt in range(max_retries + 1):
            try:
                if attempt > 0:
                    wait_time = random.uniform(2, 5)  # 随机等待2-5秒
                    print(f"   第{attempt+1}次尝试，等待{wait_time:.1f}秒...")
                    time.sleep(wait_time)
                
                response = requests.post(
                    f"{base_url}/v1/chat/completions",
                    headers={"Content-Type": "application/json"},
                    json=payload,
                    timeout=15
                )
                
                if response.status_code == 200:
                    result = response.json()
                    content = result['choices'][0]['message']['content'].strip()
                    print(f"   ✅ 成功! 回复: '{content}'")
                    success_count += 1
                    break
                    
                elif response.status_code == 429:
                    if attempt < max_retries:
                        print("   ⚠️  遇到频率限制，准备重试...")
                        continue
                    else:
                        print("   ⚠️  频率限制，跳过此测试")
                        break
                        
                else:
                    error_detail = response.json() if response.content else {}
                    print(f"   ❌ 失败: HTTP {response.status_code}")
                    if "detail" in error_detail:
                        print(f"      错误: {error_detail['detail']}")
                    break
                    
            except requests.exceptions.Timeout:
                print("   ❌ 请求超时")
                break
            except Exception as e:
                print(f"   ❌ 异常: {e}")
                break
    
    # 3. 流式响应测试（可选）
    print("\n🌊 流式响应测试（可选）...")
    
    try:
        payload = {
            "model": "default",
            "messages": [{"role": "user", "content": "说'测试成功'"}],
            "max_tokens": 10,
            "stream": True
        }
        
        response = requests.post(
            f"{base_url}/v1/chat/completions",
            headers={"Content-Type": "application/json"},
            json=payload,
            stream=True,
            timeout=10
        )
        
        if response.status_code == 200:
            print("   ✅ 流式连接建立成功")
            
            # 快速检查流式响应
            content_received = False
            for line in response.iter_lines():
                if line:
                    line_str = line.decode('utf-8')
                    if line_str.startswith('data: '):
                        data_str = line_str[6:]
                        if data_str.strip() == '[DONE]':
                            break
                        
                        try:
                            import json
                            data = json.loads(data_str)
                            if 'choices' in data and data['choices']:
                                delta = data['choices'][0].get('delta', {})
                                if 'content' in delta and delta['content']:
                                    content_received = True
                                    break
                        except:
                            continue
            
            if content_received:
                print("   ✅ 流式数据传输正常")
                success_count += 0.5  # 流式测试部分成功
            else:
                print("   ⚠️  流式数据传输异常")
                
        elif response.status_code == 429:
            print("   ⚠️  流式测试遇到频率限制")
        else:
            print(f"   ❌ 流式测试失败: HTTP {response.status_code}")
            
    except Exception as e:
        print(f"   ❌ 流式测试异常: {e}")
    
    # 4. 结果统计
    print("\n" + "=" * 60)
    print("📊 测试结果统计:")
    total_tests = len(test_cases) + 0.5  # 流式测试算0.5分
    success_rate = (success_count / total_tests) * 100
    
    print(f"   总测试项: {total_tests:.1f}")
    print(f"   成功项: {success_count:.1f}")
    print(f"   成功率: {success_rate:.1f}%")
    
    if success_rate >= 70:
        print("\n🎉 测试结果: 优秀 - 服务运行良好!")
        print("💡 您可以正常使用星河社区中继服务")
    elif success_rate >= 40:
        print("\n✅ 测试结果: 良好 - 服务基本可用")
        print("💡 遇到频率限制是正常现象，服务功能正常")
    else:
        print("\n⚠️  测试结果: 需要关注")
        print("💡 建议检查网络连接或稍后重试")
    
    print("\n🔧 使用建议:")
    print("   • 如遇429错误，请等待1-5分钟后重试")
    print("   • 这是星河社区API的正常频率限制")
    print("   • 服务本身运行正常，功能完整")
    
    return success_rate >= 40  # 40%以上认为测试通过

if __name__ == "__main__":
    try:
        smart_test()
    except KeyboardInterrupt:
        print("\n\n⏹️  测试被用户中断")
    except Exception as e:
        print(f"\n❌ 测试过程中出现异常: {e}")

客户端调用示例：

from openai import OpenAI

client = OpenAI(
    api_key="no-auth-required",
    base_url="http://127.0.0.1:1337/v1"
)

response = client.chat.completions.create(
    model="default",
    messages=[{"role": "user", "content": "你好"}]
)