62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
"""Dynamic LangGraph StateGraph Factory with Skill Injection."""
|
|
|
|
import logging
|
|
from typing import Any, Callable
|
|
from langgraph.graph import END, START, StateGraph
|
|
from langgraph.graph.state import CompiledStateGraph
|
|
|
|
from app.config import get_settings
|
|
from app.nodes import design_node, discover_node, package_node, source_discover_node, validate_node
|
|
from app.skills.loader import SkillLoader
|
|
from app.states.state import GCPArchitectureState
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_gcp_architecture_graph(skill_loader: SkillLoader | None = None) -> CompiledStateGraph:
|
|
"""Creates and compiles the LangGraph StateGraph with dynamic local skill injection."""
|
|
if skill_loader is None:
|
|
settings = get_settings()
|
|
skill_loader = SkillLoader(settings.SKILLS_DIR)
|
|
skill_loader.load_skills()
|
|
|
|
builder = StateGraph(GCPArchitectureState)
|
|
|
|
# Wrap nodes with skill loader injection
|
|
def run_source_discover(state: GCPArchitectureState) -> dict[str, Any]:
|
|
return source_discover_node(state, skill_loader)
|
|
|
|
def run_discover(state: GCPArchitectureState) -> dict[str, Any]:
|
|
return discover_node(state, skill_loader)
|
|
|
|
def run_design(state: GCPArchitectureState) -> dict[str, Any]:
|
|
return design_node(state, skill_loader)
|
|
|
|
def run_validate(state: GCPArchitectureState) -> dict[str, Any]:
|
|
return validate_node(state, skill_loader)
|
|
|
|
def run_package(state: GCPArchitectureState) -> dict[str, Any]:
|
|
return package_node(state, skill_loader)
|
|
|
|
# Add graph nodes
|
|
builder.add_node("source_discover", run_source_discover)
|
|
builder.add_node("discover", run_discover)
|
|
builder.add_node("design", run_design)
|
|
builder.add_node("validate", run_validate)
|
|
builder.add_node("package", run_package)
|
|
|
|
# Add sequential workflow edges
|
|
builder.add_edge(START, "source_discover")
|
|
builder.add_edge("source_discover", "discover")
|
|
builder.add_edge("discover", "design")
|
|
builder.add_edge("design", "validate")
|
|
builder.add_edge("validate", "package")
|
|
builder.add_edge("package", END)
|
|
|
|
return builder.compile()
|
|
|
|
|
|
def create_agent(skill_loader: SkillLoader | None = None) -> CompiledStateGraph:
|
|
"""Alias for create_gcp_architecture_graph to support standardized agent creation."""
|
|
return create_gcp_architecture_graph(skill_loader)
|