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

python examples/demo_simple.py

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

from src.digital_employees.runtime.scenarios import invoke_role_bound_scenario

scenario_context = {"project_key": "codegraph", "namespace": "codegraph/default/codegraph", "task_id": "TASK-EXAMPLE", "language": "ru"}
question = "Какие методы обрабатывают фиксацию транзакций?"
result = invoke_role_bound_scenario(query=question, context=scenario_context)

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

result = invoke_role_bound_scenario(
    query="Найти уязвимости SQL-инъекций",
    scenario_id="scenario_02",
    context=scenario_context,
)

Найди метод '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.core_agents.retriever_agent import RetrieverAgent
from src.agents.core_agents.analyzer_agent import AnalyzerAgent
from src.services.cpg import CPGQueryService
from src.workflow._components import get_retrieval_adapter

# Инициализация зависимостей
retrieval_adapter = get_retrieval_adapter(collection_prefix="codegraph")
analyzer_agent = AnalyzerAgent()
cpg_service = CPGQueryService(db_path="data/projects/codegraph.duckdb", grpc_only=True)

# Создание ретривера с гибридным режимом
retriever = RetrieverAgent(
    retrieval_adapter,
    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()
project = pm.resolve_project(project_name="postgresql")

from src.digital_employees.runtime.scenarios import invoke_role_bound_scenario
scenario_context = {"project_key": "codegraph", "namespace": "codegraph/default/codegraph", "task_id": "TASK-EXAMPLE", "language": "ru"}
result = invoke_role_bound_scenario(
    query="Найти риски переполнения буфера",
    context={
        **scenario_context,
        "project_key": project.name,
        "db_path": project.db_path,
        "source_path": project.source_path,
        "domain": project.domain,
    },
)

from src.digital_employees.runtime.scenarios import invoke_role_bound_scenario

scenario_context = {"project_key": "codegraph", "namespace": "codegraph/default/codegraph", "task_id": "TASK-EXAMPLE", "language": "ru"}

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

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

# Принудительный выбор сценария (например, security = scenario_02)
result = invoke_role_bound_scenario(
    query="Проанализировать модуль аутентификации",
    scenario_id="scenario_02",
    context=scenario_context,
)

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(db_path="data/projects/postgres.duckdb", grpc_only=True)

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

# .github/workflows/code-analysis.yml
- name: Запустить анализ кода
  run: |
    python -c "
    from src.digital_employees.runtime.scenarios import invoke_role_bound_scenario
    scenario_context = {'project_key': 'codegraph', 'namespace': 'codegraph/default/codegraph', 'task_id': 'TASK-EXAMPLE', 'language': 'ru'}
    result = invoke_role_bound_scenario(query='Найти потенциальные проблемы безопасности', context=scenario_context)
    if result.get('error'):
        print(f'Ошибка анализа: {result[\"error\"]}')
        exit(1)
    print(result['answer'])
    "

# Запуск автоматической проверки staged changes
python -m src.cli review --staged

# Доступные флаги:
#   --db PATH       Путь к DuckDB CPG-базе данных
#   --base-ref REF  Git base ref для diff review
#   --format FORMAT Формат вывода: text, json, markdown или sarif
#   --output-file   Необязательный файл вывода

from src.digital_employees.runtime.scenarios import invoke_role_bound_scenario

scenario_context = {"project_key": "codegraph", "namespace": "codegraph/default/codegraph", "task_id": "TASK-EXAMPLE", "language": "ru"}
result = invoke_role_bound_scenario(query="Документируй подсистему транзакций", context=scenario_context)

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