diff --git a/parse_log.py b/parse_log.py new file mode 100644 index 0000000..7eff89d --- /dev/null +++ b/parse_log.py @@ -0,0 +1,85 @@ +"""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()