WIP

Author: dmontagu

docs/builtin-tools.md

@@ -87,7 +87,7 @@ agent = Agent(
 )
 
 result = agent.run_sync('Use the web to get the current time.')
-# > In San Francisco, it's 8:21:41 pm PDT on Wednesday, August 6, 2025.
+#> In San Francisco, it's 8:21:41 pm PDT on Wednesday, August 6, 2025.
 ```
 
 ### Parameter Support by Provider
@@ -129,7 +129,7 @@ from pydantic_ai import Agent, CodeExecutionTool
 agent = Agent('anthropic:claude-sonnet-4-0', builtin_tools=[CodeExecutionTool()])
 
 result = agent.run_sync('Calculate the factorial of 15 and show your work')
-# > The factorial of 15 is **1,307,674,368,000**.
+#> The factorial of 15 is **1,307,674,368,000**.
 ```
 
 ## URL Context Tool
@@ -158,7 +158,7 @@ from pydantic_ai import Agent, UrlContextTool
 agent = Agent('google-gla:gemini-2.5-flash', builtin_tools=[UrlContextTool()])
 
 result = agent.run_sync('What is this? https://ai.pydantic.dev')
-# > A Python agent framework for building Generative AI applications.
+#> A Python agent framework for building Generative AI applications.
 ```
 
 ## Memory Tool

docs/graph/beta/index.md

@@ -123,7 +123,7 @@ Here's an example showcasing parallel execution with a map operation:
 ```python {title="parallel_processing.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -147,7 +147,7 @@ async def main():
         return ctx.inputs * ctx.inputs
 
     # Create a join to collect results
-    collect_results = g.join(ListReducer[int])
+    collect_results = g.join(ListAppendReducer[int])
 
     # Build the graph with map operation
     g.add(
@@ -172,7 +172,7 @@ In this example:
 
 1. The start node receives a list of integers
 2. The `.map()` operation fans out each item to a separate parallel execution of the `square` step
-3. All results are collected back together using a [`ListReducer`][pydantic_graph.beta.join.ListReducer]
+3. All results are collected back together using a [`ListAppendReducer`][pydantic_graph.beta.join.ListAppendReducer]
 4. The joined results flow to the end node
 
 ## Next Steps

docs/graph/beta/joins.md

@@ -17,7 +17,7 @@ Create a join using [`g.join()`][pydantic_graph.beta.graph_builder.GraphBuilder.
 ```python {title="basic_join.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -36,7 +36,7 @@ async def square(ctx: StepContext[SimpleState, None, int]) -> int:
     return ctx.inputs * ctx.inputs
 
 # Create a join to collect all squared values
-collect = g.join(ListReducer[int])
+collect = g.join(ListAppendReducer[int])
 
 g.add(
     g.edge_from(g.start_node).to(generate_numbers),
@@ -59,14 +59,14 @@ _(This example is complete, it can be run "as is" — you'll need to add `import
 
 Pydantic Graph provides several common reducer types out of the box:
 
-### ListReducer
+### ListAppendReducer
 
-[`ListReducer`][pydantic_graph.beta.join.ListReducer] collects all inputs into a list:
+[`ListAppendReducer`][pydantic_graph.beta.join.ListAppendReducer] collects all inputs into a list:
 
 ```python {title="list_reducer.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -85,7 +85,7 @@ async def main():
     async def to_string(ctx: StepContext[SimpleState, None, int]) -> str:
         return f'value-{ctx.inputs}'
 
-    collect = g.join(ListReducer[str])
+    collect = g.join(ListAppendReducer[str])
 
     g.add(
         g.edge_from(g.start_node).to(generate),
@@ -109,7 +109,7 @@ _(This example is complete, it can be run "as is" — you'll need to add `import
 ```python {title="dict_reducer.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import DictReducer, GraphBuilder, StepContext
+from pydantic_graph.beta import DictUpdateReducer, GraphBuilder, StepContext
 
 
 @dataclass
@@ -128,7 +128,7 @@ async def main():
     async def create_entry(ctx: StepContext[SimpleState, None, str]) -> dict[str, int]:
         return {ctx.inputs: len(ctx.inputs)}
 
-    merge = g.join(DictReducer[str, int])
+    merge = g.join(DictUpdateReducer[str, int])
 
     g.add(
         g.edge_from(g.start_node).to(generate_keys),
@@ -339,7 +339,7 @@ A graph can have multiple independent joins:
 ```python {title="multiple_joins.py"}
 from dataclasses import dataclass, field
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -366,8 +366,8 @@ async def main():
     async def process_b(ctx: StepContext[MultiState, None, int]) -> int:
         return ctx.inputs * 3
 
-    join_a = g.join(ListReducer[int], node_id='join_a')
-    join_b = g.join(ListReducer[int], node_id='join_b')
+    join_a = g.join(ListAppendReducer[int], node_id='join_a')
+    join_b = g.join(ListAppendReducer[int], node_id='join_b')
 
     @g.step
     async def store_a(ctx: StepContext[MultiState, None, list[int]]) -> None:
@@ -412,9 +412,11 @@ _(This example is complete, it can be run "as is" — you'll need to add `import
 Like steps, joins can have custom IDs:
 
 ```python {title="join_custom_id.py" requires="basic_join.py"}
-from basic_join import ListReducer, g
+from pydantic_graph.beta import ListAppendReducer
 
-my_join = g.join(ListReducer[int], node_id='my_custom_join_id')
+from basic_join import g
+
+my_join = g.join(ListAppendReducer[int], node_id='my_custom_join_id')
 ```
 
 ## How Joins Work

docs/graph/beta/parallel.md

@@ -16,7 +16,7 @@ Broadcasting sends identical data to multiple destinations simultaneously:
 ```python {title="basic_broadcast.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -43,7 +43,7 @@ async def main():
     async def add_three(ctx: StepContext[SimpleState, None, int]) -> int:
         return ctx.inputs + 3
 
-    collect = g.join(ListReducer[int])
+    collect = g.join(ListAppendReducer[int])
 
     # Broadcasting: send the value from source to all three steps
     g.add(
@@ -70,7 +70,7 @@ Spreading fans out elements from an iterable, processing each element in paralle
 ```python {title="basic_map.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -89,7 +89,7 @@ async def main():
     async def square(ctx: StepContext[SimpleState, None, int]) -> int:
         return ctx.inputs * ctx.inputs
 
-    collect = g.join(ListReducer[int])
+    collect = g.join(ListAppendReducer[int])
 
     # Spreading: each item in the list gets its own parallel execution
     g.add(
@@ -114,7 +114,7 @@ The convenience method [`add_mapping_edge()`][pydantic_graph.beta.graph_builder.
 ```python {title="mapping_convenience.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext, Reducer
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -133,7 +133,7 @@ async def main():
     async def stringify(ctx: StepContext[SimpleState, None, int]) -> str:
         return f'Value: {ctx.inputs}'
 
-    collect = g.join(ListReducer[str])
+    collect = g.join(ListAppendReducer[str])
 
     g.add(g.edge_from(g.start_node).to(generate_numbers))
     g.add_mapping_edge(generate_numbers, stringify)
@@ -157,7 +157,7 @@ When mapping an empty iterable, you can specify a `downstream_join_id` to ensure
 ```python {title="empty_map.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -176,7 +176,7 @@ async def main():
     async def double(ctx: StepContext[SimpleState, None, int]) -> int:
         return ctx.inputs * 2
 
-    collect = g.join(ListReducer[int])
+    collect = g.join(ListAppendReducer[int])
 
     g.add(g.edge_from(g.start_node).to(generate_empty))
     g.add_mapping_edge(generate_empty, double, downstream_join_id=collect.id)
@@ -202,7 +202,7 @@ You can nest broadcasts and maps for complex parallel patterns:
 ```python {title="map_then_broadcast.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -225,7 +225,7 @@ async def main():
     async def add_two(ctx: StepContext[SimpleState, None, int]) -> int:
         return ctx.inputs + 2
 
-    collect = g.join(ListReducer[int])
+    collect = g.join(ListAppendReducer[int])
 
     g.add(
         g.edge_from(g.start_node).to(generate_list),
@@ -252,7 +252,7 @@ The result contains:
 ```python {title="sequential_maps.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -275,7 +275,7 @@ async def main():
     async def stringify(ctx: StepContext[SimpleState, None, int]) -> str:
         return f'num:{ctx.inputs}'
 
-    collect = g.join(ListReducer[str])
+    collect = g.join(ListAppendReducer[str])
 
     g.add(
         g.edge_from(g.start_node).to(generate_pairs),
@@ -302,7 +302,7 @@ Add labels to parallel edges for better documentation:
 ```python {title="labeled_parallel.py"}
 from dataclasses import dataclass
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -321,7 +321,7 @@ async def main():
     async def process(ctx: StepContext[SimpleState, None, int]) -> str:
         return f'item-{ctx.inputs}'
 
-    collect = g.join(ListReducer[str])
+    collect = g.join(ListAppendReducer[str])
 
     g.add(g.edge_from(g.start_node).to(generate))
     g.add_mapping_edge(
@@ -350,7 +350,7 @@ All parallel tasks share the same graph state. Be careful with mutations:
 ```python {title="parallel_state.py"}
 from dataclasses import dataclass, field
 
-from pydantic_graph.beta import GraphBuilder, ListReducer, StepContext
+from pydantic_graph.beta import GraphBuilder, ListAppendReducer, StepContext
 
 
 @dataclass
@@ -371,7 +371,7 @@ async def main():
         ctx.state.values.append(ctx.inputs)
         return ctx.inputs * ctx.inputs
 
-    collect = g.join(ListReducer[int])
+    collect = g.join(ListAppendReducer[int])
 
     g.add(
         g.edge_from(g.start_node).to(generate),

docs/input.md

@@ -20,7 +20,7 @@ result = agent.run_sync(
     ]
 )
 print(result.output)
-# > This is the logo for Pydantic, a data validation and settings management library in Python.
+#> This is the logo for Pydantic, a data validation and settings management library in Python.
 ```
 
 If you have the image locally, you can also use [`BinaryContent`][pydantic_ai.BinaryContent]:
@@ -40,7 +40,7 @@ result = agent.run_sync(
     ]
 )
 print(result.output)
-# > This is the logo for Pydantic, a data validation and settings management library in Python.
+#> This is the logo for Pydantic, a data validation and settings management library in Python.
 ```
 
 1. To ensure the example is runnable we download this image from the web, but you can also use `Path().read_bytes()` to read a local file's contents.
@@ -79,7 +79,7 @@ result = agent.run_sync(
     ]
 )
 print(result.output)
-# > This document is the technical report introducing Gemini 1.5, Google's latest large language model...
+#> This document is the technical report introducing Gemini 1.5, Google's latest large language model...
 ```
 
 The supported document formats vary by model.
@@ -99,7 +99,7 @@ result = agent.run_sync(
     ]
 )
 print(result.output)
-# > The document discusses...
+#> The document discusses...
 ```
 
 ## User-side download vs. direct file URL

pydantic_graph/pydantic_graph/beta/__init__.py

@@ -10,17 +10,17 @@
 
 from .graph import Graph
 from .graph_builder import GraphBuilder
-from .join import DictReducer, ListReducer, NullReducer, Reducer
+from .join import DictUpdateReducer, ListAppendReducer, NullReducer, Reducer
 from .node import EndNode, StartNode
 from .step import StepContext, StepNode
 from .util import TypeExpression
 
 __all__ = (
-    'DictReducer',
+    'DictUpdateReducer',
     'EndNode',
     'Graph',
     'GraphBuilder',
-    'ListReducer',
+    'ListAppendReducer',
     'NullReducer',
     'Reducer',
     'StartNode',

pydantic_graph/pydantic_graph/beta/graph.py

@@ -251,6 +251,11 @@ def render(self, *, title: str | None = None, direction: StateDiagramDirection |
         return build_mermaid_graph(self).render(title=title, direction=direction)
 
     def __repr__(self) -> str:
+        super_repr = super().__repr__()  # include class and memory address
+        # Insert the result of calling `__str__` before the final '>' in the repr
+        return f'{super_repr[:-1]}\n{self}\n{super_repr[-1]}'
+
+    def __str__(self) -> str:
         """Return a Mermaid diagram representation of the graph.
 
         Returns:

pydantic_graph/pydantic_graph/beta/graph_builder.py

@@ -58,6 +58,7 @@
 T = TypeVar('T', infer_variance=True)
 
 
+# TODO(P1): Should we make this method private? Not sure why it was public..
 @overload
 def join(
     *,
@@ -766,12 +767,13 @@ def _normalize_forks(
             if isinstance(item, MapMarker):
                 assert item.fork_id in new_nodes
                 new_edges[item.fork_id] = [path.next_path]
-            if isinstance(item, BroadcastMarker):
+                paths_to_handle.append(path.next_path)
+                break
+            elif isinstance(item, BroadcastMarker):
                 assert item.fork_id in new_nodes
-                # if item.fork_id not in new_nodes:
-                #     new_nodes[new_fork.id] = Fork[Any, Any](id=item.fork_id, is_map=False)
                 new_edges[item.fork_id] = [*item.paths]
                 paths_to_handle.extend(item.paths)
+                break
 
     return new_nodes, new_edges