你是一个专业的客服助手，你的任务是帮助用户解决他们遇到的各种问题。
你需要保持礼貌和专业，如果遇到你无法解决的问题，应该转接给人工客服。
请注意保护用户隐私，不要询问不必要的个人信息。
在回答问题时，请确保信息准确，如果不确定，请明确告知用户。
回答应当简洁明了，避免冗长。

角色：客服助手。原则：礼貌专业、信息准确。不确定时告知用户。
隐私：不收集非必要信息。超出能力范围时转接人工。

def trim_history_by_token(messages: list, max_tokens: int = 4000, encoding_name: str = "cl100k_base") -> list:
    import tiktoken
    enc = tiktoken.get_encoding(encoding_name)
    
    system_msg = messages[0]
    system_tokens = len(enc.encode(system_msg["content"]))
    kept = [system_msg]
    total_tokens = system_tokens
    
    for msg in reversed(messages[1:]):
        msg_tokens = len(enc.encode(msg["content"]))
        if total_tokens + msg_tokens > max_tokens:
            break
        kept.insert(1, msg)
        total_tokens += msg_tokens
    
    return kept

def build_prompt(system_parts: list[str], user_input: str) -> list[dict]:
    stable_prefix = "\n".join(system_parts[:2])
    dynamic_instructions = "\n".join(system_parts[2:])
    return [
        {"role": "system", "content": f"{stable_prefix}\n{dynamic_instructions}"},
        {"role": "user", "content": user_input}
    ]

import hashlib
import json

def compute_cache_key(messages: list, model: str, temperature: float, max_tokens: int) -> str:
    content = json.dumps({
        "messages": messages,
        "model": model,
        "temperature": temperature,
        "max_tokens": max_tokens
    }, sort_keys=True)
    return hashlib.sha256(content.encode()).hexdigest()

请求 → [精确缓存层（Redis Hash）] 
         ↓ 未命中
       [语义缓存层（Redis + Embedding）] 
         ↓ 未命中
       [LLM API 调用]
         ↓
       写入两层缓存

import openai

def stream_chat(messages: list, model: str = "gpt-4o"):
    response = openai.OpenAI().chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        max_tokens=1024
    )
    
    full_response = []
    for chunk in response:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_response.append(token)
            yield token
    
    return "".join(full_response)

import asyncio
import openai

client = openai.AsyncOpenAI()

async def call_llm(task_name: str, prompt: str, model: str = "gpt-4o-mini") -> dict:
    response = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    return {"task": task_name, "result": response.choices[0].message.content}

async def parallel_multi_task(tasks: dict[str, str]) -> list[dict]:
    coroutines = [call_llm(name, prompt) for name, prompt in tasks.items()]
    return await asyncio.gather(*coroutines)

tasks = {
    "sentiment": "分析以下评论的情感倾向：'这家餐厅太好吃了'",
    "summary": "用一句话总结：用户在餐厅享用了美味的食物",
    "keywords": "提取以下文本的关键词：'这家餐厅太好吃了'"
}

results = asyncio.run(parallel_multi_task(tasks))

def select_model(query: str, query_complexity: str = "auto") -> str:
    if query_complexity == "auto":
        if len(query) < 50 and any(kw in query for kw in ["是", "吗", "多少", "哪个"]):
            query_complexity = "simple"
        elif len(query) > 500 or "分析" in query or "比较" in query:
            query_complexity = "complex"
        else:
            query_complexity = "medium"
    
    model_map = {
        "simple": "gpt-4o-mini",
        "medium": "gpt-4o",
        "complex": "claude-3-5-sonnet"
    }
    return model_map.get(query_complexity, "gpt-4o")

async def batch_classify(texts: list[str], batch_size: int = 20) -> list[str]:
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        numbered_texts = "\n".join(f"{j+1}. {t}" for j, t in enumerate(batch))
        
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"对以下评论标注情感（正面/负面/中性），每行返回一个结果：\n{numbered_texts}"
            }],
            max_tokens=200
        )
        results.extend(response.choices[0].message.content.strip().split("\n"))
    
    return results

