Plan Visualization#
LangGOAP ships built-in renderers that turn a Plan into Mermaid, Graphviz DOT,
and plain ASCII representations. Every renderer is pure-Python and dependency-free:
the Mermaid output renders inline in Jupyter; DOT can be piped to Graphviz; ASCII is
perfect for logs and terminals.
When a plan carries CSP metadata (from the A* → CSP pipeline), the renderers additionally show parallel clusters, resource usage, schedule Gantt charts and constraint satisfaction indicators.
Setup#
Define a three-step linear pipeline and plan a goal against it.
from datetime import timedelta
from langgoap import (
ActionSpec,
ConstraintSpec,
GoalSpec,
)
from langgoap.planner.pipeline import plan as pipeline_plan
actions = [
ActionSpec(
name="collect_specs",
effects={"specs_collected": True},
resources={"tokens": 100.0, "cost_usd": 0.10},
),
ActionSpec(
name="design_layout",
preconditions={"specs_collected": True},
effects={"layout_designed": True},
resources={"tokens": 200.0, "cost_usd": 0.20},
),
ActionSpec(
name="compile_report",
preconditions={"layout_designed": True},
effects={"report_ready": True},
resources={"tokens": 150.0, "cost_usd": 0.15},
),
]
start = {}
goal = GoalSpec(
conditions={"report_ready": True},
constraints=(
ConstraintSpec(key="tokens", max=1000.0),
ConstraintSpec(key="cost_usd", max=1.0),
),
)
plan = pipeline_plan(start, goal, actions)
print(plan)
Mermaid flowchart#
Plan.visualize() auto-detects IPython and returns an IPython.display.Markdown
that Jupyter renders as a Mermaid diagram. Dependency arrows come from the
same effect→precondition graph the scheduler uses.
plan.visualize(format="mermaid")
flowchart TD
classDef action fill:#f8f9fa,stroke:#4a90d9,stroke-width:2px,color:#1a1a2e
classDef cost fill:#d4edda,stroke:#2d8a4e,stroke-width:1px,color:#155724,font-size:10px
a0_collect_specs["collect_specs"]:::action
a1_design_layout["design_layout"]:::action
a2_compile_report["compile_report"]:::action
a0_collect_specs --> a1_design_layout
a1_design_layout --> a2_compile_report
%% Resource usage
%% tokens: 450 / 1000 [OK]
%% cost_usd: 0.45 / 1 [OK]
ASCII tree#
For logs and terminals, Plan.to_ascii() returns a plain-text tree with
per-action dependency indices and a resource usage summary.
print(plan.to_ascii())
Plan (3 steps, cost=3)
├── [0] collect_specs
├── [1] design_layout (after: [0])
├── [2] compile_report (after: [1])
Resources:
tokens: 450 / 1000 [OK]
cost_usd: 0.45 / 1 [OK]
Graphviz DOT#
Plan.to_dot() returns Graphviz source code. Pipe it to the dot binary
for high-quality renders, or pass it to the graphviz Python package.
print(plan.to_dot())
digraph Plan {
rankdir=TB;
node [shape=box, style="rounded,filled", fillcolor="#e8f4ff"];
a0_collect_specs [label="collect_specs"];
a1_design_layout [label="design_layout"];
a2_compile_report [label="compile_report"];
a0_collect_specs -> a1_design_layout;
a1_design_layout -> a2_compile_report;
legend [shape=note, fillcolor="#fff8dc", label="Resources:\ltokens: 450 / 1000 [OK]\lcost_usd: 0.45 / 1 [OK]\l"];
}
Scheduled plan with parallelism#
Adding duration to actions triggers the CSP temporal scheduler.
Actions with no mutual dependencies are scheduled in parallel —
renderers group them visually.
parallel_actions = [
ActionSpec(
name="fetch_source_a",
effects={"source_a": True},
duration=timedelta(seconds=2),
resources={"cost_usd": 0.10},
),
ActionSpec(
name="fetch_source_b",
effects={"source_b": True},
duration=timedelta(seconds=3),
resources={"cost_usd": 0.15},
),
ActionSpec(
name="merge_sources",
preconditions={"source_a": True, "source_b": True},
effects={"merged": True},
duration=timedelta(seconds=1),
resources={"cost_usd": 0.05},
),
]
scheduled_plan = pipeline_plan(
{},
GoalSpec(
conditions={"merged": True},
constraints=(ConstraintSpec(key="cost_usd", max=1.0),),
),
parallel_actions,
)
assert scheduled_plan is not None
print(f"Makespan: {scheduled_plan.metadata.csp.makespan}")
scheduled_plan.visualize(format="mermaid")
Gantt chart#
format="gantt" renders a Mermaid gantt from the CSP schedule.
format="ascii_gantt" gives the same visualization as plain text.
scheduled_plan.visualize(format="gantt")
gantt
title Plan Schedule
dateFormat x
axisFormat %Lms
section Plan
fetch_source_a :task_0_2000_fetch_source_a, 0, 2000ms
fetch_source_b :task_0_3000_fetch_source_b, 0, 3000ms
merge_sources :task_3000_1000_merge_sources, 3000, 1000ms
print(scheduled_plan.visualize(format="ascii_gantt"))
action |------------------------------------------------------------| 0..4s
fetch_source_a |############################## | 0-2s
fetch_source_b |############################################# | 0-3s
merge_sources | ###############| 3-4s
Constraint violation reporting#
When a plan violates its budget, renderers mark the violated resource
as VIOLATED so it stands out in the diagram.
tight_goal = GoalSpec(
conditions={"report_ready": True},
constraints=(ConstraintSpec(key="tokens", max=100.0),), # tight
)
tight_plan = pipeline_plan(start, tight_goal, actions)
print(tight_plan.to_ascii())
No alternative plans found
Plan (3 steps, cost=3)
├── [0] collect_specs
├── [1] design_layout (after: [0])
├── [2] compile_report (after: [1])
Resources:
tokens: 450 / 100 [VIOLATED]
cost_usd: 0.45 [OK]
Infeasibility Explanation:
Conflicting constraints:
- tokens (max=100, weight=1)
Resource shortfalls:
- tokens: 450 / 100 (overrun: 350)
Suggestion: Resource 'tokens' uses 450 but the hard limit is 100 (overrun: 350). Consider relaxing the constraint, reducing action costs, or removing expensive actions.
Saving to disk#
Plan.save() writes a rendered representation to a file. The format
is inferred from the extension: .mmd → Mermaid, .dot → DOT,
anything else → ASCII. Pass format= to override.
import tempfile
from pathlib import Path
tmp = Path(tempfile.mkdtemp())
plan.save(tmp / "plan.mmd")
plan.save(tmp / "plan.dot")
plan.save(tmp / "plan.txt")
print(sorted(p.name for p in tmp.iterdir()))
['plan.dot', 'plan.mmd', 'plan.txt']