Spaces:
Sleeping
Sleeping
File size: 2,714 Bytes
c98ef88 e151643 c98ef88 8cef3d5 c98ef88 8cef3d5 c98ef88 8cef3d5 c98ef88 8cef3d5 |
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 84 85 |
"""
LangGraph Workflow Orchestrations
Updated for dictionary-based state and persistence.
"""
from typing import Dict, Any, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from state_schema import WorkflowState, create_initial_state
from agents import (
agent_0_node,
agent_1_node,
agent_2_node,
agent_3_integrated_node
)
# Global checkpointer to persist state between function calls
_global_checkpointer = MemorySaver()
def create_workflow():
"""Create the LangGraph workflow."""
global _global_checkpointer
workflow = StateGraph(WorkflowState)
# Add Nodes
workflow.add_node("setup", agent_0_node)
workflow.add_node("capture_design", agent_1_node)
workflow.add_node("capture_website", agent_2_node)
workflow.add_node("analysis", agent_3_integrated_node)
# Define Edges
workflow.set_entry_point("setup")
workflow.add_edge("setup", "capture_design")
workflow.add_edge("capture_design", "capture_website")
# Human-in-the-loop breakpoint before analysis
workflow.add_edge("capture_website", "analysis")
workflow.add_edge("analysis", END)
return workflow.compile(
checkpointer=_global_checkpointer,
interrupt_before=["analysis"]
)
def run_workflow_step_1(figma_id, figma_key, url, execution_id, thread_id, hf_token=""):
"""Run the first part of the workflow until the breakpoint."""
print(f" ⚙️ Initializing workflow for thread: {thread_id}")
app = create_workflow()
config = {"configurable": {"thread_id": thread_id}}
initial_state = create_initial_state(
figma_file_key=figma_id,
figma_access_token=figma_key,
website_url=url,
hf_token=hf_token,
execution_id=execution_id
)
# Run until interrupt
print(" 🏃 Running workflow nodes...")
try:
for event in app.stream(initial_state, config, stream_mode="values"):
# Log the current node if possible
if event:
print(f" 📍 Current state updated")
except Exception as e:
print(f" ❌ Error during workflow execution: {str(e)}")
raise
return app.get_state(config)
def resume_workflow(thread_id, user_approval=True):
"""Resume the workflow after human approval."""
app = create_workflow()
config = {"configurable": {"thread_id": thread_id}}
# Update state with approval
app.update_state(config, {"user_approval": user_approval})
# Resume execution
for event in app.stream(None, config, stream_mode="values"):
state = event
return app.get_state(config) |