fix: main.py — Экзамен: Самокорректирующийся агент

main.py

@@ -1,10 +1,10 @@
 import os
 import asyncio
 import random
-from typing import TypedDict, Literal, Annotated
+from typing import TypedDict, Annotated
 
 from langchain_openai import ChatOpenAI
-from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.messages import HumanMessage
 from langchain.tools import tool
 
 from deepagents import create_deep_agent
@@ -14,7 +14,7 @@ from langgraph.graph import StateGraph, START, END
 from langgraph.graph.message import add_messages
 
 # ----------------------------------------------------------------------
-# Configuration
+# LLM configuration (OpenRouter, free tier)
 # ----------------------------------------------------------------------
 llm = ChatOpenAI(
     model="openai/gpt-oss-20b:free",
@@ -23,6 +23,9 @@ llm = ChatOpenAI(
     temperature=0.0,
 )
 
+# ----------------------------------------------------------------------
+# Backend for deepagents (required by the framework)
+# ----------------------------------------------------------------------
 backend = CompositeBackend(
     [
         LocalShellBackend(workspace_dir="./workspace"),
@@ -31,25 +34,20 @@ backend = CompositeBackend(
 )
 
 # ----------------------------------------------------------------------
-# Unreliable tool used for demonstration
+# Unreliable tool used to demonstrate retry logic
 # ----------------------------------------------------------------------
 @tool
 def unreliable_tool(query: str) -> str:
     """
-    Simulates an unreliable external tool.
+    Simulates an unreliable external service.
-    With ~30% probability it raises a ValueError to trigger a retry.
+    With ~30% probability it raises a ValueError.
     """
     if random.random() < 0.3:
         raise ValueError("Simulated tool failure")
-    # Simple evaluation for arithmetic expressions
+    return f"Result for '{query}'"
-    try:
-        result = eval(query, {"__builtins__": {}})
-    except Exception:
-        result = f"cannot evaluate: {query}"
-    return str(result)
 
 # ----------------------------------------------------------------------
-# DeepAgent - required by the course
+# DeepAgent creation (required by the course)
 # ----------------------------------------------------------------------
 deep_agent = create_deep_agent(
     model=llm,
@@ -59,76 +57,74 @@ deep_agent = create_deep_agent(
 )
 
 # ----------------------------------------------------------------------
-# Agent state definition
+# State definition for the LangGraph workflow
 # ----------------------------------------------------------------------
 class AgentState(TypedDict):
     task: str
     result: str
     attempts: int
-    status: Literal["pending", "success", "failed", "max_attempts"]
+    status: str  # pending | success | failed | max_attempts
     error: str | None
     max_attempts: int
-    messages: Annotated[list, add_messages]
 
 # ----------------------------------------------------------------------
 # Node: execute_task
+# Calls the deep agent to perform the task using the unreliable tool.
 # ----------------------------------------------------------------------
-async def execute_task(state: AgentState) -> dict:
+async def execute_task(state: AgentState) -> AgentState:
-    """Run the task using the unreliable tool."""
     try:
-        # Call the tool directly; deep_agent is not needed here
+        # Invoke the deep agent with the current task description
-        tool_result = unreliable_tool(state["task"])
+        response = await deep_agent.ainvoke(
-        new_state = {
+            {"messages": [HumanMessage(content=state["task"])]},
-            "result": tool_result,
+            {"configurable": {"thread_id": f"session-{state['attempts'] + 1}"}},
-            "error": None,
+        )
-            "status": "pending",
+        # Extract the assistant's final message
-        }
+        result_msg = response["messages"][-1].content
+        state["result"] = result_msg
+        state["error"] = None
     except Exception as e:
-        new_state = {
+        # Capture any exception from the tool or agent
-            "result": "",
+        state["result"] = ""
-            "error": str(e),
+        state["error"] = str(e)
-            "status": "failed",
+    return state
-        }
-    return new_state
 
 # ----------------------------------------------------------------------
 # Node: verify_result
+# Uses the LLM as a judge to decide if the result is acceptable.
 # ----------------------------------------------------------------------
-async def verify_result(state: AgentState) -> dict:
+async def verify_result(state: AgentState) -> AgentState:
-    """Ask LLM to judge whether the result satisfies the task."""
+    # Prompt the LLM to judge the result. We ask for a strict "success" or "failed".
-    judge_prompt = f"""You are a judge. The original task is:
+    judge_prompt = (
-{state['task']}
+        "You are a verifier. Given the original task and the agent's result, "
-
+        "respond with only the word 'success' if the result correctly fulfills the task, "
-The agent produced the following result:
+        "otherwise respond with 'failed'. Do not add any other text."
-{state['result']}
+    )
-
-Respond with only one word: "success" if the result correctly solves the task,
-otherwise respond with "failed"."""
     messages = [
-        SystemMessage(content="You are an objective judge."),
         HumanMessage(content=judge_prompt),
+        HumanMessage(content=f"Task: {state['task']}\nResult: {state['result']}"),
     ]
-    response = await llm.ainvoke(messages)
+    judge_response = await llm.ainvoke(messages)
-    verdict = response.content.strip().lower()
+    verdict = judge_response.content.strip().lower()
     if verdict == "success":
-        new_status = "success"
+        state["status"] = "success"
     else:
-        new_status = "failed"
+        state["status"] = "failed"
-    return {"status": new_status, "error": None if new_status == "success" else "Verification failed"}
+    return state
 
 # ----------------------------------------------------------------------
 # Node: handle_error
+# Increments attempts and decides whether to retry or stop.
 # ----------------------------------------------------------------------
-async def handle_error(state: AgentState) -> dict:
+def handle_error(state: AgentState) -> AgentState:
-    """Increase attempt counter and decide whether to retry."""
+    state["attempts"] += 1
-    attempts = state["attempts"] + 1
+    if state["attempts"] >= state["max_attempts"]:
-    if attempts >= state["max_attempts"]:
+        state["status"] = "max_attempts"
-        return {"attempts": attempts, "status": "max_attempts", "error": "Maximum attempts reached"}
     else:
-        return {"attempts": attempts, "status": "pending", "error": None, "result": ""}
+        state["status"] = "pending"
+    return state
 
 # ----------------------------------------------------------------------
-# Build the StateGraph
+# Build the StateGraph with the defined nodes and transitions
 # ----------------------------------------------------------------------
 workflow = StateGraph(AgentState)
 
@@ -140,49 +136,49 @@ workflow.add_edge(START, "execute_task")
 workflow.add_edge("execute_task", "verify_result")
 workflow.add_conditional_edges(
     "verify_result",
-    lambda state: state["status"],
+    lambda state: "success" if state["status"] == "success" else "retry",
     {
         "success": END,
-        "failed": "handle_error",
+        "retry": "handle_error",
-        "max_attempts": END,
     },
 )
 workflow.add_edge("handle_error", "execute_task")
+workflow.add_conditional_edges(
+    "handle_error",
+    lambda state: "end" if state["status"] in ("max_attempts", "success") else "retry",
+    {
+        "end": END,
+        "retry": "execute_task",
+    },
+)
 
 graph = workflow.compile()
 
 # ----------------------------------------------------------------------
-# Main entry point
+# Main entry point: runs the graph for a sample task and prints progress
 # ----------------------------------------------------------------------
-async def run_task(task: str, max_attempts: int = 5):
+async def main():
     initial_state: AgentState = {
-        "task": task,
+        "task": "Calculate 2+2 and return the answer as a plain number.",
         "result": "",
         "attempts": 0,
         "status": "pending",
         "error": None,
-        "max_attempts": max_attempts,
+        "max_attempts": 5,
-        "messages": [],
     }
+
     async for event in graph.astream(initial_state):
-        # Print progress information
+        # The graph yields intermediate states; we log useful info.
         if "attempts" in event:
-            print(f"Attempt {event['attempts']}: status={event['status']}")
+            print(f"Attempt {event['attempts']}: status={event['status']}", end="")
-        if event.get("error"):
+            if event["error"]:
-            print(f"Error: {event['error']}")
+                print(f", error={event['error']}")
-        if event.get("result"):
+            else:
-            print(f"Result: {event['result']}")
+                print(f", result={event['result'][:50]}")
-    final = event
+        if event["status"] in ("success", "max_attempts"):
-    print("\n=== Final Outcome ===")
+            print("\nFinal status:", event["status"])
-    print(f"Task: {task}")
+            print("Result:", event["result"])
-    print(f"Status: {final['status']}")
+            break
-    print(f"Attempts: {final['attempts']}")
-    if final["status"] == "success":
-        print(f"Successful result: {final['result']}")
-    else:
-        print("Failed to obtain a correct result.")
 
 if __name__ == "__main__":
-    # Example task: simple arithmetic
+    asyncio.run(main())
-    example_task = "2 + 2"
-    asyncio.run(run_task(example_task, max_attempts=5))