Skip to content

Commit c47ad81

Browse files
chris-colinskychris-colinsky
committed
chore(examples): tighten 05/07/08 from PR-review feedback
- 05 fan-out-with-retry: _timings.clear() at the top of main() so a REPL or repeated-main() driver doesn't accumulate timings across invocations. Module-level retention was an oversight. - 07 multimodal-prompt: replace the two type/shape asserts in caption() with an explicit isinstance check that raises RuntimeError. Asserts strip under python -O; the new shape narrows for pyright AND fails loudly if PromptManager's return contract drifts. - 08 checkpointing-and-migration: switch main() from tempfile.mkdtemp to tempfile.TemporaryDirectory wrapping the phase 1/2 logic. The SQLite DB + temp folder are now cleaned up on both the happy path and any raised exception, instead of leaving /tmp/oa-checkpoint-demo-* behind across runs.
1 parent 398c648 commit c47ad81

3 files changed

Lines changed: 88 additions & 78 deletions

File tree

‎examples/05-fan-out-with-retry/main.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,10 @@ def build_graph() -> CompiledGraph[BatchState]:
260260

261261

262262
async def main() -> None:
263+
# Reset module-level capture so a REPL or repeated-main() driver
264+
# doesn't accumulate timings across invocations.
265+
_timings.clear()
266+
263267
graph = build_graph()
264268

265269
initial = BatchState(headlines=HEADLINES)

‎examples/07-multimodal-prompt/main.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -138,9 +138,13 @@ async def caption(s: CaptionState) -> Mapping[str, Any]:
138138
# UserMessage. Pull out the text and compose a multimodal user
139139
# message that also carries the image.
140140
rendered_msg = rendered.messages[0]
141-
assert isinstance(rendered_msg, UserMessage)
141+
if not isinstance(rendered_msg, UserMessage) or not isinstance(rendered_msg.content, str):
142+
raise RuntimeError(
143+
"PromptManager.render() returned an unexpected shape; expected a single "
144+
f"UserMessage with str content, got {type(rendered_msg).__name__} "
145+
f"with content type {type(rendered_msg.content).__name__}"
146+
)
142147
rendered_text = rendered_msg.content
143-
assert isinstance(rendered_text, str)
144148

