Compare commits

...
2 changed files with 78 additions and 1 deletions

No files matched your search

+34
View File
@@ -232,6 +232,8 @@ def draw_graph(
tools_by_name = getattr(node.bound, "tools_by_name", None)
if isinstance(tools_by_name, Mapping) and tools_by_name:
metadata["__tools"] = list(tools_by_name)
if subagents := _get_tool_subagents(tools_by_name):
metadata["__subagents"] = subagents
graph.add_node(node.bound, name, metadata=metadata or None)
# add start node
if START not in nodes:
@@ -281,6 +283,38 @@ def draw_graph(
return graph
def _get_tool_subagents(
tools_by_name: Mapping[str, Any],
) -> dict[str, list[dict[str, Any]]]:
"""Collect the subagents each tool can run, keyed by tool name."""
subagents: dict[str, list[dict[str, Any]]] = {}
for tool_name, tool in tools_by_name.items():
tool_subagents = getattr(tool, "subagents", None)
if (
isinstance(tool_subagents, Sequence)
and not isinstance(tool_subagents, str)
and tool_subagents
and all(
isinstance(subagent, Mapping) and isinstance(subagent.get("name"), str)
for subagent in tool_subagents
)
):
subagents[tool_name] = [
_subagent_entry(subagent) for subagent in tool_subagents
]
return subagents
def _subagent_entry(subagent: Mapping[str, Any]) -> dict[str, Any]:
"""Keep only the subagent fields graph viewers display."""
entry: dict[str, Any] = {"name": subagent["name"]}
if isinstance(description := subagent.get("description"), str):
entry["description"] = description
if subagent.get("forked") is True:
entry["forked"] = True
return entry
def add_edge(
graph: Graph,
source: str,
+44 -1
View File
@@ -36,7 +36,7 @@ from langchain_core.runnables import (
RunnablePassthrough,
)
from langchain_core.runnables.graph import Edge
from langchain_core.tools import tool
from langchain_core.tools import StructuredTool, tool
from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import (
@@ -3105,6 +3105,49 @@ def test_simple_multi_edge(snapshot: SnapshotAssertion) -> None:
)
def test_tool_node_lists_tool_subagents_in_graph_metadata() -> None:
class TaskTool(StructuredTool):
subagents: list[dict[str, Any]] = []
def run_task(description: str, subagent_type: str) -> str:
return subagent_type
task = TaskTool.from_function(run_task, name="task", description="Run a subagent.")
task.subagents = [
{"name": "researcher", "description": "Researches a topic."},
{
"name": "critic",
"description": "Reviews drafts.",
"forked": True,
"system_prompt": "You are a critic.",
"runnable": object(),
},
{"name": "planner", "description": None, "forked": "yes"},
]
@tool
def search(query: str) -> str:
"""Search the web."""
return query
builder = StateGraph(MessagesState)
builder.add_node("agent", lambda state: state)
builder.add_node("tools", ToolNode([search, task]))
builder.add_edge(START, "agent")
builder.add_edge("agent", "tools")
graph = builder.compile().get_graph().to_json()
nodes = {node["id"]: node for node in graph["nodes"]}
assert nodes["tools"]["metadata"]["__subagents"] == {
"task": [
{"name": "researcher", "description": "Researches a topic."},
{"name": "critic", "description": "Reviews drafts.", "forked": True},
{"name": "planner"},
]
}
assert "__subagents" not in (nodes["agent"].get("metadata") or {})
def test_nested_graph_xray(snapshot: SnapshotAssertion) -> None:
class State(TypedDict):
my_key: Annotated[str, operator.add]