122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
"""
|
||
# main.py
|
||
# Structured log parsing with LangChain and Pydantic v2
|
||
# No deepagents dependency – uses standard LangChain tooling
|
||
# Author: Student
|
||
# Date: 2026-06-30
|
||
|
||
import os
|
||
import argparse
|
||
import textwrap
|
||
from typing import List, Annotated, Union
|
||
|
||
from langchain_openai import ChatOpenAI
|
||
from langchain_core.messages import HumanMessage
|
||
from langchain_core.output_parsers import PydanticOutputParser
|
||
from langchain_core.prompts import ChatPromptTemplate
|
||
from pydantic import BaseModel, Field, Literal
|
||
|
||
# ---------- Pydantic models ----------
|
||
|
||
class HttpOkEvent(BaseModel):
|
||
kind: Literal["ok"] = Field("ok", description="Event kind: ok")
|
||
status: Literal[200] = Field(200, description="HTTP status code")
|
||
path: str = Field(..., description="Requested path")
|
||
duration_ms: int = Field(..., description="Duration in milliseconds")
|
||
|
||
class HttpErrorEvent(BaseModel):
|
||
kind: Literal["error"] = Field("error", description="Event kind: error")
|
||
status: int = Field(..., description="HTTP error status code (4xx/5xx)")
|
||
path: str = Field(..., description="Requested path")
|
||
error_message: str = Field(..., description="Error description")
|
||
|
||
ApiEvent = Annotated[Union[HttpOkEvent, HttpErrorEvent], Field(discriminator="kind")]
|
||
|
||
# ---------- LLM & Parser ----------
|
||
|
||
llm = ChatOpenAI(
|
||
model="openai/gpt-oss-20b:free",
|
||
base_url="https://openrouter.ai/api/v1",
|
||
api_key=os.getenv("OPENAI_API_KEY"),
|
||
temperature=0.0,
|
||
)
|
||
|
||
parser = PydanticOutputParser(pydantic_object=ApiEvent)
|
||
|
||
prompt = ChatPromptTemplate.from_messages([
|
||
("system", "You are a log parser that outputs structured events as JSON.")
|
||
])
|
||
|
||
# ---------- Helper functions ----------
|
||
|
||
def parse_block(block: str) -> ApiEvent:
|
||
"""Parse a single log block using the LLM and structured output parser."""
|
||
# Construct prompt for the block
|
||
messages = [HumanMessage(content=block.strip())]
|
||
# Ask LLM to output JSON matching the ApiEvent schema
|
||
response = llm.invoke(messages)
|
||
# Parse the response JSON into the Pydantic model
|
||
return parser.parse(response.content)
|
||
|
||
|
||
def split_blocks(text: str) -> List[str]:
|
||
"""Split raw log text into individual event blocks.
|
||
Supports both line‑by‑line and '---' separators.
|
||
"""
|
||
if "---" in text:
|
||
return [b.strip() for b in text.split("---") if b.strip()]
|
||
return [line.strip() for line in text.splitlines() if line.strip()]
|
||
|
||
# ---------- CLI ----------
|
||
|
||
DEFAULT_LOG = textwrap.dedent("""
|
||
200 /api/users 120ms
|
||
404 /api/unknown 30ms
|
||
500 /api/orders 250ms
|
||
200 /api/products 80ms
|
||
403 /api/admin 15ms
|
||
""")
|
||
|
||
def main():
|
||
parser_cli = argparse.ArgumentParser(description="Parse raw log into structured events.")
|
||
parser_cli.add_argument("--log", type=str, help="Path to log file or raw log string.")
|
||
args = parser_cli.parse_args()
|
||
|
||
if args.log:
|
||
if os.path.isfile(args.log):
|
||
raw = open(args.log, "r", encoding="utf-8").read()
|
||
else:
|
||
raw = args.log
|
||
else:
|
||
raw = DEFAULT_LOG
|
||
|
||
blocks = split_blocks(raw)
|
||
events: List[ApiEvent] = []
|
||
for block in blocks:
|
||
try:
|
||
event = parse_block(block)
|
||
events.append(event)
|
||
except Exception as e:
|
||
print(f"Failed to parse block: {block!r}\nError: {e}")
|
||
|
||
# Output structured events
|
||
print("\nParsed events:\n")
|
||
for ev in events:
|
||
print(ev.model_dump())
|
||
|
||
# Pretty table
|
||
print("\nTable:\n")
|
||
header = f"{'kind':<6} | {'path':<15} | {'status':<6} | details"
|
||
print(header)
|
||
print('-' * len(header))
|
||
for ev in events:
|
||
if ev.kind == "ok":
|
||
print(f"{ev.kind:<6} | {ev.path:<15} | {ev.status:<6} | duration {ev.duration_ms}ms")
|
||
else:
|
||
print(f"{ev.kind:<6} | {ev.path:<15} | {ev.status:<6} | error: {ev.error_message}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
"""
|
||
|