"""Main file for constructing the EA4ALL hierarchical graph""" """ EA4ALL Hierarchical Graph This module defines the main file for constructing the EA4ALL hierarchical graph. It contains functions and classes for creating and managing the graph structure. Functions: - make_supervisor_node: Creates a supervisor node for managing a conversation between architect workers. - call_landscape_agentic: Calls the landscape agentic graph. - call_diagram_agentic: Calls the diagram agentic graph. - call_togaf_agentic: Calls the togaf agentic graph. - websearch: Search for real-time data to answer user's question Classes: - Router: TypedDict representing the worker to route to next. Attributes: - model: The LLM client for the supervisor model. - super_builder: The StateGraph builder for constructing the graph. - super_graph: The compiled EA4ALL Agentic Workflow Graph. Note: This module depends on other modules and packages such as langchain_core, langgraph, shared, ea4all_apm, ea4all_vqa, and ea4all_gra. """ """Changelog: - lanchain_openapi: 0.2.9 (0.3.6 issue with max_tokens for HF models) #2025-06-03 - Refactored code to fix problems with linter and type checking (Standard mode) #2025-06-13 – Deployed to ea4all-langgraph-lgs - Renamed graphs: after removing .langgraph_api folder the errors caused by old-graph names being still exposed was FIXED. #20250624 - Retrofitted StateGraph to use OverallState, InputState, and OutputState #20250626 - Call_generate_websearch State dict updated to match APMGraph OverallState #20250627 - Diagram graph invoke returning empty message - fixed 30-Jun #20250707 - AgentState refactored to OverallState, InputState, and OutputState - Microsoft and AWS MCP tools added to the EA4ALL Agentic Workflow Graph #20250717 - AgentState class is exclusively used for React Agent #20250718 - Refactored super graph InputState as MessagesState schema and OutputState with question attribute - Refactored websearch & apm_grap to MessagesState - get_llm_client updated with new HF Inference API endpoint https://router.huggingface.co/v1 - ISSUE: APM Graph not passing internally the Question attribute coming from the Supervisor Node - fixed #20250719 - Refactored ea4all_vqa to use MessagesState - Added extracted_and_validated_path function to utils.py for extracting and validating file paths from user input - ISSUE: VQA Diagram not returning response but safety_check status only #20250720 - Refactored ea4all_mcp to use standard AgentState state schema as react_agent does not use InputState schema - ISSUE: VQA Diagram not returning response but safety_check status only - fixed -- Re-added messages attribute to OutputState schema -- Response added to AgentFinish log """ from langgraph.types import Command from langchain_core.messages import ( AIMessage ) from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.runnables import RunnableConfig from langchain_core.messages import HumanMessage from langchain import hub from langgraph.graph import ( START, END, StateGraph, ) from typing_extensions import Literal, TypedDict import uuid from ea4all.src.shared.configuration import BaseConfiguration from ea4all.src.shared.utils import get_llm_client, extract_text_from_message_content, extract_and_validate_path from ea4all.src.shared.state import InputState, OutputState, OverallState from ea4all.src.tools.tools import websearch from ea4all.src.ea4all_indexer.graph import indexer_graph from ea4all.src.ea4all_apm.graph import apm_graph from ea4all.src.ea4all_vqa.graph import diagram_graph from ea4all.src.ea4all_gra.graph import togaf_graph from ea4all.src.ea4all_mcp.graph import mcp_docs_graph async def call_indexer_apm(state: OverallState, config: RunnableConfig): response = await indexer_graph.ainvoke(input={"docs":[]}, config=config) def make_supervisor_node(model: BaseChatModel, members: list[str]): options = ["FINISH"] + members system_prompt = hub.pull("ea4all_super_graph").template class Router(TypedDict): """Worker to route to next. If no workers needed, route to FINISH.""" next: Literal["FINISH", "portfolio_team", "diagram_team", "blueprint_team", "websearch_team", "mcp_docs_team"] async def supervisor_node(state: OverallState, config: RunnableConfig) -> Command[Literal["portfolio_team", "diagram_team", "blueprint_team", "websearch_team", "mcp_docs_team", '__end__']]: """An LLM-based router.""" messages = [ {"role": "system", "content": system_prompt}, ] messages.append({"role": "user", "content": state['messages'][-1].content}) response = await model.with_structured_output(Router).ainvoke(input=messages, config=config) _goto = "__end__" if isinstance(response, dict): _goto = response["next"] # Ensure _goto is one of the allowed Literal values if _goto not in ["portfolio_team", "diagram_team", "blueprint_team", "websearch_team", "mcp_docs_team"]: _goto = "__end__" question = extract_text_from_message_content(state['messages'][-1].content) print(f"---Supervisor got a request--- Question: {question} ==> Routing to {_goto}\n") return Command(update={'question':question}, goto=_goto) return supervisor_node async def call_landscape_agentic(state: OverallState, config: RunnableConfig) -> Command[Literal['__end__']]: ##2025-02-21: NOT passing CHAT MEMORY to the APM_graph response = await apm_graph.ainvoke(state, config=config) return Command( update={ "messages": [ AIMessage( content=str(response), name="landscape_agentic" ) ] }, goto="__end__", ) async def call_diagram_agentic(state: OverallState, config: RunnableConfig) -> Command[Literal['__end__']]: #NOT passing CHAT MEMORY to the Diagram_graph from ea4all.src.ea4all_vqa.configuration import AgentConfiguration from ea4all.src.ea4all_vqa.state import InputState image_files = AgentConfiguration.from_runnable_config(config).vqa_images user_input = extract_and_validate_path(text=state.get('question'), allowed_extensions=image_files) if user_input['found'] and user_input['is_file']: inputs = InputState( #question=user_input['message'], messages=[HumanMessage(content=user_input['message'])], image=user_input['path'] ) response = await diagram_graph.ainvoke( input=inputs, config=config ) msg = response['messages'][-1].content if response['messages'] and hasattr(response['messages'][-1], 'content') else "No valid response from Diagram agentic graph." else: msg = f"No valid image file provided. Please provide one of the accepted file types: {image_files}." return Command( update={ "messages": [ AIMessage( content=msg, name="diagram_agentic" ) ] }, goto="__end__", ) async def call_togaf_agentic(state: OverallState, config: RunnableConfig) -> Command[Literal["__end__"]]: #NOT passing CHAT MEMORY to the Togaf_graph print(f"---TOGAF ROUTE team node ready to --- CALL_TOGAF_AGENTIC Routing to {getattr(state, 'next')} with User Question: {getattr(state, 'question')}") inputs = { "messages": [ { "role": "user", "content": getattr(state, "question")[-1].content } ] } #user response response = await togaf_graph.ainvoke( input=inputs, config=config ) #astream not loading the graph return Command( update={ "messages": [ AIMessage( content=response["messages"][-1].content, name="togaf_agentic" ) ] }, goto="__end__", ) # Wrap-up websearch answer to user's question async def call_generate_websearch(state:OverallState, config: RunnableConfig) -> Command[Literal["__end__"]]: from ea4all.src.ea4all_apm.state import OverallState if config is not None: source = config.get('metadata', {}).get('langgraph_node', 'unknown') # Invoke GENERATOR node in the APMGraph messages = state.get('messages') last_message_content = messages[-1].content if messages and hasattr(messages[-1], "content") else "" state_dict = { "messages": last_message_content, "source": "websearch", "question": state.get('question'), } apm_state = OverallState(**state_dict) generation = await apm_graph.nodes["generate"].ainvoke(apm_state, config) return Command( update={ "messages": [ AIMessage( content=generation['generation'], name="generate_websearch" ) ] }, goto="__end__", ) async def blueprint_team(state: OverallState) -> Command[Literal["togaf_route"]]: print("---Blueprint team got a request--- Routing to TOGAF_ROUTE node") return Command(update={**state}, goto="togaf_route") async def diagram_team(state: OverallState) -> Command[Literal["diagram_route"]]: print("---Diagram team got a request--- Routing to DIAGRAM_ROUTE node") return Command(update={**state}, goto="diagram_route") async def mcp_docs_team(state: OverallState) -> Command[Literal["__end__"]]: print("---MCP Docs team got a request--- Routing to MCP_DOCS_AGENTIC node") # Invoke the EA4ALL MCP Docs graph mcp_response = await mcp_docs_graph.ainvoke( {"messages": [{"role": "user", "content": state.get('question')}]} ) return Command( update={ "messages": [ AIMessage( content=mcp_response['messages'][-1].content, name="mcp_docs_agentic" ) ] }, goto="__end__", ) async def call_websearch(state: OverallState, config: RunnableConfig) -> Command[Literal["generate_websearch"]]: """ Web search based on the re-phrased question. Args: state (OverallState): The current graph state config (RunnableConfig): Configuration with the model used for query analysis. Returns: Command: Command to route to the generate_websearch node. """ print(f"---Websearch team got a request--- Routing to GENERATE_WEBSEARCH node with User Question: {state.get('question')}") _state = dict(state) response = await websearch(_state) return Command(update=response, goto="generate_websearch") async def super_graph_entry_point(state: OverallState): # Generate a unique thread ID thread_config = RunnableConfig({"configurable": {"thread_id": str(uuid.uuid4())}}) # Initialize state if not provided if state is None: state = { "messages": [ ("system", "You are a helpful assistant"), ("human", "Start the workflow") ] } # Build and compile the graph graph = build_super_graph() # Async invocation try: # Use ainvoke for async execution result = await graph.ainvoke(state, config=RunnableConfig(thread_config)) return result except Exception as e: print(f"Graph execution error: {e}") raise # Define & build the graph. def build_super_graph(): model = get_llm_client(BaseConfiguration.supervisor_model, api_base_url="", streaming=BaseConfiguration.streaming) teams_supervisor_node = make_supervisor_node(model, ["portfolio_team", "diagram_team", "blueprint_team","websearch_team"]) super_builder = StateGraph(state_schema=OverallState, input_schema=InputState, output_schema=OutputState, config_schema=BaseConfiguration) super_builder.add_node("apm_indexer", call_indexer_apm) super_builder.add_node("supervisor", teams_supervisor_node) super_builder.add_node("portfolio_team", call_landscape_agentic) super_builder.add_node("websearch_team", call_websearch) super_builder.add_node("diagram_team", diagram_team) super_builder.add_node("blueprint_team", blueprint_team) super_builder.add_node("generate_websearch", call_generate_websearch) super_builder.add_node("diagram_route", call_diagram_agentic) super_builder.add_node("togaf_route", call_togaf_agentic) super_builder.add_node("mcp_docs_team", mcp_docs_team) super_builder.add_edge(START, "apm_indexer") super_builder.add_edge("apm_indexer", "supervisor") super_builder.add_edge("websearch_team", "generate_websearch") super_builder.add_edge("blueprint_team", "togaf_route") super_builder.add_edge("diagram_team", "diagram_route") super_builder.add_edge("portfolio_team", END) super_builder.add_edge("generate_websearch", END) super_builder.add_edge("togaf_route", END) super_builder.add_edge("diagram_route", END) super_builder.add_edge("mcp_docs_team", END) super_graph = super_builder.compile() super_graph.name = "ea4all_supervisor" return super_graph # Export the graph for LangGraph Dev/Studio super_graph = build_super_graph()