PythonAgentAI_ResponsibilityMapping/agent_graph.py at main · kb0422bk/PythonAgentAI_ResponsibilityMapping · 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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from langgraph.graph import StateGraph, END
from state import AgentState
from nodes import (
    fetch_law_changes_node,
    fetch_target_users_node,
    send_notifications_node,
    classify_intent_node,
    fetch_law_list_node,
    fetch_specific_law_node,
    fetch_all_laws_node,
    analyze_law_changes_node,
    generate_optimized_search_query_node,
    retrieve_rag_context_node,
    analyze_todo_list_node,
    fallback_router_node
)


# =====================================================================
# PIPELINE 1: BACKGROUND AUTOMATION GRAPH (Scheduled Cron Tasks)
# =====================================================================
# This state graph orchestrates automated, non-interactive batch processes
# executed by background runners (e.g., run_cron.py).
cron_workflow = StateGraph(AgentState)

# 1. Register operational batch processing nodes
cron_workflow.add_node("fetch_law_changes", fetch_law_changes_node)
cron_workflow.add_node("fetch_target_users", fetch_target_users_node)
cron_workflow.add_node("send_notifications", send_notifications_node)

# 2. Establish linear background automation execution chain
cron_workflow.set_entry_point("fetch_law_changes")
cron_workflow.add_edge("fetch_law_changes", "fetch_target_users")
cron_workflow.add_edge("fetch_target_users", "send_notifications")
cron_workflow.add_edge("send_notifications", END)

# Compile the Background Automation Application Engine
automation_app = cron_workflow.compile()

# =====================================================================
# PIPELINE 2: INTERACTIVE CHATFLOW GRAPH (Real-Time FastAPI Web SSE)
# =====================================================================
# This complex state graph drives the live multi-agent chat interface.
# It processes asynchronous user requests via api.py.
chat_workflow = StateGraph(AgentState)

# 1. Register interactive runtime processing nodes
chat_workflow.add_node("classifier", classify_intent_node)
chat_workflow.add_node("fetch_law_list", fetch_law_list_node)
chat_workflow.add_node("fetch_specific_law", fetch_specific_law_node)
chat_workflow.add_node("fetch_all_laws", fetch_all_laws_node)
chat_workflow.add_node("llm_analyzer", analyze_law_changes_node)
chat_workflow.add_node("generate_optimized_search_query", generate_optimized_search_query_node)
chat_workflow.add_node("retrieve_rag_context", retrieve_rag_context_node)
chat_workflow.add_node("analyze_todo_list", analyze_todo_list_node)
chat_workflow.add_node("fallback_router", fallback_router_node)


# 2. Initialize graph entry point at the Intent Classifier node
chat_workflow.set_entry_point("classifier")

# 3. Define the Dynamic Conditional Routing Edge
def routing_condition_edge(state: AgentState):
    """
        Evaluates the mutated state context to dynamically route execution paths
        based on the intent classification result.
    """
    classification = state.get("target_classification")

    if classification == 1:
        return "law_list"
    elif classification == 2:
        return "specific_law"
    elif classification == 3:
        return "all_laws"
    elif classification == 4:
        return "optimized_search_query"
    else:
        return "fallback"

chat_workflow.add_conditional_edges(
    "classifier",
    routing_condition_edge,
    {
        "law_list": "fetch_law_list",
        "specific_law": "fetch_specific_law",
        "all_laws": "fetch_all_laws",
        "optimized_search_query": "generate_optimized_search_query",
        "fallback": "fallback_router"
    }
)

# 4. Map Downstream Directed Edges & Sub-Graph Finalizations
# Branch 1: Classification 1 (Standard Law Directory Retrieval)
chat_workflow.add_edge("fetch_law_list", END)

# Branch 2: Classification 2 (Specific Regulation Analysis)
chat_workflow.add_edge("fetch_specific_law", "llm_analyzer")
chat_workflow.add_edge("llm_analyzer", END)

# Branch 3: Classification 3 (Global Regulatory Delta Processing)
chat_workflow.add_edge("fetch_all_laws", "llm_analyzer")
chat_workflow.add_edge("llm_analyzer", END)

# Branch 4: Classification 4 (Deep RAG-Powered Compliance To-Do Evaluation)
chat_workflow.add_edge("generate_optimized_search_query", "retrieve_rag_context")
chat_workflow.add_edge("retrieve_rag_context", "analyze_todo_list")
chat_workflow.add_edge("analyze_todo_list", END)

# Branch 5: Classification 5 (Fallback / Unhandled Exception Routing)
chat_workflow.add_edge("fallback_router", END)

# Compile the Runtime Chat Application Engine
app = chat_workflow.compile()