feat: solution for 'Экзамен: Планирующий агент'

README.md

@@ -1,11 +1,11 @@
-# Экзамен: Самокорректирующийся агент
+# Экзамен: Планирующий агент
 
 Главная
 Мои задания
-Экзамен: Самокорректирующийся агент
+Экзамен: Планирующий агент
 5Д
 EN
-Экзамен: Самокорректирующийся агент
+Экзамен: Планирующий агент
 Зачёт
 Версия 1
 Дедлайн сдачи: 31.08.2026
@@ -28,7 +28,7 @@ EN
 
 Задание
 
-Практическое задание: Самокорректирующийся агент
+Практическое задание: Планирующий агент
 Цель
 
-Реализовать LangGraph-агента, который после выполнения задачи проверяет результат (LLM-as-
+Собрать LangGraph-агента с отдельным этапом планирования: сначала LLM разбивает задачу на шаги, затем выполняет их по

requirements.txt

@@ -1,4 +1,3 @@
-langchain-core>=0.2.0
+langgraph
-langchain-openai>=0.2.0
+langchain-openai
-pydantic>=2.0
+openai
-python-dotenv>=1.0

src/graph.py

@@ -0,0 +1,92 @@
+import json
+from typing import Dict, Any
+from langgraph.graph import StateGraph, END
+from langchain_openai import ChatOpenAI
+from .state import PlanningState
+
+def planning(state: PlanningState) -> PlanningState:
+    """LLM node that splits the task into 3‑6 concrete steps."""
+    llm = ChatOpenAI(temperature=0)
+    prompt = (
+        f"Task: {state['task']}\n\n"
+        "Please break this task into 3-6 concrete steps. "
+        "Return the steps as a numbered list or a JSON array. "
+        "Do not add any extra text."
+    )
+    response = llm.invoke(prompt)
+    text = response.content.strip()
+
+    # Try to parse JSON first
+    plan: List[str] | None = None
+    try:
+        parsed = json.loads(text)
+        if isinstance(parsed, list):
+            plan = [str(item) for item in parsed]
+    except Exception:
+        pass
+
+    # Fallback: parse numbered list
+    if plan is None:
+        plan = []
+        for line in text.splitlines():
+            line = line.strip()
+            if not line:
+                continue
+            # Remove leading number if present
+            if '.' in line:
+                _, rest = line.split('.', 1)
+                step = rest.strip()
+            else:
+                step = line
+            plan.append(step)
+
+    state["plan"] = plan
+    state["current_step"] = 0
+    state["results"] = []
+    return state
+
+def execution(state: PlanningState) -> PlanningState:
+    """Execute one step of the plan."""
+    llm = ChatOpenAI(temperature=0)
+    step = state["plan"][state["current_step"]]
+    prompt = (
+        f"Task: {state['task']}\n\n"
+        f"You are executing step {state['current_step'] + 1} of the plan.\n\n"
+        f"Step: {step}\n\n"
+        "Provide the result of this step."
+    )
+    response = llm.invoke(prompt)
+    result = response.content.strip()
+    state["results"].append(result)
+    state["current_step"] += 1
+    return state
+
+def should_continue(state: PlanningState) -> str:
+    """Decide whether to loop back to execution or finish."""
+    if state["current_step"] < len(state["plan"]):
+        return "execute"
+    return "finish"
+
+def create_graph() -> StateGraph:
+    graph = StateGraph(PlanningState)
+
+    graph.add_node("planning", planning)
+    graph.add_node("execution", execution)
+    graph.add_node("finish", lambda state: state)
+
+    graph.add_conditional_edges(
+        "planning",
+        lambda _: "execute",
+        {"execute": "execution"}
+    )
+
+    graph.add_conditional_edges(
+        "execution",
+        should_continue,
+        {"execute": "execution", "finish": "finish"}
+    )
+
+    graph.set_entry_point("planning")
+    graph.set_finish_point("finish")
+
+    return graph

src/main.py

@@ -1,116 +1,34 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-"""
-Assignment Card Extraction
-
-This script demonstrates how to extract structured assignment details from a
-natural language description using LangChain and Pydantic.
-"""
-
 import os
-from typing import List
+from src.graph import create_graph
+from src.state import PlanningState
 
-from dotenv import load_dotenv
-from langchain_core.output_parsers import PydanticOutputParser
-from langchain_core.prompts import PromptTemplate
-from langchain_openai import ChatOpenAI
-from pydantic import BaseModel, Field
-
-# Load environment variables (expects OPENAI_API_KEY)
-load_dotenv()
-
-# --------------------------------------------------------------------------- #
-# Pydantic model definition
-# --------------------------------------------------------------------------- #
-class AssignmentCard(BaseModel):
-    """
-    Structured representation of an assignment description.
-    """
-
-    title: str = Field(
-        ...,
-        description="Short title of the assignment (e.g., 'Mini-report on LangChain').",
-    )
-    subject: str = Field(
-        ...,
-        description="Subject or topic of the assignment (e.g., 'LangChain').",
-    )
-    deadline_hint: str = Field(
-        ...,
-        description="A short phrase indicating the deadline (e.g., 'by Friday').",
-    )
-    deliverable_type: str = Field(
-        ...,
-        description="What to submit: report, code, presentation, etc.",
-    )
-    grading_hints: List[str] = Field(
-        ...,
-        description="List of key grading criteria mentioned in the description.",
-    )
-
-# --------------------------------------------------------------------------- #
-# LangChain components
-# --------------------------------------------------------------------------- #
-# Parser that will convert the LLM output into an AssignmentCard instance
-parser = PydanticOutputParser(pydantic_object=AssignmentCard)
-
-# Prompt template that instructs the LLM to output JSON matching the model
-prompt = PromptTemplate(
-    template=(
-        "You are an assignment extraction assistant. "
-        "Given the following assignment description, extract the following fields:\n\n"
-        "- title: short title of the assignment\n"
-        "- subject: subject or topic\n"
-        "- deadline_hint: a short phrase indicating the deadline\n"
-        "- deliverable_type: what to submit (e.g., report, code, presentation)\n"
-        "- grading_hints: list of key grading criteria mentioned\n\n"
-        "Return a JSON object with exactly these keys. Do not include any additional keys or text.\n\n"
-        "Description: {description}\n\n"
-        "{format_instructions}"
-    ),
-    input_variables=["description"],
-    partial_variables={"format_instructions": parser.get_format_instructions()},
-)
-
-# LLM configuration
-llm = ChatOpenAI(
-    temperature=0,
-    model="gpt-3.5-turbo",
-)
-
-# Chain: prompt -> LLM -> parser
-chain = prompt | llm | parser
-
-# --------------------------------------------------------------------------- #
-# Main execution
-# --------------------------------------------------------------------------- #
 def main() -> None:
-    # Sample assignment description
+    # Ensure the OpenAI API key is set
-    sample_description = (
+    if "OPENAI_API_KEY" not in os.environ:
-        "Сдайте к пятнице мини-отчёт по LangChain: 2 страницы, упор на агентов. "
+        raise RuntimeError("Please set the OPENAI_API_KEY environment variable.")
-        "Оценка: за полноту и за пример кода."
-    )
 
-    # Run the chain
+    task = "Compare Python and JavaScript"
-    try:
+    initial_state: PlanningState = {
-        result = chain.invoke({"description": sample_description})
+        "task": task,
-    except Exception as e:
+        "plan": None,
-        print(f"Error during chain execution: {e}")
+        "current_step": 0,
-        return
+        "results": []
+    }
 
-    # The result is already a validated AssignmentCard instance
+    graph = create_graph()
-    print("\n=== Parsed Assignment Card ===")
+    final_state = graph.invoke(initial_state)
-    print(result.model_dump(indent=2))
 
-    # Human-readable summary
+    print("\n=== Plan ===")
-    print("\n=== Human-readable Summary ===")
+    for i, step in enumerate(final_state["plan"], 1):
-    print(f"Title: {result.title}")
+        print(f"{i}. {step}")
-    print(f"Subject: {result.subject}")
-    print(f"Deadline: {result.deadline_hint}")
-    print(f"Deliverable: {result.deliverable_type}")
-    print(f"Grading Hints: {', '.join(result.grading_hints)}")
 
+    print("\n=== Results ===")
+    for i, res in enumerate(final_state["results"], 1):
+        print(f"[Step {i}] {res}")
+
+    print("\n=== Final Summary ===")
+    summary = "\n".join(final_state["results"])
+    print(summary)
 
 if __name__ == "__main__":
     main()

src/state.py

@@ -1,9 +1,7 @@
-from typing import TypedDict, Optional
+from typing import TypedDict, List, Optional
 
-class AgentState(TypedDict):
+class PlanningState(TypedDict):
     task: str
-    result: str
+    plan: Optional[List[str]]
-    attempts: int
+    current_step: int
-    status: str  # pending | success | failed | max_attempts
+    results: List[str]
-    error: Optional[str]
-    max_attempts: int