API 调用示例

import requests
import json

class AutoNavClient:
    def __init__(self, api_key, base_url="https://your-domain.com/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def chat_completion(self, app_id, messages, model="gpt-3.5-turbo", **kwargs):
        """同步聊天完成"""
        url = f"{self.base_url}/autobox/{app_id}/sync"
        
        payload = {
            "messages": messages,
            "model": model,
            **kwargs
        }
        
        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"错误详情: {e.response.text}")
            raise

# 使用示例
client = AutoNavClient("ak-your32characterapikeyhere")

messages = [
    {"role": "user", "content": "请解释什么是机器学习"}
]

try:
    result = client.chat_completion(
        app_id="your_app_id",
        messages=messages,
        temperature=0.7,
        max_tokens=1000
    )
    
    print("AI回复:", result["choices"][0]["message"]["content"])
    print("Token使用:", result["usage"])
    
except Exception as e:
    print(f"调用失败: {e}")

import requests
import json

def stream_chat_completion(api_key, app_id, messages):
    """处理流式响应"""
    url = f"https://your-domain.com/v1/autobox/{app_id}/sync"
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "messages": messages,
        "model": "gpt-3.5-turbo",
        "stream": True,
        "temperature": 0.7
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload, stream=True)
        response.raise_for_status()
        
        full_content = ""
        for line in response.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    data = line[6:]  # 移除 'data: ' 前缀
                    if data == '[DONE]':
                        break
                    
                    try:
                        chunk = json.loads(data)
                        if chunk['choices'][0]['delta'].get('content'):
                            content = chunk['choices'][0]['delta']['content']
                            full_content += content
                            print(content, end='', flush=True)
                    except json.JSONDecodeError:
                        continue
        
        print()  # 换行
        return full_content
        
    except requests.exceptions.RequestException as e:
        print(f"流式请求失败: {e}")
        raise

# 使用示例
messages = [{"role": "user", "content": "写一个Python快速排序的实现"}]
content = stream_chat_completion("ak-your32characterapikeyhere", "your_app_id", messages)

import time
import requests

class AsyncTaskHandler:
    def __init__(self, api_key, base_url="https://your-domain.com/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def submit_async_task(self, app_id, messages, **kwargs):
        """提交异步任务"""
        url = f"{self.base_url}/autobox/{app_id}/async"
        
        payload = {
            "messages": messages,
            **kwargs
        }
        
        response = requests.post(url, headers=self.headers, json=payload)
        response.raise_for_status()
        return response.json()
    
    def get_task_status(self, task_number):
        """查询任务状态"""
        url = f"{self.base_url}/task-query/{task_number}"
        
        response = requests.get(url, headers=self.headers)
        response.raise_for_status()
        return response.json()
    
    def wait_for_completion(self, task_number, max_wait=300, poll_interval=5):
        """等待任务完成"""
        start_time = time.time()
        
        while time.time() - start_time < max_wait:
            status = self.get_task_status(task_number)
            
            if status["status"] == "已完成":
                return status
            elif status["status"] == "已报错":
                raise Exception(f"任务执行失败: {status.get('comments', '未知错误')}")
            
            print(f"任务状态: {status['status']}, 进度: {status.get('task_progress', 'N/A')}")
            time.sleep(poll_interval)
        
        raise TimeoutError("任务等待超时")

# 使用示例
handler = AsyncTaskHandler("ak-your32characterapikeyhere")

# 提交长时间任务
messages = [{"role": "user", "content": "请详细分析这个复杂的数据集..."}]
task_result = handler.submit_async_task("your_app_id", messages, model="gpt-4")

print(f"任务已提交: {task_result['task_number']}")

# 等待完成
try:
    final_result = handler.wait_for_completion(task_result['task_number'])
    print("任务完成:", final_result['task_result'])
except Exception as e:
    print(f"任务失败: {e}")

try:
    result = client.chat_completion(app_id, messages)
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("API_KEY无效，请检查密钥是否正确")
    elif e.response.status_code == 403:
        print("访问被拒绝，请检查账户状态")

try:
    result = client.chat_completion(app_id, messages)
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 429:
        error_detail = e.response.json().get('detail', '')
        if '配额不足' in error_detail:
            print("配额已用完，请充值后继续使用")
        else:
            print("请求过于频繁，请稍后重试")

try:
    result = client.chat_completion(app_id, messages, temperature=3.0)  # 错误的参数
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 422:
        errors = e.response.json().get('detail', [])
        for error in errors:
            field = error.get('loc', [])[-1]
            message = error.get('msg', '')
            print(f"参数 {field} 错误: {message}")

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 配置重试策略
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

# 使用session进行请求
response = session.post(url, headers=headers, json=payload)

import asyncio
import aiohttp

async def async_chat_completion(session, app_id, messages):
    url = f"https://your-domain.com/v1/autobox/{app_id}/sync"
    
    payload = {
        "messages": messages,
        "model": "gpt-3.5-turbo"
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    async with session.post(url, json=payload, headers=headers) as response:
        return await response.json()

async def batch_requests():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(10):
            messages = [{"role": "user", "content": f"请求 {i}"}]
            task = async_chat_completion(session, "your_app_id", messages)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results

# 运行批量请求
results = asyncio.run(batch_requests())