class RequestMerger:
    def __init__(self, merge_window_ms: int = 100, max_batch_size: int = 10):
        self.merge_window = merge_window_ms / 1000
        self.max_batch_size = max_batch_size
        self.pending: list[dict] = []
    
    async def add_request(self, request: dict) -> asyncio.Future:
        future = asyncio.get_event_loop().create_future()
        self.pending.append({"request": request, "future": future})
        
        if len(self.pending) >= self.max_batch_size:
            await self._flush()
        
        return await future
    
    async def _flush(self):
        if not self.pending:
            return
        batch = self.pending[:self.max_batch_size]
        self.pending = self.pending[self.max_batch_size:]
        
        merged_prompt = self._build_merged_prompt([r["request"] for r in batch])
        response = await client.chat.completions.create(**merged_prompt)
        results = self._parse_batch_response(response)
        
        for item, result in zip(batch, results):
            item["future"].set_result(result)

import asyncio
from collections import deque

class RateLimiter:
    def __init__(self, max_rpm: int = 500, max_tpm: int = 150_000):
        self.max_rpm = max_rpm
        self.max_tpm = max_tpm
        self.semaphore = asyncio.Semaphore(max_rpm)
        self.token_usage: deque = deque()
    
    async def acquire(self, estimated_tokens: int = 500):
        await self.semaphore.acquire()
        
        now = asyncio.get_event_loop().time()
        self.token_usage.append((now, estimated_tokens))
        
        while self.token_usage and self.token_usage[0][0] < now - 60:
            self.token_usage.popleft()
        
        total_tokens = sum(t for _, t in self.token_usage)
        if total_tokens + estimated_tokens > self.max_tpm:
            self.semaphore.release()
            await asyncio.sleep(1)
            return await self.acquire(estimated_tokens)
    
    def release(self):
        self.semaphore.release()

async def compress_history(messages: list[dict], max_summary_tokens: int = 200) -> str:
    conversation_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages[1:]
    )
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"用{max_summary_tokens}字以内总结以下对话的关键信息，保留用户偏好和重要决策：\n{conversation_text}"
        }],
        max_tokens=max_summary_tokens
    )
    return response.choices[0].message.content

async def extract_relevant_summary(document: str, query: str, max_tokens: int = 200) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"从以下文档中提取与问题相关的关键信息，用{max_tokens}字以内回答：\n\n问题：{query}\n文档：{document}"
        }],
        max_tokens=max_tokens,
        temperature=0
    )
    return response.choices[0].message.content

async def hierarchical_query(query: str) -> str:
    classification = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": "判断查询复杂度，只回复: simple, medium, complex"
        }, {
            "role": "user",
            "content": query
        }],
        max_tokens=10,
        temperature=0
    )
    complexity = classification.choices[0].message.content.strip().lower()
    
    model_map = {
        "simple": ("gpt-4o-mini", 0.0001),
        "medium": ("gpt-4o", 0.0025),
        "complex": ("gpt-4o", 0.0050)
    }
    model_name, _ = model_map.get(complexity, ("gpt-4o", 0.0025))
    
    response = await client.chat.completions.create(
        model=model_name,
        messages=[{"role": "user", "content": query}]
    )
    return response.choices[0].message.content

from locust import HttpUser, task, between
import json
import random

class LLMUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        self.test_queries = [
            "什么是深度学习？",
            "请比较 Python 和 Java 的优缺点",
            "帮我写一首关于春天的诗",
            "解释量子计算的基本原理",
            "如何优化 PostgreSQL 查询性能？",
        ]
    
    @task(3)
    def simple_query(self):
        query = random.choice(self.test_queries[:2])
        self._send_llm_request(query)
    
    @task(2)
    def complex_query(self):
        query = random.choice(self.test_queries[2:])
        self._send_llm_request(query)
    
    @task(1)
    def streaming_query(self):
        query = random.choice(self.test_queries)
        self._send_llm_request(query, stream=True)
    
    def _send_llm_request(self, query: str, stream: bool = False):
        payload = {
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": query}],
            "max_tokens": 512,
            "temperature": 0.7,
            "stream": stream
        }
        with self.client.post(
            "/v1/chat/completions",
            json=payload,
            catch_response=True,
            name="/chat/completions"
        ) as response:
            if response.status_code == 429:
                response.failure("Rate limited")
            elif response.status_code != 200:
                response.failure(f"HTTP {response.status_code}")

import json
import hashlib
import numpy as np
import openai
import redis
from dataclasses import dataclass
from typing import Optional

@dataclass
class CacheConfig:
    embedding_model: str = "text-embedding-3-small"
    similarity_threshold: float = 0.92
    max_cache_size: int = 100_000
    ttl_seconds: int = 86400 * 7
    embedding_dim: int = 1536

