From 02e29495c08eeb0f061784dc1ecefbbd1bfb53e0 Mon Sep 17 00:00:00 2001 From: Adam Moussa <166072409+amoussa1229@users.noreply.github.com> Date: Fri, 8 May 2026 12:54:00 -0400 Subject: [PATCH] Initial LangGraph + Composio orchestration graph Multi-model agent routing with 7 agent nodes (Sonnet, Haiku, GPT-4.1, Gemini, DeepSeek) and 15 pre-loaded Composio tools for Slack, Notion, GitHub, and Google Drive integration. --- .env.example | 13 +++++ .gitignore | 5 ++ agents.py | 96 ++++++++++++++++++++++++++++++++ graph.py | 142 +++++++++++++++++++++++++++++++++++++++++++++++ models.py | 37 ++++++++++++ requirements.txt | 7 +++ state.py | 20 +++++++ tools.py | 31 +++++++++++ 8 files changed, 351 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 agents.py create mode 100644 graph.py create mode 100644 models.py create mode 100644 requirements.txt create mode 100644 state.py create mode 100644 tools.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..eca36a8 --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +# LangSmith / LangGraph Cloud +LANGSMITH_API_KEY= +LANGCHAIN_TRACING_V2=true +LANGCHAIN_PROJECT=seahaven-orchestrator + +# Composio +COMPOSIO_API_KEY= + +# LLM Provider Keys (for multi-model routing) +ANTHROPIC_API_KEY= +OPENAI_API_KEY= +GOOGLE_API_KEY= +DEEPSEEK_API_KEY= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..eec7379 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.env +__pycache__/ +*.pyc +.venv/ +.langgraph/ diff --git a/agents.py b/agents.py new file mode 100644 index 0000000..8b5b8d9 --- /dev/null +++ b/agents.py @@ -0,0 +1,96 @@ +from langchain_core.messages import SystemMessage, HumanMessage +from state import OrchestratorState +from models import ( + get_implementer, + get_reviewer, + get_researcher, + get_cross_reviewer, + get_scanner, + get_fast_coder, +) + +IMPLEMENTER_PROMPT = """You are an implementation agent. You write clean, production-ready code. +Follow the spec exactly. No over-engineering, no unnecessary abstractions. +Return the complete implementation with file paths.""" + +REVIEWER_PROMPT = """You are a code review agent. Review code changes for correctness, security, and maintainability. + +For each issue found, categorize it: +- BLOCK — Must fix before merge. Security vulnerabilities, data loss risks, broken logic. +- FIX — Should fix. Bugs, performance issues, missing error handling at boundaries. +- NIT — Optional. Style, naming, minor improvements. +- QUESTION — Needs clarification. Intent is unclear. + +Start with a one-line summary: APPROVE, REQUEST CHANGES, or NEEDS DISCUSSION. +Then list findings grouped by category (BLOCK > FIX > NIT > QUESTION). +If the code is fine, say "No issues found." No praise or filler.""" + +RESEARCHER_PROMPT = """You are a research agent. Look up documentation, API references, and technical answers. +Be concise and cite sources. Return factual information, not opinions.""" + +CROSS_REVIEWER_PROMPT = """You are a cross-family code review agent. You provide an independent review perspective. +Review the provided code for bugs, security issues, and improvements. +Categorize findings as BLOCK, FIX, or NIT. Be concise. +Focus on issues that might be missed by the primary development team.""" + +SCANNER_PROMPT = """You are a large-context analysis agent. You analyze codebases, identify patterns, +check consistency across files, and find structural issues. +Summarize findings concisely with file references.""" + +FAST_CODER_PROMPT = """You are a fast coding agent for well-specified tasks. +Implement exactly what is asked. No extras, no refactoring beyond scope. +Return complete, working code.""" + + +def implementer_node(state: OrchestratorState) -> dict: + llm = get_implementer() + response = llm.invoke([ + SystemMessage(content=IMPLEMENTER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} + + +def reviewer_node(state: OrchestratorState) -> dict: + llm = get_reviewer() + response = llm.invoke([ + SystemMessage(content=REVIEWER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} + + +def researcher_node(state: OrchestratorState) -> dict: + llm = get_researcher() + response = llm.invoke([ + SystemMessage(content=RESEARCHER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} + + +def cross_reviewer_node(state: OrchestratorState) -> dict: + llm = get_cross_reviewer() + response = llm.invoke([ + SystemMessage(content=CROSS_REVIEWER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} + + +def scanner_node(state: OrchestratorState) -> dict: + llm = get_scanner() + response = llm.invoke([ + SystemMessage(content=SCANNER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} + + +def fast_coder_node(state: OrchestratorState) -> dict: + llm = get_fast_coder() + response = llm.invoke([ + SystemMessage(content=FAST_CODER_PROMPT), + HumanMessage(content=state["task"]), + ]) + return {"result": response.content, "messages": [response]} diff --git a/graph.py b/graph.py new file mode 100644 index 0000000..835338a --- /dev/null +++ b/graph.py @@ -0,0 +1,142 @@ +from dotenv import load_dotenv + +load_dotenv(".env") + +from langchain_core.messages import SystemMessage, HumanMessage +from langgraph.graph import StateGraph, START, END +from langgraph.prebuilt import ToolNode + +from state import OrchestratorState +from models import get_orchestrator +from agents import ( + implementer_node, + reviewer_node, + researcher_node, + cross_reviewer_node, + scanner_node, + fast_coder_node, +) +from tools import get_composio_tools + +ROUTER_PROMPT = """You are a task router for Sea Haven Industries. Analyze the incoming task and decide which agent should handle it. + +Available agents: +- implementer: Write new code, add features, fix bugs. Use for any coding task with a clear spec. +- reviewer: Review code changes (diffs, PRs) for correctness, security, maintainability. Uses Claude. +- researcher: Look up documentation, API references, technical questions. Fast and cheap. +- cross_reviewer: Independent code review using a different AI model (GPT). Use when you want a second opinion that catches different blind spots than Claude. +- scanner: Analyze large codebases for patterns, consistency, structural issues. Uses Gemini's large context window. +- fast_coder: Quick, bounded coding for crystal-clear specs. Uses DeepSeek. Best for small, well-defined tasks. +- connector: Interact with external services (Slack, Notion, Google Drive, GitHub) — send messages, read/update pages, find files. +- done: The task is complete or doesn't need agent delegation (e.g., a simple question you can answer directly). + +Respond with ONLY the agent name, nothing else. Pick the single best match.""" + + +composio_tools = get_composio_tools() + + +def router_node(state: OrchestratorState) -> dict: + llm = get_orchestrator() + response = llm.invoke([ + SystemMessage(content=ROUTER_PROMPT), + HumanMessage(content=state["task"]), + ]) + route = response.content.strip().lower() + valid = { + "implementer", + "reviewer", + "researcher", + "cross_reviewer", + "scanner", + "fast_coder", + "connector", + "done", + } + if route not in valid: + route = "researcher" + return {"route": route, "messages": [response]} + + +def connector_node(state: OrchestratorState) -> dict: + llm = get_orchestrator().bind_tools(composio_tools) + response = llm.invoke([ + SystemMessage( + content=( + "You help interact with external services. Use the available tools to complete the task. " + "Make exactly ONE tool call, then stop. Do not chain multiple calls." + ) + ), + HumanMessage(content=state["task"]), + ]) + return {"messages": [response]} + + +def summarizer_node(state: OrchestratorState) -> dict: + last_msg = state["messages"][-1] + content = last_msg.content if hasattr(last_msg, "content") else str(last_msg) + if isinstance(content, list): + content = "\n".join(str(c) for c in content) + if len(content) > 2000: + content = content[:2000] + "...(truncated)" + return {"result": content} + + +def route_task(state: OrchestratorState) -> str: + return state["route"] + + +def build_graph(): + graph = StateGraph(OrchestratorState) + + graph.add_node("router", router_node) + graph.add_node("implementer", implementer_node) + graph.add_node("reviewer", reviewer_node) + graph.add_node("researcher", researcher_node) + graph.add_node("cross_reviewer", cross_reviewer_node) + graph.add_node("scanner", scanner_node) + graph.add_node("fast_coder", fast_coder_node) + graph.add_node("connector", connector_node) + graph.add_node("tool_executor", ToolNode(composio_tools)) + graph.add_node("summarizer", summarizer_node) + + graph.add_edge(START, "router") + + graph.add_conditional_edges( + "router", + route_task, + { + "implementer": "implementer", + "reviewer": "reviewer", + "researcher": "researcher", + "cross_reviewer": "cross_reviewer", + "scanner": "scanner", + "fast_coder": "fast_coder", + "connector": "connector", + "done": END, + }, + ) + + graph.add_edge("implementer", END) + graph.add_edge("reviewer", END) + graph.add_edge("researcher", END) + graph.add_edge("cross_reviewer", END) + graph.add_edge("scanner", END) + graph.add_edge("fast_coder", END) + + graph.add_edge("connector", "tool_executor") + graph.add_edge("tool_executor", "summarizer") + graph.add_edge("summarizer", END) + + return graph.compile() + + +app = build_graph() + +if __name__ == "__main__": + import sys + + task = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else "What is the capital of France?" + result = app.invoke({"task": task, "messages": []}) + print(f"\n--- Route: {result['route']} ---") + print(result.get("result", "No result")) diff --git a/models.py b/models.py new file mode 100644 index 0000000..1880861 --- /dev/null +++ b/models.py @@ -0,0 +1,37 @@ +import os +from langchain_anthropic import ChatAnthropic +from langchain_openai import ChatOpenAI +from langchain_google_genai import ChatGoogleGenerativeAI + + +def get_orchestrator(): + return ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0) + + +def get_implementer(): + return ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0) + + +def get_reviewer(): + return ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0) + + +def get_researcher(): + return ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0) + + +def get_cross_reviewer(): + return ChatOpenAI(model="gpt-4.1", temperature=0.2) + + +def get_scanner(): + return ChatGoogleGenerativeAI(model="gemini-2.5-pro", temperature=0) + + +def get_fast_coder(): + return ChatOpenAI( + model="deepseek-coder", + base_url="https://api.deepseek.com/v1", + api_key=os.getenv("DEEPSEEK_API_KEY"), + temperature=0, + ) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..97aceac --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +langgraph>=1.1.0 +langchain-anthropic>=1.4.0 +langchain-openai>=1.2.0 +langchain-google-genai>=4.2.0 +langchain-community>=0.4.0 +composio-langgraph>=0.13.0 +python-dotenv>=1.0.0 diff --git a/state.py b/state.py new file mode 100644 index 0000000..5f7fc9f --- /dev/null +++ b/state.py @@ -0,0 +1,20 @@ +from typing import Annotated, Literal +from typing_extensions import TypedDict +from langgraph.graph.message import add_messages +from langchain_core.messages import AnyMessage + + +class OrchestratorState(TypedDict): + messages: Annotated[list[AnyMessage], add_messages] + task: str + route: Literal[ + "implementer", + "reviewer", + "researcher", + "cross_reviewer", + "scanner", + "fast_coder", + "connector", + "done", + ] + result: str diff --git a/tools.py b/tools.py new file mode 100644 index 0000000..0facd31 --- /dev/null +++ b/tools.py @@ -0,0 +1,31 @@ +from composio import Composio +from composio_langgraph import LanggraphProvider + +TOOL_SLUGS = [ + # Slack + "SLACK_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL", + "SLACK_FETCH_CONVERSATION_HISTORY", + "SLACK_FIND_CHANNELS", + "SLACK_FIND_USERS", + "SLACK_ADD_REACTION_TO_AN_ITEM", + "SLACK_FETCH_MESSAGE_THREAD_FROM_A_CONVERSATION", + # Notion + "NOTION_SEARCH_NOTION_PAGE", + "NOTION_RETRIEVE_PAGE", + "NOTION_UPDATE_PAGE", + "NOTION_CREATE_NOTION_PAGE", + "NOTION_ADD_PAGE_CONTENT", + # GitHub + "GITHUB_CREATE_AN_ISSUE", + "GITHUB_LIST_REPO_ISSUES", + "GITHUB_GET_A_REPOSITORY", + # Google Drive + "GOOGLEDRIVE_FIND_FILE", + "GOOGLEDRIVE_GET_FILE_METADATA", +] + + +def get_composio_tools(): + client = Composio(provider=LanggraphProvider()) + tools = client.tools.get(user_id="default", tools=TOOL_SLUGS) + return tools