86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""CLI tool to parse raw log lines into typed ApiEvent objects using LangChain structured output.
|
||
|
||
Usage:
|
||
python parse_log.py [--log FILE]
|
||
|
||
If --log is omitted, a built‑in example log is used.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Iterable, List
|
||
|
||
from langchain_core.output_parsers import PydanticOutputParser
|
||
from langchain_core.prompts import ChatPromptTemplate
|
||
from langchain_openai import ChatOpenAI
|
||
|
||
from models import ApiEvent
|
||
|
||
# Example log with mixed OK and error events
|
||
EXAMPLE_LOG = """
|
||
GET /api/users 200 123ms
|
||
POST /api/login 404 Not Found
|
||
GET /api/data 200 456ms
|
||
GET /api/health 500 Internal Server Error
|
||
"""
|
||
|
||
# Prompt template that instructs LLM to output a single ApiEvent in JSON
|
||
PROMPT = ChatPromptTemplate.from_messages(
|
||
[
|
||
("system", "You are a helpful assistant that parses a single log line into a structured event.")
|
||
]
|
||
)
|
||
|
||
# Create a parser for the ApiEvent union
|
||
parser = PydanticOutputParser(pydantic_object=ApiEvent)
|
||
|
||
# LLM model – replace with your own key or use a local model
|
||
llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")
|
||
|
||
|
||
def parse_line(line: str) -> ApiEvent:
|
||
"""Parse a single log line using the LLM and the structured output parser."""
|
||
# Build a prompt that includes the line and asks for JSON
|
||
prompt = PROMPT.format_messages(user=line)
|
||
# Get raw response
|
||
raw = llm.invoke(prompt)
|
||
# Parse into ApiEvent
|
||
return parser.parse(raw.content)
|
||
|
||
|
||
def parse_log(lines: Iterable[str]) -> List[ApiEvent]:
|
||
return [parse_line(line) for line in lines if line.strip()]
|
||
|
||
|
||
def main() -> None:
|
||
parser_cli = argparse.ArgumentParser(description="Parse raw log into typed events")
|
||
parser_cli.add_argument("--log", type=Path, help="Path to a log file; if omitted, example log is used")
|
||
args = parser_cli.parse_args()
|
||
|
||
if args.log and args.log.exists():
|
||
raw_lines = args.log.read_text().splitlines()
|
||
else:
|
||
raw_lines = EXAMPLE_LOG.strip().splitlines()
|
||
|
||
events = parse_log(raw_lines)
|
||
|
||
# Print each event as JSON
|
||
for ev in events:
|
||
print(ev.model_dump_json(indent=2))
|
||
|
||
# Simple table
|
||
print("\nParsed events table:\n")
|
||
print("{:<6} {:<20} {:<6} {}".format("kind", "path", "status", "details"))
|
||
for ev in events:
|
||
if ev.kind == "ok":
|
||
print(f"{ev.kind:<6} {ev.path:<20} {ev.status:<6} duration={ev.duration_ms}ms")
|
||
else:
|
||
print(f"{ev.kind:<6} {ev.path:<20} {ev.status:<6} error={ev.error_message}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|