class SemanticCache:
    def __init__(self, config: CacheConfig = None):
        self.config = config or CacheConfig()
        self.redis = redis.Redis(host="localhost", port=6379, db=0, decode_responses=False)
        self.openai_client = openai.OpenAI()
        self._index_key = "semantic_cache:index"
    
    def _get_embedding(self, text: str) -> list[float]:
        response = self.openai_client.embeddings.create(
            model=self.config.embedding_model,
            input=text
        )
        return response.data[0].embedding
    
    def _compute_exact_key(self, messages: list, model: str, temperature: float, max_tokens: int) -> str:
        content = json.dumps({
            "messages": messages,
            "model": model,
            "temperature": temperature,
            "max_tokens": max_tokens
        }, sort_keys=True, ensure_ascii=False)
        return f"exact:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def _cosine_similarity(self, vec_a: list[float], vec_b: list[float]) -> float:
        a = np.array(vec_a)
        b = np.array(vec_b)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    def get(self, messages: list, model: str = "gpt-4o",
            temperature: float = 0.7, max_tokens: int = 1024) -> Optional[dict]:
        exact_key = self._compute_exact_key(messages, model, temperature, max_tokens)
        cached = self.redis.get(exact_key)
        if cached:
            result = json.loads(cached)
            result["cache_hit"] = "exact"
            return result
        
        query_text = messages[-1]["content"] if messages else ""
        query_embedding = self._get_embedding(query_text)
        
        index_entries = self.redis.zrange(self._index_key, 0, -1, withscores=False)
        best_score = -1
        best_key = None
        
        for entry_id in index_entries:
            entry_data = self.redis.hgetall(f"semantic_cache:entry:{entry_id.decode()}")
            if not entry_data:
                continue
            cached_embedding = json.loads(entry_data[b"embedding"])
            similarity = self._cosine_similarity(query_embedding, cached_embedding)
            if similarity > best_score:
                best_score = similarity
                best_key = entry_id.decode()
        
        if best_score >= self.config.similarity_threshold and best_key:
            entry_data = self.redis.hgetall(f"semantic_cache:entry:{best_key}")
            result = json.loads(entry_data[b"response"])
            result["cache_hit"] = "semantic"
            result["similarity"] = round(best_score, 4)
            return result
        
        return None
    
    def set(self, messages: list, response: dict, model: str = "gpt-4o",
            temperature: float = 0.7, max_tokens: int = 1024):
        exact_key = self._compute_exact_key(messages, model, temperature, max_tokens)
        self.redis.setex(exact_key, self.config.ttl_seconds, json.dumps(response, ensure_ascii=False).encode())
        
        query_text = messages[-1]["content"] if messages else ""
        embedding = self._get_embedding(query_text)
        
        entry_id = hashlib.md5(f"{query_text}:{model}".encode()).hexdigest()
        pipe = self.redis.pipeline()
        pipe.hset(f"semantic_cache:entry:{entry_id}", mapping={
            "embedding": json.dumps(embedding),
            "response": json.dumps(response, ensure_ascii=False),
            "query": query_text,
            "model": model
        })
        pipe.expire(f"semantic_cache:entry:{entry_id}", self.config.ttl_seconds)
        pipe.zadd(self._index_key, {entry_id.encode(): 1.0})
        pipe.execute()
        
        cache_size = self.redis.zcard(self._index_key)
        if cache_size > self.config.max_cache_size:
            self.redis.zremrangebyrank(self._index_key, 0, cache_size - self.config.max_cache_size - 1)
    
    def stats(self) -> dict:
        exact_keys = len([k for k in self.redis.scan_iter("exact:*")])
        semantic_entries = self.redis.zcard(self._index_key)
        return {
            "exact_cache_entries": exact_keys,
            "semantic_cache_entries": semantic_entries
        }


async def cached_llm_call(
    messages: list,
    model: str = "gpt-4o",
    temperature: float = 0.7,
    max_tokens: int = 1024
) -> dict:
    cache = SemanticCache()
    
    cached = cache.get(messages, model, temperature, max_tokens)
    if cached:
        return cached
    
    client = openai.AsyncOpenAI()
    response = await client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        max_tokens=max_tokens
    )
    
    result = {
        "content": response.choices[0].message.content,
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens
        },
        "cache_hit": "miss"
    }
    
    cache.set(messages, result, model, temperature, max_tokens)
    return result