Merge branch 'next' into chris-villegas/AGX1-241/implement-sdk-credential-abstraction

Author: cdvillegas

examples/tutorials/10_async/10_temporal/130_langgraph/project/graph.py

@@ -10,6 +10,15 @@
 The router and tools are ``async`` so LangGraph awaits them directly — a sync
 callable would be offloaded via ``run_in_executor``, which Temporal's workflow
 event loop does not support.
+
+The in-workflow ``tools`` node is a plain ``async`` function rather than
+LangGraph's ``ToolNode`` prebuilt on purpose. The plugin wraps an in-workflow
+node in ``wrap_workflow``, whose closure captures the wrapped object. When that
+object is itself a LangChain ``Runnable`` (as ``ToolNode`` is), LangGraph's
+``compile()`` subgraph detection (``find_subgraph_pregel`` →
+``get_function_nonlocals``) recurses through that wrapper without cycle
+detection and never terminates, tripping Temporal's deadlock detector. A plain
+function isn't a ``Runnable``, so compile stays trivial.
 """
 
 from __future__ import annotations
@@ -26,12 +35,14 @@
 
 from langgraph.graph import END, START, StateGraph
 from langchain_openai import ChatOpenAI
-from langgraph.prebuilt import ToolNode
-from langchain_core.messages import SystemMessage
+from langchain_core.messages import ToolMessage, SystemMessage
 from langgraph.graph.message import add_messages
 
 from project.tools import TOOLS
 
+# Look up tools by name for the in-workflow tools node.
+_TOOLS_BY_NAME = {tool.name: tool for tool in TOOLS}
+
 # Name this graph is registered under in the LangGraphPlugin (acp.py / run_worker.py).
 GRAPH_NAME = "at130-langgraph"
 MODEL_NAME = "gpt-4o"
@@ -58,6 +69,28 @@ async def agent_node(state: AgentState) -> dict[str, Any]:
     return {"messages": [await llm.ainvoke(messages)]}
 
 
+async def tools_node(state: AgentState) -> dict[str, Any]:
+    """Run the tool calls the model requested. Runs inline in the workflow.
+
+    A plain ``async`` function (not LangGraph's ``ToolNode``) — see the module
+    docstring for why a ``Runnable`` tools node can't be compiled here.
+    """
+    last = state["messages"][-1]
+    results: list[Any] = []
+    for call in getattr(last, "tool_calls", None) or []:
+        tool = _TOOLS_BY_NAME.get(call["name"])
+        # Mirror ToolNode: surface an unknown/hallucinated tool name as an error
+        # ToolMessage so the graph keeps running instead of crashing the node.
+        if tool is None:
+            output = f"Error: unknown tool {call['name']!r}. Available: {list(_TOOLS_BY_NAME)}"
+        else:
+            output = await tool.ainvoke(call["args"])
+        results.append(
+            ToolMessage(content=str(output), tool_call_id=call["id"], name=call["name"])
+        )
+    return {"messages": results}
+
+
 async def route_after_agent(state: AgentState) -> str:
     """Go to the tools node if the model requested tools, else finish (async router)."""
     last = state["messages"][-1]
@@ -72,7 +105,7 @@ def build_graph() -> StateGraph:
         agent_node,
         metadata={"execute_in": "activity", "start_to_close_timeout": timedelta(minutes=5)},
     )
-    builder.add_node("tools", ToolNode(TOOLS), metadata={"execute_in": "workflow"})
+    builder.add_node("tools", tools_node, metadata={"execute_in": "workflow"})
     builder.add_edge(START, "agent")
     builder.add_conditional_edges("agent", route_after_agent, {"tools": "tools", END: END})
     builder.add_edge("tools", "agent")