Add dynamic node/task summaries to LangGraph plugin (#1612)

* Add dynamic node/task summaries to LangGraph plugin

Adds a per-node/per-task summary_fn(args, kwargs) -> str | None (and a
plugin-wide default_summary_fn) that computes a Temporal summary at
runtime from the node's input.

- execute_in="activity" nodes: the result sets the activity summary
  (user_metadata, shown on each scheduled-activity event).
- execute_in="workflow" nodes: the result updates the workflow's current
  details via workflow.set_current_details (last-writer-wins).

A static summary activity option already flowed through to
execute_activity; this is now documented. Setting both a static summary
and summary_fn on the same node raises ValueError. summary_fn runs in
workflow context (must be deterministic and must not raise) and is
replay-safe, since summaries ride in user_metadata.

* Comment why summary_fn is popped from node opts

* Scope summary_fn per-node and normalize static summary

- Drop the plugin-level default_summary_fn; summary_fn is now set only
  per node (Graph metadata) / per task (activity_options), matching the
  ADK plugin's per-model scoping so each node manages its own summary.
- Normalize a static summary into a summary_fn (a constant function), so
  static and dynamic summaries share one resolution path feeding the
  activity summary kwarg / set_current_details.
- Workflow-bound nodes now clear current details when summary_fn returns
  empty, so a node no longer shows a stale summary from an earlier node.

* Layer node summary settings over defaults instead of conflicting

summary and summary_fn are two forms of one setting, but as separate
dict keys a default of one form and a node-level value of the other
survived the merge together and tripped the exclusivity guard (e.g.
default_activity_options={"summary_fn": fn} + one node with a static
summary raised ValueError). Merge now drops the inherited form when a
node/task supplies either, so the more-specific setting wins, and the
exclusivity check only rejects both forms set at the same level
(per-node, or in default_activity_options via a new init check).

Author: DABH

temporalio/contrib/langgraph/README.md

@@ -143,6 +143,46 @@ await g.ainvoke({...}, context=Context(user_id="alice"))
 
 Your `context` object must be serializable by the configured Temporal payload converter, since it crosses the Activity boundary.
 
+## Summaries
+
+Summaries are short, human-readable labels that show up in the Temporal UI and CLI, making it easier to see what each step of a run is doing.
+
+### Static summary
+
+`summary` is an ordinary Activity option, so a fixed per-node label works today — pass it like any other option:
+
+```python
+g.add_node("plan", plan, metadata={"execute_in": "activity", "summary": "Planning step"})
+```
+
+It is attached to the node's scheduled-activity event (`execute_in="activity"` only).
+
+### Dynamic summary (`summary_fn`)
+
+To derive the label from the node's input at runtime, supply a `summary_fn`. It receives the node's `(args, kwargs)` and returns a summary string, or `None`/`""` for no summary. For a `StateGraph` node `args[0]` is the state; for a Functional `@task` it is the task's arguments.
+
+```python
+def summarize(args, kwargs) -> str | None:
+    state = args[0]
+    return f"stage={state['stage']} doc={state['doc_id']}"
+
+# Graph API: per-node
+g.add_node("plan", plan, metadata={"execute_in": "activity", "summary_fn": summarize})
+
+# Functional API: per-task
+plugin = LangGraphPlugin(
+    tasks=[plan],
+    activity_options={"plan": {"execute_in": "activity", "summary_fn": summarize}},
+)
+```
+
+`summary_fn` is set per node/task (like the static `summary`), so different nodes — which receive different inputs — can compute their summaries independently. You can also put a `summary` or `summary_fn` in `default_activity_options` as a fallback for every node; a node/task that sets either form overrides the inherited default (you just can't set both forms at the same level).
+
+- For `execute_in="activity"` nodes the result sets the activity `summary` (one per scheduled-activity event, visible in history).
+- For `execute_in="workflow"` nodes there is no activity, so the result updates the workflow's current details via [`workflow.set_current_details()`](https://python.temporal.io/temporalio.workflow.html#set_current_details). This is a single workflow-level slot (last-writer-wins) reflecting the most recent workflow-bound node that defines a `summary_fn`; a `None`/`""` result clears it. It is queryable via `__temporal_workflow_metadata`.
+
+`summary_fn` runs in workflow context on every replay, so it **must be deterministic and must not raise** (an exception fails the workflow task). Setting both a static `summary` and a `summary_fn` on the same node raises `ValueError`.
+
 ## Streaming
 
 When `streaming_topic` is set on `LangGraphPlugin`, calls to `langgraph.config.get_stream_writer()` inside a node publish to the named topic on the workflow's [`WorkflowStream`](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/workflow_streams). Activity-side nodes publish via `WorkflowStreamClient` (a signal carrying batched items, controlled by `streaming_batch_interval`); workflow-side nodes publish synchronously to the in-workflow stream (no signal). External subscribers consume the stream with `WorkflowStreamClient.create(...).topic(...).subscribe(...)`.

temporalio/contrib/langgraph/_activity.py

@@ -109,6 +109,7 @@ def thread_safe_writer(value: Any) -> None:
 def wrap_execute_activity(
     afunc: Callable[[ActivityInput], Awaitable[ActivityOutput]],
     task_id: str = "",
+    summary_fn: Callable[[tuple[Any, ...], dict[str, Any]], str | None] | None = None,
     **execute_activity_kwargs: Any,
 ) -> Callable[..., Any]:
     """Wrap an activity function to be called via workflow.execute_activity with caching."""
@@ -156,9 +157,15 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
         input = ActivityInput(
             args=args, kwargs=kwargs, langgraph_config=langgraph_config
         )
-        output = await workflow.execute_activity(
-            afunc, input, **execute_activity_kwargs
-        )
+        # Compute a dynamic activity summary (if configured) on the schedule
+        # path only; a cache hit above returns before reaching here, so no
+        # activity is scheduled and no summary is needed.
+        call_kwargs = dict(execute_activity_kwargs)
+        if summary_fn is not None:
+            summary = summary_fn(args, kwargs)
+            if summary:
+                call_kwargs["summary"] = summary
+        output = await workflow.execute_activity(afunc, input, **call_kwargs)
         if output.langgraph_interrupts is not None:
             raise GraphInterrupt(output.langgraph_interrupts)
 

temporalio/contrib/langgraph/_plugin.py

@@ -35,6 +35,36 @@
 _ACTIVITY_OPTION_KEYS: frozenset[str] = frozenset(
     {"execute_in", *inspect.signature(workflow.execute_activity).parameters}
 )
+# Node/task option keys beyond the raw execute_activity parameters:
+# 'summary_fn' is a callable consumed in the workflow (not a Temporal
+# option), so it must be split out of Graph API metadata too.
+_LANGGRAPH_OPTION_KEYS: frozenset[str] = _ACTIVITY_OPTION_KEYS | frozenset(
+    {"summary_fn"}
+)
+
+
+def _constant_summary_fn(
+    value: str,
+) -> Callable[[tuple[Any, ...], dict[str, Any]], str]:
+    """Adapt a static summary string to the summary_fn interface."""
+    return lambda args, kwargs: value
+
+
+def _merge_activity_opts(
+    defaults: dict[str, Any] | None, specific: dict[str, Any]
+) -> dict[str, Any]:
+    """Layer per-node/task options over the plugin defaults.
+
+    ``summary`` and ``summary_fn`` are two forms of one setting, so a node or
+    task that supplies either form overrides an inherited default of *either*
+    form, rather than coexisting with it and tripping the exclusivity check.
+    """
+    merged = dict(defaults or {})
+    if "summary" in specific or "summary_fn" in specific:
+        merged.pop("summary", None)
+        merged.pop("summary_fn", None)
+    merged.update(specific)
+    return merged
 
 
 class LangGraphPlugin(SimplePlugin):
@@ -69,7 +99,9 @@ class LangGraphPlugin(SimplePlugin):
             Functional API has no per-task ``metadata`` channel.
         default_activity_options: Activity options applied to every
             activity-bound node and task, overridable per-node (Graph API
-            ``metadata``) or per-task (``activity_options[name]``).
+            ``metadata``) or per-task (``activity_options[name]``). A
+            node/task that sets ``summary`` or ``summary_fn`` overrides an
+            inherited default of either form.
         streaming_topic: When set, ``langgraph.config.get_stream_writer()``
             inside a node publishes to this topic on the workflow's
             :class:`WorkflowStream`. The workflow must construct
@@ -130,6 +162,16 @@ def __init__(
                 "activity_options[task_name] (Functional API)."
             )
 
+        if (
+            default_activity_options
+            and "summary" in default_activity_options
+            and "summary_fn" in default_activity_options
+        ):
+            raise ValueError(
+                "Set either 'summary' or 'summary_fn' in default_activity_options, "
+                "not both."
+            )
+
         self.activities: list = []
         self._streaming_topic = streaming_topic
         self._streaming_batch_interval = streaming_batch_interval
@@ -168,20 +210,22 @@ def __init__(
                     # the node function via config["metadata"].
                     node_meta = node.metadata or {}
                     node_opts = {
-                        k: v for k, v in node_meta.items() if k in _ACTIVITY_OPTION_KEYS
+                        k: v
+                        for k, v in node_meta.items()
+                        if k in _LANGGRAPH_OPTION_KEYS
                     }
                     node.metadata = {
                         k: v
                         for k, v in node_meta.items()
-                        if k not in _ACTIVITY_OPTION_KEYS
+                        if k not in _LANGGRAPH_OPTION_KEYS
                     }
                     if "execute_in" not in node_opts:
                         raise ValueError(
                             f"Node {graph_name}.{node_name} is missing required "
                             f"'execute_in' in metadata. Set it to 'activity' or "
                             f"'workflow'."
                         )
-                    opts = {**(default_activity_options or {}), **node_opts}
+                    opts = _merge_activity_opts(default_activity_options, node_opts)
                     # Route all LangGraph node calls through afunc so the async
                     # activity wrapper is always used. wrap_activity handles
                     # sync vs. async user functions inside the activity itself.
@@ -208,10 +252,7 @@ def __init__(
                         f"activity_options[{name!r}]. Set it to 'activity' or "
                         f"'workflow'."
                     )
-                opts = {
-                    **(default_activity_options or {}),
-                    **task_opts,
-                }
+                opts = _merge_activity_opts(default_activity_options, task_opts)
 
                 task.func = self.execute(task_id(task.func), task.func, opts)
                 task.func.__name__ = name
@@ -253,6 +294,18 @@ def execute(
         """Prepare a node or task to execute as an activity or inline in the workflow."""
         opts = kwargs or {}
         execute_in = opts.pop("execute_in")
+        # Normalize the node's summary to a single summary_fn. Both keys are
+        # popped so neither reaches workflow.execute_activity (which takes no
+        # summary_fn, and would get a duplicate summary kwarg); a static
+        # summary becomes a summary_fn that ignores its input.
+        summary_fn = opts.pop("summary_fn", None)
+        static_summary = opts.pop("summary", None)
+        if summary_fn is not None and static_summary is not None:
+            raise ValueError(
+                f"{activity_name}: set either 'summary' or 'summary_fn', not both."
+            )
+        if static_summary is not None:
+            summary_fn = _constant_summary_fn(static_summary)
 
         if execute_in == "activity":
             wrapped = wrap_activity(
@@ -262,9 +315,13 @@ def execute(
             )
             a = activity.defn(name=activity_name)(wrapped)
             self.activities.append(a)
-            return wrap_execute_activity(a, task_id=task_id(func), **opts)
+            return wrap_execute_activity(
+                a, task_id=task_id(func), summary_fn=summary_fn, **opts
+            )
         elif execute_in == "workflow":
-            return wrap_workflow(func, streaming_topic=self._streaming_topic)
+            return wrap_workflow(
+                func, streaming_topic=self._streaming_topic, summary_fn=summary_fn
+            )
         else:
             raise ValueError(f"Invalid execute_in value: {execute_in}")
 

temporalio/contrib/langgraph/_workflow.py

@@ -20,6 +20,7 @@ def wrap_workflow(
     func: Callable[..., Any],
     *,
     streaming_topic: str | None = None,
+    summary_fn: Callable[[tuple[Any, ...], dict[str, Any]], str | None] | None = None,
 ) -> Callable[..., Awaitable[Any]]:
     """Wrap a function as a workflow-side LangGraph node.
 
@@ -28,9 +29,19 @@ def wrap_workflow(
     function with the writer installed. Workflow-side nodes publish
     synchronously to the in-workflow ``WorkflowStream`` (no signal
     round-trip); activity-side nodes go through ``WorkflowStreamClient``.
+
+    Workflow-side nodes have no activity to carry a summary, so a
+    ``summary_fn`` result updates the workflow's current details via
+    :func:`temporalio.workflow.set_current_details` (last-writer-wins);
+    an empty result clears it.
     """
 
     async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        if summary_fn is not None:
+            # Always write (clearing when empty) so this node never shows a
+            # stale summary left by an earlier workflow-bound node.
+            workflow.set_current_details(summary_fn(args, kwargs) or "")
+
         async def run(stream_writer: Callable[[Any], None] | None) -> Any:
             token = None
             if stream_writer is not None: