Войти в систему
Войти с помощью соц. сетей:
python
# Упрощенная логика KV Cache для генерации
class KVCache:
def __init__(self):
self.keys = [] # список K-матриц для каждого токена
self.values = [] # список V-матриц для каждого токена
def forward(self, new_tokens, past_kv):
# new_tokens - только что сгенерированные токены
# past_kv - кэш с предыдущими K и V
if past_kv is None:
# Первый шаг: считаем для всех токенов
keys, values = compute_all(new_tokens)
else:
# Следующие шаги: считаем только для новых токенов
keys, values = compute_only_new(new_tokens, past_kv)
# Сохраняем в кэш
self.keys.append(keys)
self.values.append(values)
# Используем для следующего шага
return attention(self.keys, self.values)
Что здесь происходит:
функция compute_only_new - это и есть магия KV Cache. Она не пересчитывает K и V для старых токенов, а берёт их из кэша, вычисляя только для нового.
python
def calculate_tco(scenario, monthly_requests, avg_input_tokens, avg_output_tokens):
"""
Расчет TCO для трех сценариев
"""
if scenario == "api":
# GPT-4o цены ( условные )
price_input = 5 / 1_000_000 # $5 за 1M токенов
price_output = 15 / 1_000_000 # $15 за 1M токенов
monthly_input_cost = monthly_requests * avg_input_tokens * price_input
monthly_output_cost = monthly_requests * avg_output_tokens * price_output
return monthly_input_cost + monthly_output_cost
elif scenario == "gpu_rent":
# 4 × A100, 16 часов в сутки
gpu_hourly = 2.5 # $2.5 за час
hours_per_day = 16
days_per_month = 30
engineer_cost = 7000 # $7k инженер в месяц
gpu_cost = gpu_hourly * hours_per_day * days_per_month * 4
return gpu_cost + engineer_cost
elif scenario == "on_prem":
# 8 × H100, 3 года амортизации
capex = 350_000 # $350k
amortization = capex / 36 # в месяц
engineer_cost = 15000 # два инженера
electricity = 2000 # электричество и охлаждение
return amortization + engineer_cost + electricity
# Пример использования
print(calculate_tco("api", 1_000_000, 1000, 500)) # ~$20k
print(calculate_tco("gpu_rent", 10_000_000, 1000, 500)) # ~$40k
print(calculate_tco("on_prem", 100_000_000, 1000, 500)) # ~$30k
python
import fasttext
# 1. Подготовка данных (пример)
# Каждая строка: __label__{class} {текст}
train_data = [
"__label__simple Привет как дела",
"__label__simple Какая погода",
"__label__math Реши уравнение x^2 + 5x + 6 = 0",
"__label__math Вычисли производную sin(x)",
"__label__code Напиши функцию на Python",
"__label__code Как отсортировать список"
]
# Сохраняем в файл
with open("router_train.txt", "w") as f:
f.write("\n".join(train_data))
# 2. Обучение классификатора
model = fasttext.train_supervised(
input="router_train.txt",
lr=1.0,
epoch=25,
wordNgrams=2
)
# 3. Роутинг
def route_query(query):
label, prob = model.predict(query, k=1)
category = label[0].replace("__label__", "")
if category == "simple":
return "gpt-3.5-turbo"
elif category == "math":
return "deepseek-math"
elif category == "code":
return "claude-3-sonnet"
else:
return "gpt-4o" # fallback
# Тест
print(route_query("Привет")) # gpt-3.5-turbo
print(route_query("Реши уравнение")) # deepseek-math
print(route_query("Напиши функцию")) # claude-3-sonnet
python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import openai
import redis
import hashlib
app = FastAPI(title="Digital Twin Prototype")
# Подключение к Redis для короткой памяти
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
# Конфигурация OpenAI (или локальной модели)
openai.api_key = "your-key"
class UserQuery(BaseModel):
user_id: str
query: str
session_id: str = None
class TwinResponse(BaseModel):
answer: str
source: str # "cache", "llm", "error"
def get_personality(user_id):
"""Загружает ядро личности из БД"""
# В реальности - запрос к векторной БД
return "Ты - цифровой двойник. Отвечай кратко и по делу."
def get_short_memory(session_id):
"""Загружает контекст последних сообщений"""
key = f"memory:{session_id}"
return redis_client.lrange(key, -10, -1) # последние 10 сообщений
def update_memory(session_id, user_query, twin_answer):
"""Обновляет короткую память"""
key = f"memory:{session_id}"
redis_client.rpush(key, f"User: {user_query}")
redis_client.rpush(key, f"Twin: {twin_answer}")
redis_client.ltrim(key, -20, -1) # храним только 20 последних
@app.post("/ask", response_model=TwinResponse)
async def ask_twin(query: UserQuery):
# 1. Проверка кэша (если вопрос уже был)
cache_key = hashlib.md5(f"{query.user_id}:{query.query}".encode()).hexdigest()
cached = redis_client.get(f"cache:{cache_key}")
if cached:
return TwinResponse(answer=cached, source="cache")
# 2. Сбор контекста
personality = get_personality(query.user_id)
short_memory = get_short_memory(query.session_id) if query.session_id else []
# 3. Формирование промта
prompt = f"{personality}\n\n"
for msg in short_memory:
prompt += f"{msg}\n"
prompt += f"User: {query.query}\nTwin:"
# 4. Вызов LLM
try:
response = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=500
)
answer = response.choices[0].message.content
# 5. Сохраняем в кэш и память
redis_client.setex(f"cache:{cache_key}", 3600, answer) # час
if query.session_id:
update_memory(query.session_id, query.query, answer)
return TwinResponse(answer=answer, source="llm")
except Exception as e:
return TwinResponse(answer=f"Error: {str(e)}", source="error")
# Запуск: uvicorn main:app –reload
Войти с помощью соц. сетей: