102 lines
4.8 KiB
Python
102 lines
4.8 KiB
Python
import os
|
||
from typing import List
|
||
from langchain_core.messages import HumanMessage, SystemMessage
|
||
from langchain_openai import ChatOpenAI
|
||
# Для локальной разработки без OpenAI:
|
||
# from langchain_ollama import ChatOllama
|
||
|
||
from models import ApiEvent, HttpOkEvent, HttpErrorEvent, ApiEventAdapter
|
||
|
||
# Системный промпт для извлечения структурированных событий
|
||
SYSTEM_PROMPT = """Ты — парсер логов HTTP API. Из входного текста извлеки структурированные события.
|
||
|
||
Правила:
|
||
1. Каждая строка лога — отдельное событие
|
||
2. Если статус 2xx (200, 201 и т.д.) — создавай HttpOkEvent (kind='ok')
|
||
3. Если статус 4xx или 5xx — создавай HttpErrorEvent (kind='error')
|
||
4. Всегда заполняй все поля: kind, status, path, duration_ms (для ok) или error_message (для error)
|
||
5. Если длительности нет в логе, укажи 0.0
|
||
6. Если сообщения об ошибке нет, составь стандартное: "HTTP {status} error"
|
||
|
||
Входной формат лога может быть разным, но старайся извлечь:
|
||
- status: число 200, 404, 500 и т.д.
|
||
- path: строка типа /api/users, /products/123
|
||
- duration_ms: число
|
||
- error_message: текст ошибки (если есть)
|
||
"""
|
||
|
||
class LogEventParser:
|
||
def __init__(self, model_name: str = "gpt-3.5-turbo", temperature: float = 0.0, openai_api_key: str = None):
|
||
"""Инициализация парсера с LangChain structured output"""
|
||
# Настройка OpenAI (можно заменить на Ollama)
|
||
kwargs = {
|
||
"model": model_name,
|
||
"temperature": temperature,
|
||
}
|
||
if openai_api_key:
|
||
kwargs["api_key"] = openai_api_key
|
||
elif os.getenv("OPENAI_API_KEY"):
|
||
kwargs["api_key"] = os.getenv("OPENAI_API_KEY")
|
||
|
||
self.llm = ChatOpenAI(**kwargs)
|
||
|
||
# Привязываем структуру вывода напрямую к LLM (LangChain >=0.2)
|
||
self.structured_llm = self.llm.with_structured_output(ApiEvent, method="function_calling")
|
||
|
||
def parse_single_event(self, log_line: str) -> ApiEvent:
|
||
"""Парсит одну строку лога в ApiEvent"""
|
||
messages = [
|
||
SystemMessage(content=SYSTEM_PROMPT),
|
||
HumanMessage(content=f"Извлеки событие из строки лога:\n{log_line}")
|
||
]
|
||
|
||
# LangChain автоматически вернёт нужный объект благодаря discriminated union
|
||
event = self.structured_llm.invoke(messages)
|
||
return event
|
||
|
||
def parse_multiple_events(self, log_text: str, separator: str = "\n") -> List[ApiEvent]:
|
||
"""Парсит несколько событий из текста (по строкам или разделителю)"""
|
||
events = []
|
||
|
||
# Разбиваем на блоки
|
||
blocks = log_text.strip().split(separator)
|
||
|
||
for block in blocks:
|
||
if not block.strip():
|
||
continue
|
||
try:
|
||
event = self.parse_single_event(block)
|
||
events.append(event)
|
||
except Exception as e:
|
||
print(f"⚠️ Ошибка парсинга строки '{block[:50]}...': {e}")
|
||
continue
|
||
|
||
return events
|
||
|
||
# Альтернативный парсер через PydanticOutputParser (без function calling)
|
||
from langchain_core.output_parsers import PydanticOutputParser
|
||
from langchain_core.prompts import PromptTemplate
|
||
|
||
class LogEventParserWithParser:
|
||
"""Версия с PydanticOutputParser (работает без function calling)"""
|
||
|
||
def __init__(self, model_name: str = "gpt-3.5-turbo", temperature: float = 0.0):
|
||
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
|
||
self.parser = PydanticOutputParser(pydantic_object=ApiEventAdapter) # Используем адаптер
|
||
|
||
self.prompt = PromptTemplate(
|
||
template="""Извлеки событие из строки лога.
|
||
{format_instructions}
|
||
|
||
Лог: {log_line}
|
||
|
||
Верни только JSON с полем kind и соответствующими полями для события.""",
|
||
input_variables=["log_line"],
|
||
partial_variables={"format_instructions": self.parser.get_format_instructions()}
|
||
)
|
||
|
||
self.chain = self.prompt | self.llm | self.parser
|
||
|
||
def parse_single_event(self, log_line: str) -> ApiEvent:
|
||
result = self.chain.invoke({"log_line": log_line})
|
||
return result |