Реализация кодирования для разработки самоэволюционирующего механизма навыков с помощью OpenSpace для обучения навыкам, повышения эффективности использования токенов и коллективного интеллекта
При масштабировании работы с LLM основное ограничение заключается не в вычислительных мощностях, а в памяти GPU, поскольку каждый запрос требует кэширования данных на уровне токенов. В традиционных настройках для каждого запроса резервируется большой фиксированный блок памяти, размер которого определяется максимальной длиной последовательности. Это приводит к значительному неиспользованию пространства и ограничивает параллелизм.
Внимание с разбивкой по страницам (Paged Attention) улучшает ситуацию, разбивая кэш KV на более мелкие гибкие блоки, которые выделяются только при необходимости, подобно тому, как работает виртуальная память. Это также позволяет нескольким запросам с одинаковым начальным запросом совместно использовать память и дублировать её только тогда, когда их выходные данные начинают различаться. Такой подход значительно повышает эффективность использования памяти, позволяя значительно увеличить пропускную способность с минимальными накладными расходами.
Импорт зависимостей
«`python
import math
import random
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from collections import defaultdict
random.seed(42)
np.random.seed(42)
«`
Настройка констант
Перед моделированием нам нужно знать, сколько памяти GPU занимает один токен. Это зависит от архитектуры модели. Мы используем конфигурацию в стиле GPT — 32 слоя, 32 головы внимания, 128 измерений на голову, хранящихся в fp16.
«`python
NUM_LAYERS = 32
NUM_HEADS = 32
HEAD_DIM = 128
BYTES_FP16 = 2
PAGE_SIZE = 16 # tokens per page (vLLM default)
MAXSEQLEN = 2048
KVBYTESPERTOKEN = 2 NUMLAYERS NUMHEADS HEADDIM BYTES_FP16
KVMBPERTOKEN = KVBYTESPERTOKEN / 1024 / 1024
«`
Наивный подход к кэшированию KV
Наивный подход прост: когда поступает запрос, выделяется непрерывный блок памяти GPU размером с максимальной длиной последовательности — 2048 токенов. Это происходит потому, что длина ответа неизвестна заранее, поэтому резервируется наихудший случай.
«`python
AVG_RESPONSE = 500 # realistic average tokens generated
preallocatedmb = MAXSEQLEN * KVMBPER_TOKEN
actuallyusedmb = AVGRESPONSE * KVMBPERTOKEN
print(f»\nKV cache per token : {KVBYTESPER_TOKEN:,} bytes»)
print(f»Pre-allocated/request : {preallocatedmb:.2f} MB ({MAXSEQLEN} tokens)»)
print(f»Actually used/request : {actuallyusedmb:.2f} MB ({AVG_RESPONSE} tokens)»)
print(f»Utilisation : {actuallyusedmb / preallocatedmb * 100:.1f}%»)
print(f»Wasted per request : {preallocatedmb — actuallyusedmb:.2f} MB»)
«`
Paged Attention
Два класса вводятся здесь для имитации того, как Paged Attention работает на уровне управления памятью. `PagePool` представляет собой физический пул памяти GPU — плоский массив равных по размеру страниц, каждая из которых содержит 16 токенов. `PagedRequest` представляет собой отдельный запрос на вывод.
«`python
class PagePool:
def init(self, total_pages):
self.free = list(range(total_pages))
self.total = total_pages
self.ref_count = defaultdict(int)
def allocate(self):
if not self.free:
raise MemoryError(«OOM — no free pages»)
pid = self.free.pop(0)
self.ref_count[pid] = 1
return pid
def release(self, pid):
self.ref_count[pid] -= 1
if self.ref_count[pid] <= 0:
self.free.append(pid)
del self.ref_count[pid]
def share(self, pid):
«»»Increment ref count — another request is sharing this page.»»»
self.ref_count[pid] += 1
def cow_copy(self, pid):
«»»CoW: allocate a new page, decrement ref on the old one.»»»
new_pid = self.allocate()
self.release(pid)
return new_pid
@property
def utilisation(self):
return (self.total — len(self.free)) / self.total * 100
class PagedRequest:
def init(self, req_id, pool: PagePool):
self.id = req_id
self.pool = pool
self.block_table = [] # logical index → physical page id
self.tokens = 0
def generate_token(self):
if self.tokens % PAGE_SIZE == 0: # page boundary → allocate new page
self.block_table.append(self.pool.allocate())
self.tokens += 1
def free(self):
for pid in self.block_table:
self.pool.release(pid)
self.block_table.clear()
«`
Copy-on-Write: общие системные подсказки
В производстве почти каждый запрос к развёрнутой LLM содержит одну и ту же системную подсказку — инструкции, определяющие поведение модели. При наивном выделении каждый из этих запросов хранит свою полную копию KV-кэша системной подсказки.
«`python
cowpool = PagePool(totalpages=512)
SYSTEM_TOKENS = 200
systempages = math.ceil(SYSTEMTOKENS / PAGE_SIZE)
sharedpids = [cowpool.allocate() for in range(systempages)]
N = 10
user_tables = []
for i in range(N):
table = list(shared_pids)
for pid in shared_pids:
cow_pool.share(pid) # ref count up — no physical copy
user_tables.append(table)
«`
Утилизация: наивный vs Paged
Для измерения утилизации при каждом подходе определяются две функции. `naive_utilisation` рисует количество токенов из нормального распределения со средним значением 500 и стандартным отклонением 200, обрезанным до [200, 2048].
«`python
def naiveutilisation(n, maxseq=2048, avg=500, std=200):
actual = np.clip(np.random.normal(avg, std, n).astype(int), 200, max_seq)
return actual.sum() / (max_seq n) 100, actual
def pagedutilisation(actualtokens, pagesize=PAGESIZE):
pages = np.ceil(actualtokens / pagesize).astype(int)
return actualtokens.sum() / (pages pagesize).sum() 100
«`
Установка OpenSpace
В этом руководстве мы исследуем OpenSpace, механизм самоэволюционирования навыков, разработанный HKUDS, который делает агентов искусственного интеллекта умнее, экономичнее и способными к обучению на основе каждой выполняемой задачи.
«`python
import subprocess, sys, os
print(» Installing OpenSpace from GitHub (this may take 2-3 minutes)…»)
subprocess.check_call([
sys.executable, «-m», «pip», «install», «-q»,
«git+https://github.com/HKUDS/OpenSpace.git»
])
print(«\n Installation complete!»)
«`
Запуск холодного запуска
Мы начинаем с установки OpenSpace непосредственно из его репозитория GitHub вместе с OpenAI SDK, обеспечивая автоматическое извлечение всех зависимостей.
«`python
import os
import json
import shutil
import sqlite3
import glob
import asyncio
import time
from pathlib import Path
WORKSPACE = Path(«/content/openspace_tutorial»)
SKILLS_DIR = WORKSPACE / «skills»
OUTPUT_DIR = WORKSPACE / «outputs»
DB_DIR = WORKSPACE / «.openspace»
if WORKSPACE.exists():
shutil.rmtree(WORKSPACE)
WORKSPACE.mkdir(parents=True)
SKILLS_DIR.mkdir(parents=True)
OUTPUT_DIR.mkdir(parents=True)
DB_DIR.mkdir(parents=True)
os.environ[«OPENSPACE_WORKSPACE»] = str(WORKSPACE)
os.environ[«OPENSPACEHOSTSKILLDIRS»] = str(SKILLSDIR)
envcontent = f»»»OPENAIAPIKEY={os.environ[‘OPENAIAPI_KEY’]}
OPENSPACEMODEL={MODELNAME}
OPENSPACE_WORKSPACE={WORKSPACE}
«»»
env_path = WORKSPACE / «.env»
envpath.writetext(env_content)
print(f» Workspace: {WORKSPACE}»)
print(f» Skills: {SKILLS_DIR}»)
print(f» Outputs: {OUTPUT_DIR}»)
print(f» Database: {DB_DIR}»)
print(«\n Workspace ready for cold start execution!»)
«`
Запуск тёплого запуска
Затем мы выполняем второй запрос, намеренно похожий на запрос холодного запуска, позволяя OpenSpace обнаружить и повторно использовать ранее эволюционировавшие навыки.
«`python
async def runwarmstart_task():
print(«=»*60)
print(» WARM START: Reusing previously evolved skills»)
print(«=»*60)
task = (
«Create a Python script that analyzes a CSV file containing «
«inventory data with columns: date, item, quantity, cost. «
«The script should compute monthly expenditures, identify the top «
«5 most purchased items, and output a formatted summary report.»
)
print(f»\n Task: {task[:100]}…»)
print(» (Similar to cold start task — skills should be reused)\n»)
start_time = time.time()
try:
from openspace import OpenSpace
async with OpenSpace() as cs:
result = await cs.execute(task)
elapsed = time.time() — start_time
print(f»\n Execution time: {elapsed:.1f}s»)
response_text = result.get(«response», str(result))
print(f»\n Response (first 500 chars):»)
print(«-» * 40)
print(response_text[:500])
evolved = result.get(«evolved_skills», [])
reused = result.get(«reused_skills», [])
if reused:
print(f»\n Skills Reused: {len(reused)}»)
for skill in reused:
print(f» • {skill.get(‘name’, ‘unnamed’)}»)
if evolved:
print(f»\n New Skills Evolved: {len(evolved)}»)
for skill in evolved:
print(f» • {skill.get(‘name’, ‘unnamed’)} ({skill.get(‘origin’, »)})»)
return result
except Exception as e:
print(f»\n Execution error: {type(e).name}: {e}»)
print(«We’ll simulate the comparison below.»)
return None
warmstartresult = await runwarmstart_task()
«`
Демонстрация облачного сообщества
Мы исследуем интеграцию облачного сообщества на openspace.cloud, демонстрируя, как агенты ищут, загружают и загружают эволюционировавшие навыки для обмена коллективным интеллектом между командами.
«`python
async def democloudcommunity():
print(«=»*60)
print(» CLOUD COMMUNITY INTEGRATION»)
print(«=»*60)
cloudkey = os.environ.get(«OPENSPACEAPI_KEY», «»)
if not cloud_key:
print(«\n Cloud API key not set. Showing what’s possible:»)
print(«\n With a cloud key, you can:»)
print(» • Search community skills by keyword or task description»)
print(» • Download evolved skills from other agents»)
print(» • Upload your evolved skills to share»)
print(» • Create teams with shared skill repositories»)
print(» • Track skill lineage and evolution history»)
print(«\n Get a free key at: https://open-space.cloud»)
print(«\n CLI commands (outside Colab):»)
print(» $ openspace-download-skill
print(» $ openspace-upload-skill /path/to/skill/dir»)
return
try:
from openspace.cloud.client import CloudClient
client = CloudClient(apikey=cloudkey)
search_queries = [
«data analysis CSV»,
«PDF generation»,
«web scraping»,
]
for query in search_queries:
print(f»\n Searching cloud for: ‘{query}'»)
results = await client.search(query)
if results:
for r in results[:3]:
print(f» {r.get(‘name’, ‘unnamed’)}»)
print(f» Author: {r.get(‘author’, ‘unknown’)}»)
print(f» Downloads: {r.get(‘downloads’, 0)}»)
print(f» Version: {r.get(‘version’, ‘1.0’)}»)
else:
print(» (no results)»)
except ImportError:
print(«\n Cloud client not available in this installation.»)
print(» Install the full package for cloud features.»)
except Exception as e:
print(f»\n Cloud error: {e}»)
await democloudcommunity()
«`
Анализ эволюции с помощью OpenAI
Мы используем OpenAI API для выполнения анализа эволюции нашей библиотеки навыков, выявляя пробелы в охвате, потенциальные пути эволюции и рекомендуемые новые навыки для создания.
«`python
def analyzeevolutionwith_openai():
print(«=»*60)
print(» AI-POWERED EVOLUTION ANALYSIS»)
print(«=»*60)
skill_contents = {}
for skilldir in SKILLSDIR.iterdir():
if skilldir.isdir():
skillmd = skilldir / «SKILL.md»
if skill_md.exists():
skillcontents[skilldir.name] = skillmd.readtext()
if not skill_contents:
print(«\n No skills to analyze yet.»)
return
skills_summary = «\n\n».join([
f»### Skill: {name}\n{content[:500]}»
for name, content in skill_contents.items()
])
from openai import OpenAI
client = OpenAI(apikey=os.environ[«OPENAIAPI_KEY»])
response = client.chat.completions.create(
model=»gpt-4o-mini»,
messages=[
{
«role»: «system»,
«content»: (
«You are a skill evolution analyst for OpenSpace. «
«Analyze the given skills and provide insights on: «
«1) Skill coverage gaps, 2) Potential evolution paths, «
«3) Skill interaction opportunities, 4) Recommended «
«new skills to create. Be concise and actionable.»
)
},
{
«role»: «user»,
«content»: f»Analyze these OpenSpace skills:\n\n{skills_summary}»
}
],
max_tokens=800
)
analysis = response.choices[0].message.content
print(f»\n{analysis}»)
usage = response.usage
print(f»\n Analysis token cost:»)
print(f» Input: {usage.prompt_tokens} tokens»)
print(f» Output: {usage.completion_tokens} tokens»)
print(f» Total: {usage.total_tokens} tokens»)
analyzeevolutionwith_openai()
«`
Демонстрация экономии токенов
Мы запускаем трёхзадачный конвейер последовательно: анализатор CSV, генератор текстовых отчётов и средство проверки качества данных, отслеживая, как навыки накапливаются и увеличивается повторное использование с каждой последующей задачей.
«`python
def demonstratetokensavings():
print(«=»*60)
print(» TOKEN SAVINGS DEMONSTRATION»)
print(«=»*60)
from openai import OpenAI
client = OpenAI(apikey=os.environ[«OPENAIAPI_KEY»])
task = (
«computes monthly revenue, and generates a text report.»
)
cold_messages = [
{«role»: «system», «content»: «You are a coding assistant. Write complete, working Python code.»},
{«role»: «user», «content»: task}
]
cold_response = client.chat.completions.create(
model=»gpt-4o-mini», messages=coldmessages, maxtokens=1500
)
coldtokens = coldresponse.usage.total_tokens
skill_context = «»»
Available Skill: csv-data-analysis (v3, evolved from 28 executions)
Pattern: Use pandas read_csv with encoding detection. Group by
month using pd.Grouper(key=’date’, freq=’ME’). Sum revenue column.
Use tabulate for text report formatting. Fallback: plain text with
f-strings if tabulate unavailable.
Template:
«`python
import pandas as pd
df = pd.read_csv(filepath)
df[‘date’] = pd.to_datetime(df[‘date’])
monthly = df.groupby(pd.Grouper(key=’date’, freq=’ME’))[‘revenue’].sum()
«`
«»»
warm_messages = [
{
«role»: «system»,
«content»: (
«You are a coding assistant with access to pre-evolved skills. «
«Reuse the provided skill patterns to write efficient code. «
«Only add what’s missing — don’t re-derive what the skill provides.\n\n»
f»{skill_context}»
)
},
{«role»: «user», «content»: task}
]
warm_response = client.chat.completions.create(
model=»gpt-4o-mini», messages=warmmessages, maxtokens=1000
)
warmtokens = warmresponse.usage.total_tokens
savingspct = ((coldtokens — warmtokens) / coldtokens) * 100
print(f»\n Cold Start (no skills): {cold_tokens:>6} tokens»)
print(f» Warm Start (with skill): {warm_tokens:>6} tokens»)
print(f» Savings: {coldtokens — warmtokens:>6} tokens ({savings_pct:.1f}%)»)
print(f»\n Cold response length: {len(cold_response.choices[0].message.content)} chars»)
print(f» Warm response length: {len(warm_response.choices[0].message.content)} chars»)
print(f»\n In OpenSpace’s GDPVal benchmark, skill reuse achieved»)
print(f» an average 45.9% token reduction across 50 professional tasks.»)
print(f» The warm start also produces higher quality output because»)
print(f» skills encode battle-tested patterns from real executions.»)
demonstratetokensavings()
«`
В заключение мы увидели, как OpenSpace преобразует работу агентов искусственного интеллекта, превращая их из инструментов без состояния в системы, которые самосовершенствуются с каждой задачей.
1. Какие проблемы решает подход Paged Attention при работе с большими языковыми моделями (LLMs)?
Ответ: подход Paged Attention решает проблему неэффективного использования памяти при работе с LLM. В традиционных настройках для каждого запроса резервируется большой фиксированный блок памяти, что приводит к значительному неиспользованию пространства и ограничивает параллелизм. Paged Attention разбивает кэш KV на более мелкие гибкие блоки, которые выделяются только при необходимости, что позволяет значительно увеличить пропускную способность с минимальными накладными расходами.
2. Какие классы вводятся для имитации работы Paged Attention на уровне управления памятью?
Ответ: для имитации работы Paged Attention на уровне управления памятью вводятся два класса: `PagePool` и `PagedRequest`. `PagePool` представляет собой физический пул памяти GPU — плоский массив равных по размеру страниц, каждая из которых содержит 16 токенов. `PagedRequest` представляет собой отдельный запрос на вывод.
3. Как работает Copy-on-Write в контексте использования системной подсказки в LLM?
Ответ: в контексте использования системной подсказки в LLM Copy-on-Write позволяет избежать дублирования данных. При наивном выделении каждый запрос к развёрнутой LLM содержит одну и ту же системную подсказку, и при каждом запросе хранится полная копия KV-кэша системной подсказки. Однако при использовании Copy-on-Write, если несколько запросов используют одну и ту же системную подсказку, они могут совместно использовать память, что позволяет сэкономить ресурсы.
4. Какие функции используются для измерения утилизации при наивном и Paged подходах?
Ответ: для измерения утилизации при наивном подходе используется функция `naiveutilisation`, которая рисует количество токенов из нормального распределения со средним значением 500 и стандартным отклонением 200, обрезанным до [200, 2048]. Для измерения утилизации при Paged подходе используется функция `pagedutilisation`, которая рассчитывает утилизацию на основе количества фактически использованных токенов и размера страниц.
5. Какие преимущества предоставляет OpenSpace при работе с LLM?
Ответ: OpenSpace предоставляет ряд преимуществ при работе с LLM, включая экономию токенов и повышение эффективности использования навыков. OpenSpace позволяет агентам искусственного интеллекта самосовершенствоваться с каждой задачей, превращая их из инструментов без состояния в системы, которые эволюционируют и адаптируются к новым задачам. Это приводит к снижению затрат на токены и повышению качества вывода.