Руководство пользователя

python examples/demo_simple.py

> Что делает CommitTransaction?
> Найди методы, отвечающие за выделение памяти
> Покажи цепочку вызовов от исполнителя к хранилищу

from src.workflow import MultiScenarioCopilot

copilot = MultiScenarioCopilot()
question = "Какие методы обрабатывают фиксацию транзакций?"
result = copilot.run(question)

print(f"Ответ: {result['answer']}")
print(f"Достоверность: {result.get('confidence', 'N/A')}")
print(f"Интент: {result.get('intent')}")

result = copilot.run(
    query="Найти уязвимости SQL-инъекций",
    context={"scenario_id": "scenario_2"},  # Принудительный выбор сценария
    language="ru",                           # Язык ответа (en/ru)
)

Найди метод 'heap_insert'
Где определён AbortTransaction?
Покажи мне функцию RelationGetBufferForTuple

Какие методы вызывают LWLockAcquire?
Найди вызывающие функции MemoryContextCreate
Что вызывает heap_insert?

Как PostgreSQL обрабатывает MVCC?
Объясни процесс фиксации транзакций
Какой механизм обеспечивает долговечность?

Найди потенциальные точки SQL-инъекций
Покажи пути с необработанным пользовательским вводом
Найди риски переполнения буфера

{
    # Основной вывод
    "answer": "CommitTransaction завершает транзакцию путём...",
    "confidence": 0.85,                # Уверенность классификации интента (0.0–1.0)
    "intent": "security",              # Определённый интент (security, performance и т.д.)
    "scenario_id": "scenario_2",       # ID выполненного сценария

    # Подтверждающие данные
    "evidence": ["xact.c:1234 — CommitTransaction вызывает..."],
    "cpg_results": [...],              # Результаты CPG-запросов
    "metadata": {...},                 # Метаданные сценария

    # Классификация
    "classification_method": "keyword", # "keyword" или "llm"

    # Обработка ошибок
    "error": None,                     # Сообщение об ошибке (если есть)
}

from src.agents.retriever_agent import RetrieverAgent
from src.retrieval.vector_store import VectorStoreReal
from src.agents.analyzer_agent import AnalyzerAgent
from src.services.cpg import CPGQueryService

# Инициализация зависимостей
vector_store = VectorStoreReal()
analyzer_agent = AnalyzerAgent()
cpg_service = CPGQueryService()

# Создание ретривера с гибридным режимом
retriever = RetrieverAgent(
    vector_store=vector_store,
    analyzer_agent=analyzer_agent,
    cpg_service=cpg_service,   # Включает гибридный поиск
    enable_hybrid=True,
)

# Запуск гибридного поиска
results = retriever.retrieve_hybrid(
    question="Найти шаблоны выделения памяти",
    mode="hybrid",             # "hybrid", "vector_only" или "graph_only"
    query_type="structural",   # Подсказка: "semantic", "structural", "security"
    top_k=10,
)

from src.project_manager import ProjectManager

pm = ProjectManager()

# Переключение на зарегистрированный проект (активирует БД, коллекции, домен)
pm.switch_project("postgresql")

# Анализ в контексте этого проекта
from src.workflow import MultiScenarioCopilot
copilot = MultiScenarioCopilot()
result = copilot.run("Найти риски переполнения буфера")

# Переключение на другой проект
pm.switch_project("linux_kernel")
result = copilot.run("Найти шаблоны use-after-free")

from src.workflow import MultiScenarioCopilot

copilot = MultiScenarioCopilot()

# Анализ безопасности — тип определяется автоматически
result = copilot.run("Найти уязвимости к SQL-инъекциям")
print(f"Интент: {result.get('intent')}")  # → 'security'

# Анализ производительности
result = copilot.run("Найти функции с высокой цикломатической сложностью")
print(f"Интент: {result.get('intent')}")  # → 'performance'

# Принудительный выбор сценария (например, security = scenario_2)
result = copilot.run(
    "Проанализировать модуль аутентификации",
    context={"scenario_id": "scenario_2"},
)

import asyncio
from src.services.gocpg import GoCPGClient

async def main():
    client = GoCPGClient()

    # Ad-hoc поиск шаблонов (без CPG-базы)
    results = await client.search(pattern="malloc($x)", language="c", max_results=50)

    # Сканирование с CPG-ограничениями по правилам
    results = await client.scan(
        db_path="data/projects/postgres.duckdb",
        rule_id="unchecked-return",
    )
    print(results)

asyncio.run(main())

from src.services.cpg import CPGQueryService

cpg = CPGQueryService()

# Запрос сохранённых находок паттернов
findings = cpg.get_pattern_findings(severity="high")
stats = cpg.get_pattern_statistics()

# .github/workflows/code-analysis.yml
- name: Запустить анализ кода
  run: |
    python -c "
    from src.workflow import MultiScenarioCopilot
    copilot = MultiScenarioCopilot()
    result = copilot.run('Найти потенциальные проблемы безопасности')
    if result.get('error'):
        print(f'Ошибка анализа: {result[\"error\"]}')
        exit(1)
    print(result['answer'])
    "

# Запуск демо автоматической проверки изменений
python examples/demo_patch_review.py --db data/projects/myproject.duckdb

# Доступные флаги:
#   --db PATH       Путь к DuckDB CPG-базе данных
#   --no-dod        Отключить функционал Definition of Done
#   --auto-dod      Авто-генерация DoD вместо извлечения из PR body
#   --interactive   Включить интерактивное подтверждение DoD

from src.workflow import MultiScenarioCopilot

copilot = MultiScenarioCopilot()
result = copilot.run("Документируй подсистему транзакций")

# result['answer'] содержит сгенерированную документацию
print(result['answer'])