| """ |
| intelligenceAgentGraph.py - Intelligence Agent Graph with Subgraph Architecture |
| """ |
|
|
| import uuid |
| from langgraph.graph import StateGraph, END |
| from src.states.intelligenceAgentState import IntelligenceAgentState |
| from src.nodes.intelligenceAgentNode import IntelligenceAgentNode |
| from src.llms.groqllm import GroqLLM |
|
|
|
|
| class IntelligenceGraphBuilder: |
| def __init__(self, llm): |
| self.llm = llm |
|
|
| def build_profile_monitoring_subgraph( |
| self, node: IntelligenceAgentNode |
| ) -> StateGraph: |
| subgraph = StateGraph(IntelligenceAgentState) |
| subgraph.add_node("monitor_profiles", node.collect_profile_activity) |
| subgraph.set_entry_point("monitor_profiles") |
| subgraph.add_edge("monitor_profiles", END) |
| return subgraph.compile() |
|
|
| def build_competitive_intelligence_subgraph( |
| self, node: IntelligenceAgentNode |
| ) -> StateGraph: |
| subgraph = StateGraph(IntelligenceAgentState) |
|
|
| subgraph.add_node("competitor_mentions", node.collect_competitor_mentions) |
| subgraph.add_node("product_reviews", node.collect_product_reviews) |
| subgraph.add_node("market_intelligence", node.collect_market_intelligence) |
|
|
| subgraph.set_entry_point("competitor_mentions") |
| subgraph.set_entry_point("product_reviews") |
| subgraph.set_entry_point("market_intelligence") |
|
|
| subgraph.add_edge("competitor_mentions", END) |
| subgraph.add_edge("product_reviews", END) |
| subgraph.add_edge("market_intelligence", END) |
|
|
| return subgraph.compile() |
|
|
| def build_feed_generation_subgraph(self, node: IntelligenceAgentNode) -> StateGraph: |
| subgraph = StateGraph(IntelligenceAgentState) |
|
|
| subgraph.add_node("categorize", node.categorize_intelligence) |
| subgraph.add_node("llm_summary", node.generate_llm_summary) |
| subgraph.add_node("format_output", node.format_final_output) |
|
|
| subgraph.set_entry_point("categorize") |
| subgraph.add_edge("categorize", "llm_summary") |
| subgraph.add_edge("llm_summary", "format_output") |
| subgraph.add_edge("format_output", END) |
|
|
| return subgraph.compile() |
|
|
| def build_graph(self): |
| node = IntelligenceAgentNode(self.llm) |
|
|
| profile_subgraph = self.build_profile_monitoring_subgraph(node) |
| intelligence_subgraph = self.build_competitive_intelligence_subgraph(node) |
| feed_subgraph = self.build_feed_generation_subgraph(node) |
|
|
| main_graph = StateGraph(IntelligenceAgentState) |
|
|
| main_graph.add_node( |
| "profile_monitoring_module", lambda state: profile_subgraph.invoke(state) |
| ) |
| main_graph.add_node( |
| "competitive_intelligence_module", |
| lambda state: intelligence_subgraph.invoke(state), |
| ) |
| main_graph.add_node( |
| "feed_generation_module", lambda state: feed_subgraph.invoke(state) |
| ) |
| main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds) |
|
|
| main_graph.set_entry_point("profile_monitoring_module") |
| main_graph.set_entry_point("competitive_intelligence_module") |
|
|
| main_graph.add_edge("profile_monitoring_module", "feed_generation_module") |
| main_graph.add_edge("competitive_intelligence_module", "feed_generation_module") |
| main_graph.add_edge("feed_generation_module", "feed_aggregator") |
| main_graph.add_edge("feed_aggregator", END) |
|
|
| return main_graph.compile() |
|
|
|
|
| llm = GroqLLM().get_llm() |
| graph = IntelligenceGraphBuilder(llm).build_graph() |
|
|
