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

@@ -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=

.gitignore

@@ -0,0 +1,5 @@
+.env
+__pycache__/
+*.pyc
+.venv/
+.langgraph/

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]}

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"))

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,
+    )

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

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

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