79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
"""
|
||
Генерация плана действий для сложных запросов.
|
||
Модель составляет пошаговый план, который затем может быть выполнен.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from typing import List, Dict, Any
|
||
|
||
from rag.services.giga_client import GigaClient
|
||
from rag.config_models import AppConfig
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def generate_plan(
|
||
query: str,
|
||
history: List[Dict[str, str]],
|
||
config: AppConfig,
|
||
giga: GigaClient,
|
||
prompts: Dict[str, str],
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
Генерирует план действий для сложного запроса.
|
||
|
||
Аргументы:
|
||
query: запрос пользователя.
|
||
history: история диалога (может использоваться для контекста).
|
||
config: конфигурация.
|
||
giga: клиент GigaChat.
|
||
prompts: словарь с промптами (ключ 'planning').
|
||
|
||
Возвращает:
|
||
List[Dict]: список шагов, каждый с полями:
|
||
step_description: str,
|
||
required_data: str (опционально),
|
||
expected_output: str (опционально).
|
||
В случае ошибки возвращает пустой список или базовый план.
|
||
"""
|
||
prompt_template = prompts.get("planning")
|
||
if not prompt_template:
|
||
logger.error("Промпт для генерации плана не найден")
|
||
return []
|
||
|
||
# Формируем полный промпт
|
||
# Можно использовать историю для контекста, но для простоты берём только последнее сообщение
|
||
context = ""
|
||
if history:
|
||
# Берём последние 3 сообщения для контекста
|
||
context = "\n".join([f"{msg['role']}: {msg['content']}" for msg in history[-3:]])
|
||
full_prompt = prompt_template.format(query=query, context=context)
|
||
|
||
try:
|
||
response = await giga.chat(
|
||
history=[],
|
||
query=full_prompt,
|
||
system_prompt=None,
|
||
file_id=None,
|
||
temperature=getattr(config, 'planning_temperature', 0.2),
|
||
)
|
||
# Ожидаем JSON-список
|
||
data = json.loads(response.strip())
|
||
if isinstance(data, list):
|
||
# Проверяем, что каждый элемент содержит step_description
|
||
for item in data:
|
||
if "step_description" not in item:
|
||
logger.warning("План содержит шаг без step_description, игнорируем")
|
||
# Можно добавить базовые поля
|
||
item["step_description"] = item.get("description", "Неизвестный шаг")
|
||
return data
|
||
else:
|
||
logger.error("План не является списком JSON")
|
||
return []
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"Ошибка парсинга JSON плана: {e}, ответ: {response[:200]}")
|
||
return []
|
||
except Exception as e:
|
||
logger.exception(f"Ошибка при генерации плана: {e}")
|
||
return [] |