import openai
import time

client = openai.OpenAI(
    api_key="your-huawei-key",
    base_url="https://maas-api.huaweicloud.com/v1"
)

def chat_sync(prompt: str) -> str:
    resp = client.chat.completions.create(
        model="deepseek-v3",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=2048
    )
    return resp.choices[0].message.content

def chat_stream(prompt: str) -> tuple[str, float]:
    chunks = []
    first_token_time = None
    resp = client.chat.completions.create(
        model="deepseek-v3",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=2048
    )
    for chunk in resp:
        if first_token_time is None and chunk.choices[0].delta.content:
            first_token_time = time.time()
        delta = chunk.choices[0].delta.content or ""
        chunks.append(delta)
    return "".join(chunks), first_token_time

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key="your-huawei-key",
    base_url="https://maas-api.huaweicloud.com/v1"
)

async def async_chat(prompt: str) -> str:
    resp = await async_client.chat.completions.create(
        model="deepseek-v3",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1024
    )
    return resp.choices[0].message.content

async def batch_process(prompts: list[str], concurrency: int = 10):
    sem = asyncio.Semaphore(concurrency)

    async def bounded(prompt: str):
        async with sem:
            return await async_chat(prompt)

    tasks = [bounded(p) for p in prompts]
    return await asyncio.gather(*tasks)

# 50 条 prompt，并发 10
results = asyncio.run(batch_process(prompts, concurrency=10))

# ❌ 冗长，实测 328 token
SYSTEM_PROMPT_VERBOSE = """你是一个专业的客服助手。
你的职责是回答用户关于产品的问题。
请使用礼貌且专业的语气。
如果不知道答案，请如实告知用户你在该问题上的知识有限。
回答时请使用中文。
请确保回答准确无误。
每个回答请控制在 200 字以内。
不要使用表情符号。
不要在回答中编造信息。"""

# ✅ 精简，实测 67 token
SYSTEM_PROMPT_CONCISE = (
    "你是中文客服助手。回答务必准确、简洁、200字以内。未知则直接告知。"
)

# 精确控制 output 总量和推理链长度
resp = client.chat.completions.create(
    model="deepseek-r1",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=4096,  # 包含推理链的总上限
    extra_body={
        "reasoning_effort": "low",  # low / medium / high
    }
)

def truncate_history(
    messages: list[dict],
    max_recent_turns: int = 3,
    max_total_tokens: int = 4000
) -> list[dict]:
    # 保留 system prompt
    system = [m for m in messages if m["role"] == "system"]
    # 从最近的对话开始截取
    non_system = [m for m in messages if m["role"] != "system"]
    recent = non_system[-(max_recent_turns * 2):]  # 每轮 user + assistant 两条

    # 如果超出 token 上限，继续剪裁
    total = len("".join(m["content"] for m in system + recent))
    if total > max_total_tokens:
        # 只保留最后 2 轮
        recent = non_system[-(max_recent_turns * 2 - 2):]

    return system + recent

import numpy as np
from typing import Optional
import hashlib

class SemanticCache:
    def __init__(self, threshold: float = 0.92):
        self.threshold = threshold
        self.cache: dict[str, tuple[str, list[float]]] = {}

    def _get_embedding(self, text: str) -> list[float]:
        resp = client.embeddings.create(
            model="text-embedding-v2",
            input=text
        )
        return resp.data[0].embedding

    def get(self, query: str) -> Optional[str]:
        q_emb = np.array(self._get_embedding(query))
        best_score, best_key = 0.0, None
        for key, (_, cached_emb) in self.cache.items():
            score = np.dot(q_emb, cached_emb) / (
                np.linalg.norm(q_emb) * np.linalg.norm(cached_emb)
            )
            if score > best_score:
                best_score, best_key = score, key
        if best_score >= self.threshold:
            return self.cache[best_key][0]
        return None

    def set(self, query: str, answer: str):
        emb = self._get_embedding(query)
        key = hashlib.md5(query.encode()).hexdigest()
        self.cache[key] = (answer, emb)

cache = SemanticCache(threshold=0.92)

def chat_with_cache(prompt: str) -> str:
    cached = cache.get(prompt)
    if cached:
        return cached  # 命中，零延迟
    answer = chat_sync(prompt)
    cache.set(prompt, answer)
    return answer

# 所有请求固定 system prompt + 固定格式
BASE_SYSTEM = "你是智能客服，请简洁回答用户问题。"
# 第一个请求走完整 prefill
# 第二个请求使用相同 system prompt → prefill 时复用前面结果的 KV-Cache

import json
from datetime import datetime

