From 3042cc37caf1ea3bcc4e6b366a64e6e8fab0461a 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: Tue, 30 Jun 2026 19:06:20 +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 | 175 +++++++++++++++++++++++++++++--------------------------- 1 file changed, 90 insertions(+), 85 deletions(-) diff --git a/main.py b/main.py index a5ce2fc..ad94943 100644 --- a/main.py +++ b/main.py @@ -1,31 +1,39 @@ -import os -import asyncio -import argparse -from typing import List, Union, Annotated +""" +# 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 pydantic import BaseModel, Field, Literal 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 langchain_core.output_parsers import PydanticOutputParser +from langchain_core.prompts import ChatPromptTemplate +from pydantic import BaseModel, Field, Literal + +# ---------- Pydantic models ---------- -# ----------------- Pydantic models ----------------- class HttpOkEvent(BaseModel): - kind: Literal["ok"] = Field(..., description="Event kind: ok") - status: Literal[200] = Field(..., description="HTTP status code") - path: str = Field(..., description="Request path") + 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(..., 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 message") + 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 ----------------- +# ---------- LLM & Parser ---------- + llm = ChatOpenAI( model="openai/gpt-oss-20b:free", base_url="https://openrouter.ai/api/v1", @@ -33,84 +41,81 @@ llm = ChatOpenAI( temperature=0.0, ) -# ----------------- Backend ----------------- -backend = CompositeBackend([ - LocalShellBackend(workspace_dir="./workspace"), - FilesystemBackend(), +parser = PydanticOutputParser(pydantic_object=ApiEvent) + +prompt = ChatPromptTemplate.from_messages([ + ("system", "You are a log parser that outputs structured events as JSON.") ]) -# ----------------- Tools ----------------- -@tool -def parse_block(block: str) -> ApiEvent: - """Parse a single log block into a typed ApiEvent using LLM structured output.""" - # Use the LLM to produce structured output - result = llm.with_structured_output(ApiEvent).invoke({"messages": [HumanMessage(content=block)]}) - # The LLM returns an ApiEvent instance directly - return result +# ---------- Helper functions ---------- -@tool -def parse_log(log_text: str) -> str: - """Parse the entire log text and return a formatted table of events.""" - # Split into blocks (empty lines or '---' separators) - raw_blocks = [b.strip() for b in log_text.split("\n") if b.strip()] +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 raw_blocks: + for block in blocks: try: event = parse_block(block) events.append(event) except Exception as e: - # If parsing fails, skip the block but keep a note - events.append( - HttpErrorEvent( - kind="error", - status=0, - path="", - error_message=f"Failed to parse block: {e}", - ) - ) - # Build table - header = f"{'kind':<6} | {'path':<20} | {'status':<6} | {'detail':<30}" - lines = [header, "-" * len(header)] + print(f"Failed to parse block: {block!r}\nError: {e}") + + # Output structured events + print("\nParsed events:\n") for ev in events: - if isinstance(ev, HttpOkEvent): - detail = f"duration {ev.duration_ms}ms" + 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: - detail = ev.error_message - line = f"{ev.kind:<6} | {ev.path:<20} | {ev.status:<6} | {detail:<30}" - lines.append(line) - return "\n".join(lines) - -# ----------------- Agent ----------------- -agent = create_deep_agent( - model=llm, - tools=[parse_block, parse_log], - backend=backend, - system_prompt="You are a log parsing agent. Use the provided tools to parse the log and return a formatted table.", -) - -# ----------------- CLI ----------------- -async def run_agent(log_text: str): - result = await agent.ainvoke( - {"messages": [HumanMessage(content=log_text)]}, - {"configurable": {"thread_id": "log-session"}}, - ) - # The tool output is in the last message - print(result["messages"][-1].content) + print(f"{ev.kind:<6} | {ev.path:<15} | {ev.status:<6} | error: {ev.error_message}") if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Parse raw log into structured events.") - parser.add_argument("--file", type=str, help="Path to a log file. If omitted, uses default sample.") - args = parser.parse_args() - - if args.file: - with open(args.file, "r", encoding="utf-8") as f: - log = f.read() - else: - # Default sample log - log = """ -2023-10-01 12:00:00 INFO /api/users 200 123ms -2023-10-01 12:00:01 ERROR /api/orders 404 Not Found -2023-10-01 12:00:02 INFO /api/products 200 98ms -2023-10-01 12:00:03 ERROR /api/payments 500 Internal Server Error + main() """ - asyncio.run(run_agent(log)) +