feat: Add name and description support to workflows (#1183)

Add optional name and description fields to workflows in both Python and .NET implementations, matching the existing agent API pattern.

Python changes:
- Add name/description parameters to WorkflowBuilder.__init__
- Add name/description attributes to Workflow class
- Include name/description in to_dict() serialization
- Add WORKFLOW_NAME and WORKFLOW_DESCRIPTION OTEL attributes
- Add tests in test_serialization.py and test_workflow_observability.py

.NET changes:
- Add Name and Description properties to Workflow and Workflow<T>
- Add WithName() and WithDescription() fluent methods to WorkflowBuilder
- Add WorkflowName and WorkflowDescription OTEL tags
- Add test in WorkflowBuilderSmokeTests.cs

This enables applications like DevUI to display human-readable workflow names (e.g., 'Data Processing Pipeline') instead of auto-generated UUIDs (e.g., 'Workflow 50fdd917').

Fixes: #1181
This commit is contained in:
Victor Dibia
2025-10-04 08:20:17 -07:00
committed by GitHub
Unverified
parent fd819c6c02
commit 5a7ca13af6
8 changed files with 194 additions and 11 deletions
@@ -172,6 +172,8 @@ class Workflow(DictConvertible):
start_executor: Executor | str,
runner_context: RunnerContext,
max_iterations: int = DEFAULT_MAX_ITERATIONS,
name: str | None = None,
description: str | None = None,
**kwargs: Any,
):
"""Initialize the workflow with a list of edges.
@@ -182,6 +184,8 @@ class Workflow(DictConvertible):
start_executor: The starting executor for the workflow, which can be an Executor instance or its ID.
runner_context: The RunnerContext instance to be used during workflow execution.
max_iterations: The maximum number of iterations the workflow will run for convergence.
name: Optional human-readable name for the workflow.
description: Optional description of what the workflow does.
kwargs: Additional keyword arguments. Unused in this implementation.
"""
# Convert start_executor to string ID if it's an Executor instance
@@ -194,6 +198,8 @@ class Workflow(DictConvertible):
self.start_executor_id = start_executor_id
self.max_iterations = max_iterations
self.id = id
self.name = name
self.description = description
# Store non-serializable runtime objects as private attributes
self._runner_context = runner_context
@@ -236,6 +242,12 @@ class Workflow(DictConvertible):
"executors": {executor_id: executor.to_dict() for executor_id, executor in self.executors.items()},
}
# Add optional name and description if provided
if self.name is not None:
data["name"] = self.name
if self.description is not None:
data["description"] = self.description
executors_data: dict[str, dict[str, Any]] = data.get("executors", {})
for executor_id, executor_payload in executors_data.items():
if (
@@ -284,11 +296,15 @@ class Workflow(DictConvertible):
WorkflowEvent: The events generated during the workflow execution.
"""
# Create workflow span that encompasses the entire execution
attributes: dict[str, Any] = {OtelAttr.WORKFLOW_ID: self.id}
if self.name:
attributes[OtelAttr.WORKFLOW_NAME] = self.name
if self.description:
attributes[OtelAttr.WORKFLOW_DESCRIPTION] = self.description
with create_workflow_span(
OtelAttr.WORKFLOW_RUN_SPAN,
{
OtelAttr.WORKFLOW_ID: self.id,
},
attributes,
) as span:
saw_request = False
emitted_in_progress_pending = False
@@ -874,14 +890,27 @@ class WorkflowBuilder:
This class provides methods to add edges and set the starting executor for the workflow.
"""
def __init__(self, max_iterations: int = DEFAULT_MAX_ITERATIONS):
"""Initialize the WorkflowBuilder with an empty list of edges and no starting executor."""
def __init__(
self,
max_iterations: int = DEFAULT_MAX_ITERATIONS,
name: str | None = None,
description: str | None = None,
):
"""Initialize the WorkflowBuilder with an empty list of edges and no starting executor.
Args:
max_iterations: Maximum number of iterations for workflow convergence.
name: Optional human-readable name for the workflow.
description: Optional description of what the workflow does.
"""
self._edge_groups: list[EdgeGroup] = []
self._executors: dict[str, Executor] = {}
self._duplicate_executor_ids: set[str] = set()
self._start_executor: Executor | str | None = None
self._checkpoint_storage: CheckpointStorage | None = None
self._max_iterations: int = max_iterations
self._name: str | None = name
self._description: str | None = description
# Maps underlying AgentProtocol object id -> wrapped Executor so we reuse the same wrapper
# across set_start_executor / add_edge calls. Without this, unnamed agents (which receive
# random UUID based executor ids) end up wrapped multiple times, giving different ids for
@@ -1185,12 +1214,23 @@ class WorkflowBuilder:
# Create workflow instance after validation
workflow = Workflow(
self._edge_groups, self._executors, self._start_executor, context, self._max_iterations
self._edge_groups,
self._executors,
self._start_executor,
context,
self._max_iterations,
name=self._name,
description=self._description,
)
span.set_attributes({
build_attributes: dict[str, Any] = {
OtelAttr.WORKFLOW_ID: workflow.id,
OtelAttr.WORKFLOW_DEFINITION: workflow.to_json(),
})
}
if workflow.name:
build_attributes[OtelAttr.WORKFLOW_NAME] = workflow.name
if workflow.description:
build_attributes[OtelAttr.WORKFLOW_DESCRIPTION] = workflow.description
span.set_attributes(build_attributes)
# Add workflow build completed event
span.add_event(OtelAttr.BUILD_COMPLETED)
@@ -180,6 +180,8 @@ class OtelAttr(str, Enum):
# Workflow attributes
WORKFLOW_ID = "workflow.id"
WORKFLOW_NAME = "workflow.name"
WORKFLOW_DESCRIPTION = "workflow.description"
WORKFLOW_DEFINITION = "workflow.definition"
WORKFLOW_BUILD_SPAN = "workflow.build"
WORKFLOW_RUN_SPAN = "workflow.run"
@@ -599,6 +599,48 @@ class TestSerializationWorkflowClasses:
assert "_shared_state" not in data
assert "_runner" not in data
def test_workflow_name_description_serialization(self) -> None:
"""Test that workflow name and description are serialized correctly."""
# Test 1: With name and description
workflow1 = (
WorkflowBuilder(name="Test Pipeline", description="Test workflow description")
.set_start_executor(SampleExecutor(id="e1"))
.build()
)
assert workflow1.name == "Test Pipeline"
assert workflow1.description == "Test workflow description"
data1 = workflow1.to_dict()
assert data1["name"] == "Test Pipeline"
assert data1["description"] == "Test workflow description"
# Test JSON serialization
json_str1 = workflow1.to_json()
parsed1 = json.loads(json_str1)
assert parsed1["name"] == "Test Pipeline"
assert parsed1["description"] == "Test workflow description"
# Test 2: Without name and description (defaults)
workflow2 = WorkflowBuilder().set_start_executor(SampleExecutor(id="e2")).build()
assert workflow2.name is None
assert workflow2.description is None
data2 = workflow2.to_dict()
assert "name" not in data2 # Should not include None values
assert "description" not in data2
# Test 3: With only name (no description)
workflow3 = WorkflowBuilder(name="Named Only").set_start_executor(SampleExecutor(id="e3")).build()
assert workflow3.name == "Named Only"
assert workflow3.description is None
data3 = workflow3.to_dict()
assert data3["name"] == "Named Only"
assert "description" not in data3
def test_executor_field_validation(self) -> None:
"""Test that Executor field validation works correctly."""
# Valid executor
@@ -282,6 +282,23 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter)
assert "build.validation_completed" in build_event_names
assert "build.completed" in build_event_names
# Clear spans to test workflow with name and description
span_exporter.clear()
# Test workflow with name and description - verify OTEL attributes
(
WorkflowBuilder(name="Test Pipeline", description="Test workflow description")
.set_start_executor(MockExecutor("start"))
.build()
)
build_spans_with_metadata = [s for s in span_exporter.get_finished_spans() if s.name == "workflow.build"]
assert len(build_spans_with_metadata) == 1
metadata_build_span = build_spans_with_metadata[0]
assert metadata_build_span.attributes is not None
assert metadata_build_span.attributes.get(OtelAttr.WORKFLOW_NAME) == "Test Pipeline"
assert metadata_build_span.attributes.get(OtelAttr.WORKFLOW_DESCRIPTION) == "Test workflow description"
# Clear spans to separate build from run tracing
span_exporter.clear()