Fix lint errors in langgraph samples

- Fix import sorting (I001) across all sample modules
- Apply ruff formatting to 11 files
- Add type annotations for mypy (executor_agent, critique, plan)
- Use cast() for structured output return types

Author: mfateev

langgraph_samples/agentic_rag/graph.py

@@ -29,7 +29,6 @@
 import os
 from typing import Annotated, Any, Literal, Sequence, cast
 
-from langchain.agents import create_agent
 from langchain_core.documents import Document
 from langchain_core.messages import BaseMessage, HumanMessage
 from langchain_core.output_parsers import StrOutputParser
@@ -42,6 +41,8 @@
 from pydantic import BaseModel, Field
 from typing_extensions import TypedDict
 
+from langchain.agents import create_agent
+
 # Sample documents about AI agents and LangGraph for the knowledge base
 SAMPLE_DOCUMENTS = [
     Document(

langgraph_samples/agentic_rag/workflow.py

@@ -54,8 +54,6 @@ async def run(self, query: str) -> dict[str, Any]:
 
         # Execute the agent
         # The input format is a dict with "messages"
-        result = await app.ainvoke(
-            {"messages": [{"role": "user", "content": query}]}
-        )
+        result = await app.ainvoke({"messages": [{"role": "user", "content": query}]})
 
         return result

langgraph_samples/approval_workflow/activities.py

@@ -36,8 +36,12 @@ async def notify_approver(request_info: dict) -> str:
     print(f"Workflow ID: {workflow_id}")
     print(f"Request: {message}")
     print(f"\nTo respond, run:")
-    print(f"  Approve: uv run python -m langgraph_samples.approval_workflow.run_respond {workflow_id} --approve --reason 'Your reason'")
-    print(f"  Reject:  uv run python -m langgraph_samples.approval_workflow.run_respond {workflow_id} --reject --reason 'Your reason'")
+    print(
+        f"  Approve: uv run python -m langgraph_samples.approval_workflow.run_respond {workflow_id} --approve --reason 'Your reason'"
+    )
+    print(
+        f"  Reject:  uv run python -m langgraph_samples.approval_workflow.run_respond {workflow_id} --reject --reason 'Your reason'"
+    )
     print()
 
     return f"Notification sent for workflow {workflow_id}"

langgraph_samples/approval_workflow/run_workflow.py

@@ -40,7 +40,9 @@ async def main() -> None:
     )
 
     print(f"Workflow started. Waiting for result...")
-    print(f"\nTo approve/reject, use the run_respond script (see worker output for commands)")
+    print(
+        f"\nTo approve/reject, use the run_respond script (see worker output for commands)"
+    )
 
     # Wait for the workflow to complete
     result = await handle.result()

langgraph_samples/approval_workflow/workflow.py

@@ -16,11 +16,11 @@
 
 with workflow.unsafe.imports_passed_through():
     from langgraph.types import Command
+    from temporalio.contrib.langgraph import compile as lg_compile
 
     from langgraph_samples.approval_workflow.activities import (
         notify_approver,
     )
-    from temporalio.contrib.langgraph import compile as lg_compile
 
 
 @dataclass
@@ -147,7 +147,9 @@ async def run(
 
             workflow.logger.info(
                 "Received approval response: approved=%s",
-                self._approval_response.get("approved") if self._approval_response else None,
+                self._approval_response.get("approved")
+                if self._approval_response
+                else None,
             )
 
             # Resume with the approval response

langgraph_samples/deep_research/graph.py

@@ -23,7 +23,7 @@
 """
 
 import os
-from typing import Annotated, Any, Literal
+from typing import Annotated, Any, Literal, cast
 
 from langchain_community.tools import DuckDuckGoSearchRun
 from langchain_core.messages import BaseMessage, HumanMessage
@@ -142,7 +142,7 @@ def plan_research(state: ResearchState) -> dict[str, Any]:
         )
 
         planner = plan_prompt | model.with_structured_output(ResearchPlan)
-        plan = planner.invoke({"topic": topic_str})
+        plan = cast(ResearchPlan, planner.invoke({"topic": topic_str}))
 
         return {
             "topic": topic_str,
@@ -253,9 +253,7 @@ def synthesize_report(state: ResearchState) -> dict[str, Any]:
         findings = []
         for result in results:
             if result.get("relevant", False):
-                findings.append(
-                    f"### {result['purpose']}\n{result['results']}"
-                )
+                findings.append(f"### {result['purpose']}\n{result['results']}")
 
         if not findings:
             findings_text = "Limited information was found on this topic."

langgraph_samples/deep_research/workflow.py

@@ -32,9 +32,7 @@ class DeepResearchWorkflow:
     """
 
     @workflow.run
-    async def run(
-        self, topic: str, max_iterations: int = 2
-    ) -> dict[str, Any]:
+    async def run(self, topic: str, max_iterations: int = 2) -> dict[str, Any]:
         """Run deep research on a topic.
 
         Args:

langgraph_samples/hello_world/graph.py

@@ -9,7 +9,6 @@
 from langgraph.graph import END, START, StateGraph
 from typing_extensions import TypedDict
 
-
 # =============================================================================
 # State Definition
 # =============================================================================

langgraph_samples/plan_and_execute/graph.py

@@ -23,7 +23,6 @@
 import os
 from typing import Annotated, Any, Literal
 
-from langchain.agents import create_agent
 from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
 from langchain_core.output_parsers import StrOutputParser
 from langchain_core.prompts import ChatPromptTemplate
@@ -34,6 +33,8 @@
 from pydantic import BaseModel, Field
 from typing_extensions import TypedDict
 
+from langchain.agents import create_agent
+
 
 class PlanStep(BaseModel):
     """A single step in the execution plan."""
@@ -167,7 +168,7 @@ def build_plan_and_execute_graph() -> Any:
 
     # Build executor agent with tools
     tools = [calculate, lookup, analyze]
-    executor_agent = create_agent(model, tools)
+    executor_agent: Any = create_agent(model, tools)
 
     def create_plan(state: PlanExecuteState) -> dict[str, Any]:
         """Create an execution plan from the objective.
@@ -243,9 +244,7 @@ def execute_step(state: PlanExecuteState) -> dict[str, Any]:
         if result.get("messages"):
             last_msg = result["messages"][-1]
             result_content = (
-                last_msg.content
-                if hasattr(last_msg, "content")
-                else str(last_msg)
+                last_msg.content if hasattr(last_msg, "content") else str(last_msg)
             )
 
         step_result = StepResult(

langgraph_samples/react_agent/graph.py

@@ -11,9 +11,9 @@
 import os
 from typing import Any
 
-from langchain.agents import create_agent
 from langchain_openai import ChatOpenAI
 
+from langchain.agents import create_agent
 from langgraph_samples.react_agent.tools import calculate, get_weather