class CostTracker:
    def __init__(self, input_price: float, output_price: float):
        self.input_price = input_price
        self.output_price = output_price
        self.records: list[dict] = []

    def log(self, prompt: str, response: str):
        input_tokens = len(prompt) // 2  # 粗略估计，生产环境用 API 返回的 usage
        output_tokens = len(response) // 2
        cost = (
            input_tokens / 1_000_000 * self.input_price +
            output_tokens / 1_000_000 * self.output_price
        )
        self.records.append({
            "timestamp": datetime.now().isoformat(),
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        })
        return cost

    def summary(self) -> dict:
        if not self.records:
            return {"total_calls": 0, "total_cost": 0}
        total_cost = sum(r["cost"] for r in self.records)
        total_input = sum(r["input_tokens"] for r in self.records)
        total_output = sum(r["output_tokens"] for r in self.records)
        return {
            "total_calls": len(self.records),
            "total_input_tokens": total_input,
            "total_output_tokens": total_output,
            "total_cost": round(total_cost, 4),
            "avg_cost_per_call": round(total_cost / len(self.records), 6),
            "input_rate": round(total_input / (total_input + total_output) * 100, 1),
        }

def choose_model(question: str) -> str:
    """基于问题复杂度自动选择模型"""
    complex_keywords = ["比较", "分析", "推理", "为什么", "证明",
                        "计算", "推导", "对比", "评估"]
    for kw in complex_keywords:
        if kw in question:
            return "deepseek-r1"
    return "deepseek-v3"

用户请求 → 模型选择（V3/R1） → 语义缓存 → 命中 → 直接返回
                                         ↓ 未命中
                                   对话历史截断
                                         ↓
                                    RAG 检索  
                                         ↓
                                    MaaS DeepSeek (流式输出)
                                         ↓
                                    结果写入缓存 → 返回

import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class OptimizedChatConfig:
    model: str = "deepseek-v3"
    system_prompt: str = (
        "你是简洁的中文客服助手。回答务必准确、200字以内。"
    )
    max_tokens: int = 1024
    cache_threshold: float = 0.92
    concurrency: int = 10
    timeout: int = 30
    retry_times: int = 3

class OptimizedChat:
    def __init__(self, config: OptimizedChatConfig):
        self.config = config
        self.client = AsyncOpenAI(
            api_key="your-key",
            base_url="https://maas-api.huaweicloud.com/v1"
        )
        self.sem = asyncio.Semaphore(config.concurrency)
        self.cache = SemanticCache(threshold=config.cache_threshold)
        self.tracker = CostTracker(
            input_price=2.0,
            output_price=8.0
        )

    async def chat(self, user_query: str,
                   context: Optional[list] = None) -> str:
        # Step 1: 查缓存
        cached = self.cache.get(user_query)
        if cached:
            return cached

        # Step 2: 构建请求（精简历史）
        messages = [
            {"role": "system", "content": self.config.system_prompt}
        ]
        if context:
            messages.extend(context[-6:])  # 最近 3 轮
        messages.append({"role": "user", "content": user_query})

        # Step 3: 并发控制 + 重试
        for attempt in range(self.config.retry_times):
            try:
                async with self.sem:
                    resp = await asyncio.wait_for(
                        self.client.chat.completions.create(
                            model=self.config.model,
                            messages=messages,
                            stream=True,
                            max_tokens=self.config.max_tokens
                        ),
                        timeout=self.config.timeout
                    )
                    result = ""
                    async for chunk in resp:
                        result += chunk.choices[0].delta.content or ""

                    # Step 4: 写入缓存 + 成本记录
                    self.cache.set(user_query, result)
                    self.tracker.log(user_query, result)
                    return result

            except (asyncio.TimeoutError, Exception) as e:
                if attempt == self.config.retry_times - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

# 使用
config = OptimizedChatConfig()
chat_system = OptimizedChat(config)

async def main():
    queries = ["怎么退款？", "密码忘了怎么办？", "怎么退款？"] * 10
    tasks = [chat_system.chat(q) for q in queries]
    results = await asyncio.gather(*tasks)
    report = chat_system.tracker.summary()
    print(f"调用次数: {report['total_calls']}")
    print(f"总成本: ${report['total_cost']}")
    print(f"平均每次: ${report['avg_cost_per_call']}")

asyncio.run(main())

import time
import requests

def probe_and_backoff():
    """探测 rate limit 并自适应并发"""
    concurrency = 5
    while concurrency <= 30:
        try:
            test_batch(concurrency=concurrency)
            concurrency += 5  # 没触发限流，逐步增加
        except RateLimitError:
            concurrency = max(1, concurrency - 5)
            time.sleep(60)
    return concurrency

resp = client.chat.completions.create(...)
actual_input = resp.usage.prompt_tokens      # 精确的 input token 数
actual_output = resp.usage.completion_tokens  # 精确的 output token 数