diff --git a/main.py b/main.py index e389bb5..3abefb8 100644 --- a/main.py +++ b/main.py @@ -1,138 +1,89 @@ -#!/usr/bin/env python -"""Structured log parser using DeepAgents and Pydantic v2. - -This script demonstrates how to parse a raw log containing HTTP -responses into a list of typed events. It uses the DeepAgents -framework to create a lightweight agent that delegates the parsing -job to an OpenRouter LLM via structured output. The result is a -list of Pydantic models that can be printed or displayed in a table. - -Usage: - python main.py # uses built‑in demo log - python main.py "" # parse custom log passed as a single string -""" - +import argparse import os -import sys -import asyncio from typing import Annotated, Literal, Union, List -from pathlib import Path - -from langchain_openai import ChatOpenAI -from langchain_core.messages import HumanMessage -from langchain.tools import tool -from deepagents import create_deep_agent -from deepagents.backends import FilesystemBackend, LocalShellBackend, CompositeBackend -from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser -from tabulate import tabulate +from langchain_core.runnables import RunnablePassthrough +from langchain_openai import ChatOpenAI +from dotenv import load_dotenv # --------------------------------------------------------------------------- # 1. Pydantic models – Union with discriminator # --------------------------------------------------------------------------- class HttpOkEvent(BaseModel): - kind: Literal["ok"] = Field("ok", description="Event kind – OK") - status: Literal[200] = Field(200, description="HTTP status code") + kind: Literal["ok"] = Field(..., description="Event kind") + status: Literal[200] = Field(..., description="HTTP OK status") path: str = Field(..., description="Request path") - duration_ms: int = Field(..., description="Response time in milliseconds") + 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 status code (4xx/5xx)") + kind: Literal["error"] = Field(..., description="Event kind") + status: int = Field(..., description="HTTP error status") path: str = Field(..., description="Request path") - error_message: str = Field(..., description="Error description") + error_message: str = Field(..., description="Error message") ApiEvent = Annotated[Union[HttpOkEvent, HttpErrorEvent], Field(discriminator="kind")] # --------------------------------------------------------------------------- -# 2. LLM and Agent setup +# 2. LLM and Parser setup # --------------------------------------------------------------------------- -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, -) - -backend = CompositeBackend([ - LocalShellBackend(workspace_dir="./workspace"), - FilesystemBackend(), -]) - -# The agent will simply forward the prompt to the LLM and return the -# structured output. No additional tools are required for this task. -agent = create_deep_agent( - model=llm, - tools=[], - backend=backend, - system_prompt="You are a log‑parsing assistant. Return a structured - representation of the event using the provided Pydantic models.", -) +load_dotenv() +llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_output_tokens=512) +parser = PydanticOutputParser(pydantic_object=ApiEvent) +chain = RunnablePassthrough() | llm | parser # --------------------------------------------------------------------------- # 3. Parsing helper # --------------------------------------------------------------------------- -parser = PydanticOutputParser(pydantic_object=ApiEvent) - -async def parse_block(block: str) -> ApiEvent: - """Ask the LLM to parse a single log block into an ApiEvent. - - The LLM is instructed to output only the JSON that matches the - Pydantic schema. The parser then validates and returns the model. - """ - prompt = ( - "Parse the following log entry and return a JSON object that matches " - "one of the following schemas: HttpOkEvent or HttpErrorEvent. " - "Do not include any additional keys or text. - """ - f"Log entry:\n{block.strip()}" - ) - response = await agent.ainvoke( - {"messages": [HumanMessage(content=prompt)]}, - {"configurable": {"thread_id": "parser"}}, - ) - content = response["messages"][-1].content - return parser.parse(content) +def parse_log_line(line: str) -> ApiEvent: + prompt = f"Parse the following log line into JSON:\n{line}\nThe JSON should match one of the following schemas:\n- ok event: {{\"kind\": \"ok\", status: 200, path: string, duration_ms: int}}\n- error event: {{\"kind\": \"error\", status: int, path: string, error_message: string}}\nReturn only the JSON." + result = chain.invoke(prompt) + return result # --------------------------------------------------------------------------- # 4. Main logic # --------------------------------------------------------------------------- -DEFAULT_LOG = """""" -DEFAULT_LOG += "GET /api/users 200 123ms\n" -DEFAULT_LOG += "POST /api/login 404 Not Found\n" -DEFAULT_LOG += "GET /api/data 500 Internal Server Error\n" -DEFAULT_LOG += "PUT /api/update 200 98ms\n" -DEFAULT_LOG += "DELETE /api/remove 403 Forbidden\n" +def main(): + parser_cli = argparse.ArgumentParser(description="Parse log events into typed objects.") + parser_cli.add_argument("--log", type=str, help="Path to log file or raw log string.") + args = parser_cli.parse_args() -async def main(): - raw_log = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_LOG - # Split into blocks – simple newline split but ignore empty lines - blocks = [b for b in raw_log.strip().split("\n") if b] + if args.log: + if os.path.exists(args.log): + with open(args.log, "r", encoding="utf-8") as f: + raw = f.read() + else: + raw = args.log + else: + raw = """GET /api/users 200 123ms +POST /api/users 404 Not Found +GET /api/orders 500 Internal Server Error +""" + + lines = [l.strip() for l in raw.splitlines() if l.strip()] events: List[ApiEvent] = [] - for block in blocks: + for line in lines: try: - event = await parse_block(block) + event = parse_log_line(line) events.append(event) except Exception as e: - print(f"Failed to parse block: {block}\nError: {e}") - # Output each event as JSON + print(f"Failed to parse line: {line}\nError: {e}") + + print("\nParsed Events:") for ev in events: - print(ev.model_dump_json(indent=2)) - # Pretty table - table = [[ - ev.kind, - ev.path, - ev.status, - getattr(ev, "duration_ms", "-"), - getattr(ev, "error_message", "-"), - ] for ev in events] - headers = ["kind", "path", "status", "duration_ms", "error_message"] - print("\nParsed events table:\n") - print(tabulate(table, headers=headers, tablefmt="github")) + print(ev.model_dump()) + + print("\nTable:\") + header = ["kind", "path", "status"] + print("{:<6} {:<20} {:<6}".format(*header)) + for ev in events: + kind = ev.kind + path = getattr(ev, "path", "") + status = getattr(ev, "status", "") + print("{:<6} {:<20} {:<6}".format(kind, path, status)) if __name__ == "__main__": - asyncio.run(main()) + main()