operon/examples/106_workflow_generation_interface.py at v0.25.1 · coredipper/operon · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
"""
Example 106 -- Workflow Generation Interface (Phase C7)
========================================================

Demonstrates the WorkflowGenerator protocol: create a HeuristicGenerator,
generate a template from natural language, register it in a PatternLibrary,
and call refine().

Usage:
    python examples/106_workflow_generation_interface.py
"""

from operon_ai.convergence.workflow_generation import (
    HeuristicGenerator,
    WorkflowGenerator,
    generate_and_register,
)
from operon_ai.patterns.repository import PatternLibrary

# ---------------------------------------------------------------------------
# 1. Create HeuristicGenerator and verify Protocol conformance
# ---------------------------------------------------------------------------

gen = HeuristicGenerator()
print("=== Protocol conformance ===")
print(f"  isinstance(gen, WorkflowGenerator): {isinstance(gen, WorkflowGenerator)}")
assert isinstance(gen, WorkflowGenerator)

# ---------------------------------------------------------------------------
# 2. Generate template from natural language
# ---------------------------------------------------------------------------

task = "Summarize a PDF document, extract key findings, and produce a slide deck."
template = gen.generate(task)

print(f"\n=== Generated template ===")
print(f"  template_id: {template.template_id}")
print(f"  name:        {template.name}")
print(f"  topology:    {template.topology}")
print(f"  stages:      {len(template.stage_specs)}")
for spec in template.stage_specs:
    print(f"    - {spec['name']} ({spec['mode']})")
print(f"  fingerprint: shape={template.fingerprint.task_shape}, "
      f"subtasks={template.fingerprint.subtask_count}, "
      f"roles={template.fingerprint.required_roles}")

assert len(template.stage_specs) > 0, "Template must have stages"
assert template.topology == "skill_organism"

# ---------------------------------------------------------------------------
# 3. Register in PatternLibrary via generate_and_register()
# ---------------------------------------------------------------------------

library = PatternLibrary()
registered = generate_and_register(gen, task, library)

print(f"\n=== Library after generate_and_register ===")
summary = library.summary()
print(f"  template_count: {summary['template_count']}")
print(f"  topologies:     {summary['topologies']}")
assert summary["template_count"] == 1
assert library.get_template(registered.template_id) is not None

# ---------------------------------------------------------------------------
# 4. Verify library contains the template
# ---------------------------------------------------------------------------

retrieved = library.retrieve_templates(topology="skill_organism")
print(f"\n=== retrieve_templates(topology='skill_organism') ===")
print(f"  found: {len(retrieved)} template(s)")
assert len(retrieved) == 1
assert retrieved[0].template_id == registered.template_id

# ---------------------------------------------------------------------------
# 5. Call refine() -- reference impl returns unchanged
# ---------------------------------------------------------------------------

refined = gen.refine(template, feedback="Add an error-handling step.")
print(f"\n=== refine() ===")
print(f"  same object: {refined is template}")
assert refined is template, "HeuristicGenerator.refine() should return template unchanged"

print("\n--- all assertions passed ---")