From c82bfbce430a15b43402ecc2a9a8638fdd28f2b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=B0=D0=BD=D0=B8=D0=B8=D0=BB=20=D0=92=D0=B8=D0=BA?= =?UTF-8?q?=D1=82=D0=BE=D1=80=D0=BE=D0=B2?= Date: Thu, 2 Jul 2026 05:23:41 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20main.py=20=E2=80=94=20=D0=AD=D0=BA=D0=B7?= =?UTF-8?q?=D0=B0=D0=BC=D0=B5=D0=BD:=20=D0=A1=D0=B0=D0=BC=D0=BE=D0=BA?= =?UTF-8?q?=D0=BE=D1=80=D1=80=D0=B5=D0=BA=D1=82=D0=B8=D1=80=D1=83=D1=8E?= =?UTF-8?q?=D1=89=D0=B8=D0=B9=D1=81=D1=8F=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 156 +++++++++++++++++++++++++++----------------------------- 1 file changed, 76 insertions(+), 80 deletions(-) diff --git a/main.py b/main.py index b5fb030..cae6263 100644 --- a/main.py +++ b/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. - With ~30% probability it raises a ValueError to trigger a retry. + Simulates an unreliable external service. + With ~30% probability it raises a ValueError. """ if random.random() < 0.3: raise ValueError("Simulated tool failure") - # Simple evaluation for arithmetic expressions - try: - result = eval(query, {"__builtins__": {}}) - except Exception: - result = f"cannot evaluate: {query}" - return str(result) + return f"Result for '{query}'" # ---------------------------------------------------------------------- -# 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: - """Run the task using the unreliable tool.""" +async def execute_task(state: AgentState) -> AgentState: try: - # Call the tool directly; deep_agent is not needed here - tool_result = unreliable_tool(state["task"]) - new_state = { - "result": tool_result, - "error": None, - "status": "pending", - } + # Invoke the deep agent with the current task description + response = await deep_agent.ainvoke( + {"messages": [HumanMessage(content=state["task"])]}, + {"configurable": {"thread_id": f"session-{state['attempts'] + 1}"}}, + ) + # 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 = { - "result": "", - "error": str(e), - "status": "failed", - } - return new_state + # Capture any exception from the tool or agent + state["result"] = "" + state["error"] = str(e) + return state # ---------------------------------------------------------------------- # Node: verify_result +# Uses the LLM as a judge to decide if the result is acceptable. # ---------------------------------------------------------------------- -async def verify_result(state: AgentState) -> dict: - """Ask LLM to judge whether the result satisfies the task.""" - judge_prompt = f"""You are a judge. The original task is: -{state['task']} - -The agent produced the following result: -{state['result']} - -Respond with only one word: "success" if the result correctly solves the task, -otherwise respond with "failed".""" +async def verify_result(state: AgentState) -> AgentState: + # Prompt the LLM to judge the result. We ask for a strict "success" or "failed". + judge_prompt = ( + "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, " + "otherwise respond with 'failed'. Do not add any other text." + ) 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) - verdict = response.content.strip().lower() + judge_response = await llm.ainvoke(messages) + verdict = judge_response.content.strip().lower() if verdict == "success": - new_status = "success" + state["status"] = "success" else: - new_status = "failed" - return {"status": new_status, "error": None if new_status == "success" else "Verification failed"} + state["status"] = "failed" + return state # ---------------------------------------------------------------------- # Node: handle_error +# Increments attempts and decides whether to retry or stop. # ---------------------------------------------------------------------- -async def handle_error(state: AgentState) -> dict: - """Increase attempt counter and decide whether to retry.""" - attempts = state["attempts"] + 1 - if attempts >= state["max_attempts"]: - return {"attempts": attempts, "status": "max_attempts", "error": "Maximum attempts reached"} +def handle_error(state: AgentState) -> AgentState: + state["attempts"] += 1 + if state["attempts"] >= state["max_attempts"]: + state["status"] = "max_attempts" 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", - "max_attempts": END, + "retry": "handle_error", }, ) 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, - "messages": [], + "max_attempts": 5, } + 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']}") - if event.get("error"): - print(f"Error: {event['error']}") - if event.get("result"): - print(f"Result: {event['result']}") - final = event - print("\n=== Final Outcome ===") - print(f"Task: {task}") - print(f"Status: {final['status']}") - print(f"Attempts: {final['attempts']}") - if final["status"] == "success": - print(f"Successful result: {final['result']}") - else: - print("Failed to obtain a correct result.") + print(f"Attempt {event['attempts']}: status={event['status']}", end="") + if event["error"]: + print(f", error={event['error']}") + else: + print(f", result={event['result'][:50]}") + if event["status"] in ("success", "max_attempts"): + print("\nFinal status:", event["status"]) + print("Result:", event["result"]) + break if __name__ == "__main__": - # Example task: simple arithmetic - example_task = "2 + 2" - asyncio.run(run_task(example_task, max_attempts=5)) \ No newline at end of file + asyncio.run(main()) \ No newline at end of file