145149
multimodal_message = UserMessage(
146150
content=[

‎examples/08-checkpointing-and-migration/main.py‎

Lines changed: 78 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -257,82 +257,84 @@ async def main() -> None:
257257
destination = "Lunar South Pole"
258258

259259
# SQLite checkpointer in JSON mode (the migration-eligible
260-
# serialization). A real app would point at a persistent path; for
261-
# the demo a temp file keeps the workspace clean.
262-
db_dir = tempfile.mkdtemp(prefix="oa-checkpoint-demo-")
263-
db_path = Path(db_dir) / "checkpoints.sqlite"
264-
checkpointer = SQLiteCheckpointer(path=db_path, serialization="json")
265-
266-
print("=" * 72)
267-
print("Phase 1 — invoke v1 graph; checkpoints save after every node")
268-
print("=" * 72)
269-
print()
270-
print(f" destination: {destination}")
271-
print(f" checkpoint db: {db_path}")
272-
print()
273-
274-
# Pass a deterministic correlation_id so phase 2 can find the
275-
# invocation's saved records via the checkpoint filter. Without a
276-
# caller-supplied correlation_id, invoke() generates a UUIDv4.
277-
run_id = "demo-mission-plan-1"
278-
279-
graph_v1 = build_graph_v1(checkpointer)
280-
initial_v1 = MissionPlanStateV1(destination=destination)
281-
final_v1 = await graph_v1.invoke(initial_v1, correlation_id=run_id)
282-
await graph_v1.drain()
283-
284-
# Look up the saved record's invocation_id by correlation_id. The
285-
# invocation_id is generated by invoke() and isn't exposed on the
286-
# returned state; finding it through the checkpointer's list API
287-
# is the canonical lookup path.
288-
summaries = list(await checkpointer.list(CheckpointFilter(correlation_id=run_id)))
289-
assert summaries, "expected at least one saved checkpoint"
290-
invocation_id = summaries[-1].invocation_id
291-
292-
print("v1 result:")
293-
print(f" objective: {final_v1.objective}")
294-
print(f" crew_size: {final_v1.crew_size}")
295-
print(f" timeline: {final_v1.timeline}")
296-
print()
297-
print(f" v1 invocation_id: {invocation_id}")
298-
print()
299-
300-
print("=" * 72)
301-
print("Phase 2 — invoke v2 graph with resume; v1->v2 migration runs")
302-
print("=" * 72)
303-
print()
304-
print(" v2 adds: risk_assessment field + assess_risks node")
305-
print(" migration: backfills risk_assessment='' for v1 records")
306-
print()
307-
308-
graph_v2 = build_graph(checkpointer)
309-
# Resume from the v1 invocation. The engine reads the saved record,
310-
# applies migrate_v1_to_v2, re-deserializes against
311-
# MissionPlanStateV2, and continues at the first uncompleted node
312-
# (assess_risks — the v1 pipeline's three nodes are all in
313-
# completed_positions, the new v2 node is not).
314-
final_v2 = await graph_v2.invoke(
315-
MissionPlanStateV2(destination=destination),
316-
resume_invocation=invocation_id,
317-
)
318-
await graph_v2.drain()
319-
320-
print("v2 result after resume:")
321-
print(f" objective: {final_v2.objective}")
322-
print(f" crew_size: {final_v2.crew_size}")
323-
print(f" timeline: {final_v2.timeline}")
324-
print(f" risk_assessment: {final_v2.risk_assessment}")
325-
print()
326-
print(f" trace: {final_v2.trace}")
327-
print()
328-
print(
329-
"The v1 nodes appear once each in v1's trace and NOT in v2's "
330-
"trace — they were skipped on resume because completed_positions "
331-
"already covered them. Only assess_risks ran in phase 2."
332-
)
333-
334-
if _provider_instance is not None:
335-
await _provider_instance.aclose()
260+
# serialization). A real app would point at a persistent path; the
261+
# demo uses TemporaryDirectory so the DB file + folder get cleaned
262+
# up on exit (happy path or exception) without leaving cruft in /tmp.
263+
with tempfile.TemporaryDirectory(prefix="oa-checkpoint-demo-") as db_dir:
264+
db_path = Path(db_dir) / "checkpoints.sqlite"
265+
checkpointer = SQLiteCheckpointer(path=db_path, serialization="json")
266+
267+
try:
268+
print("=" * 72)
269+
print("Phase 1 — invoke v1 graph; checkpoints save after every node")
270+
print("=" * 72)
271+
print()
272+
print(f" destination: {destination}")
273+
print(f" checkpoint db: {db_path}")
274+
print()
275+
276+
# Pass a deterministic correlation_id so phase 2 can find the
277+
# invocation's saved records via the checkpoint filter. Without a
278+
# caller-supplied correlation_id, invoke() generates a UUIDv4.
279+
run_id = "demo-mission-plan-1"
280+
281+
graph_v1 = build_graph_v1(checkpointer)
282+
initial_v1 = MissionPlanStateV1(destination=destination)
283+
final_v1 = await graph_v1.invoke(initial_v1, correlation_id=run_id)
284+
await graph_v1.drain()
285+
286+
# Look up the saved record's invocation_id by correlation_id. The
287+
# invocation_id is generated by invoke() and isn't exposed on the
288+
# returned state; finding it through the checkpointer's list API
289+
# is the canonical lookup path.
290+
summaries = list(await checkpointer.list(CheckpointFilter(correlation_id=run_id)))
291+
assert summaries, "expected at least one saved checkpoint"
292+
invocation_id = summaries[-1].invocation_id
293+
294+
print("v1 result:")
295+
print(f" objective: {final_v1.objective}")
296+
print(f" crew_size: {final_v1.crew_size}")
297+
print(f" timeline: {final_v1.timeline}")
298+
print()
299+
print(f" v1 invocation_id: {invocation_id}")
300+
print()
301+
302+
print("=" * 72)
303+
print("Phase 2 — invoke v2 graph with resume; v1->v2 migration runs")
304+
print("=" * 72)
305+
print()
306+
print(" v2 adds: risk_assessment field + assess_risks node")
307+
print(" migration: backfills risk_assessment='' for v1 records")
308+
print()
309+
310+
graph_v2 = build_graph(checkpointer)
311+
# Resume from the v1 invocation. The engine reads the saved record,
312+
# applies migrate_v1_to_v2, re-deserializes against
313+
# MissionPlanStateV2, and continues at the first uncompleted node
314+
# (assess_risks — the v1 pipeline's three nodes are all in
315+
# completed_positions, the new v2 node is not).
316+
final_v2 = await graph_v2.invoke(
317+
MissionPlanStateV2(destination=destination),
318+
resume_invocation=invocation_id,
319+
)
320+
await graph_v2.drain()
321+
322+
print("v2 result after resume:")
323+
print(f" objective: {final_v2.objective}")
324+
print(f" crew_size: {final_v2.crew_size}")
325+
print(f" timeline: {final_v2.timeline}")
326+
print(f" risk_assessment: {final_v2.risk_assessment}")
327+
print()
328+
print(f" trace: {final_v2.trace}")
329+
print()
330+
print(
331+
"The v1 nodes appear once each in v1's trace and NOT in v2's "
332+
"trace — they were skipped on resume because completed_positions "
333+
"already covered them. Only assess_risks ran in phase 2."
334+
)
335+
finally:
336+
if _provider_instance is not None:
337+
await _provider_instance.aclose()
336338

337339

338340
if __name__ == "__main__":

0 commit comments

Comments
 (0)