From b83c83b25c2634a007bd1255a477c37ff4279cd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=98=D0=BB=D1=8C=D1=8F=205f1b81b8-4f5d-11e8-9c2d-fa7ae01?= =?UTF-8?q?bbebc?= Date: Wed, 1 Jul 2026 19:45:22 +0000 Subject: [PATCH] =?UTF-8?q?fix(needs=5Ffixes):=201=20=D0=B8=D1=81=D0=BF?= =?UTF-8?q?=D1=80=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D0=B9,=200=20=D0=BE?= =?UTF-8?q?=D1=82=D1=81=D1=82=D0=BE=D1=8F=D0=BD=D0=BE=20=E2=80=94=20main.p?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 173 +++++++++++++++++++++++++++++++------------------------- 1 file changed, 95 insertions(+), 78 deletions(-) diff --git a/main.py b/main.py index ad94943..e389bb5 100644 --- a/main.py +++ b/main.py @@ -1,38 +1,53 @@ +#!/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 """ -# 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 +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 langchain_core.prompts import ChatPromptTemplate -from pydantic import BaseModel, Field, Literal +from tabulate import tabulate -# ---------- Pydantic models ---------- +# --------------------------------------------------------------------------- +# 1. Pydantic models – Union with discriminator +# --------------------------------------------------------------------------- class HttpOkEvent(BaseModel): - kind: Literal["ok"] = Field("ok", description="Event kind: ok") + 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") + path: str = Field(..., description="Request path") + duration_ms: int = Field(..., description="Response time 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") + kind: Literal["error"] = Field("error", description="Event kind – Error") + status: int = Field(..., description="HTTP status code (4xx/5xx)") + path: str = Field(..., description="Request path") error_message: str = Field(..., description="Error description") ApiEvent = Annotated[Union[HttpOkEvent, HttpErrorEvent], Field(discriminator="kind")] -# ---------- LLM & Parser ---------- +# --------------------------------------------------------------------------- +# 2. LLM and Agent setup +# --------------------------------------------------------------------------- llm = ChatOpenAI( model="openai/gpt-oss-20b:free", @@ -41,81 +56,83 @@ llm = ChatOpenAI( temperature=0.0, ) -parser = PydanticOutputParser(pydantic_object=ApiEvent) - -prompt = ChatPromptTemplate.from_messages([ - ("system", "You are a log parser that outputs structured events as JSON.") +backend = CompositeBackend([ + LocalShellBackend(workspace_dir="./workspace"), + FilesystemBackend(), ]) -# ---------- Helper functions ---------- +# 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.", +) -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) +# --------------------------------------------------------------------------- +# 3. Parsing helper +# --------------------------------------------------------------------------- +parser = PydanticOutputParser(pydantic_object=ApiEvent) -def split_blocks(text: str) -> List[str]: - """Split raw log text into individual event blocks. - Supports both line‑by‑line and '---' separators. +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. """ - 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()] + 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) -# ---------- CLI ---------- +# --------------------------------------------------------------------------- +# 4. Main logic +# --------------------------------------------------------------------------- -DEFAULT_LOG = textwrap.dedent(""" - 200 /api/users 120ms - 404 /api/unknown 30ms - 500 /api/orders 250ms - 200 /api/products 80ms - 403 /api/admin 15ms -""") +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 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) +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] events: List[ApiEvent] = [] for block in blocks: try: - event = parse_block(block) + event = await 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") + print(f"Failed to parse block: {block}\nError: {e}") + # Output each event as JSON for ev in events: - print(ev.model_dump()) - + print(ev.model_dump_json(indent=2)) # 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}") + 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")) if __name__ == "__main__": - main() -""" - + asyncio.run(main())