Skip to content

script_analyzer_graph

script_analyzer_graph

compile_script_analyzer_graph 

compile_script_analyzer_graph()

Compile the LangGraph workflow for script analysis.

This function imports langgraph dynamically to avoid hard dependency when only the basic analyzer is used.

Source code in src/crew_dcs/integrations/agents/script_analyzer/script_analyzer_graph.py 
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
def compile_script_analyzer_graph():
    """Compile the LangGraph workflow for script analysis.

    This function imports langgraph dynamically to avoid hard dependency
    when only the basic analyzer is used.
    """
    try:
        from langgraph.graph import END, START, StateGraph
    except Exception as exc:  # ImportError or other
        raise RuntimeError(
            "LangGraph not installed; please add 'langgraph' dependency to use the graph."
        ) from exc

    workflow = StateGraph(PythonAnalysisState)

    workflow.add_node("code_analyzer", run_code_analysis)
    workflow.add_node("pattern_matcher", run_pattern_matching)
    workflow.add_node("recommendations", run_recommendations)

    workflow.add_edge(START, "code_analyzer")
    workflow.add_edge("code_analyzer", "pattern_matcher")
    workflow.add_edge("pattern_matcher", "recommendations")
    workflow.add_edge("recommendations", END)

    return workflow.compile()

run_workflow 

run_workflow(script: str) -> dict[str, Any]

Run the compiled workflow against a Python script and return output state.

Source code in src/crew_dcs/integrations/agents/script_analyzer/script_analyzer_graph.py 
38
39
40
41
42
43
def run_workflow(script: str) -> dict[str, Any]:
    """Run the compiled workflow against a Python script and return output state."""
    graph = compile_script_analyzer_graph()
    initial: PythonAnalysisState = {"python_script": script}
    result = graph.invoke(initial)
    return dict(result)