LangGraph Subgraphs, Send API & LangGraph Platform Deploy Errors

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class ChildState(TypedDict):
    messages: Annotated[list, add_messages]
    docs: Annotated[list, add]
    citations: Annotated[list, add]


def retrieve(state: ChildState) -> dict:
    return {"docs": ["doc1", "doc2"]}


def cite(state: ChildState) -> dict:
    return {"citations": [f"[{i}]" for i in range(len(state["docs"]))]}


child_graph = StateGraph(ChildState)
child_graph.add_node("retrieve", retrieve)
child_graph.add_node("cite", cite)
child_graph.add_edge(START, "retrieve")
child_graph.add_edge("retrieve", "cite")
child_graph.add_edge("cite", END)
child_app = child_graph.compile()


# PARENT declares docs and citations to receive them from child
class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    question: str
    docs: Annotated[list, add]
    citations: Annotated[list, add]


parent_graph = StateGraph(ParentState)
parent_graph.add_node("child", child_app)
parent_graph.add_edge(START, "child")
parent_graph.add_edge("child", END)
parent_app = parent_graph.compile()

from typing import Annotated, TypedDict
from operator import add
from langgraph.types import Send
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    topics: list[str]
    summaries: Annotated[list, add]         # reducer required for parallel Send merge
    errors: Annotated[list, add]


def fanout(state):
    return [Send("summarize_one", {"topic": t}) for t in state["topics"]]


def summarize_one(state) -> dict:
    try:
        summary = llm.invoke(f"Summarize {state['topic']} in 20 words.").content
        return {"summaries": [summary]}
    except Exception as e:
        return {"errors": [f"{state['topic']}: {e}"]}


def combine(state):
    return {"messages": [("assistant", "\n".join(state["summaries"]))]}


graph = StateGraph(State)
graph.add_node("summarize_one", summarize_one)
graph.add_node("combine", combine)
graph.add_conditional_edges(START, fanout, ["summarize_one"])
graph.add_edge("summarize_one", "combine")
graph.add_edge("combine", END)
app = graph.compile()

# Cap fan-out via config
result = app.invoke(
    {"topics": ["a", "b", "c"], "summaries": [], "errors": []},
    {"max_concurrency": 10},
)

{
  "dependencies": ["."],
  "graphs": {
    "chat_agent": "./src/agents/chat.py:app",
    "research_agent": "./src/agents/research.py:app"
  },
  "env": ".env",
  "python_version": "3.12",
  "dockerfile_lines": [
    "RUN apt-get update && apt-get install -y --no-install-recommends libmagic1 && rm -rf /var/lib/apt/lists/*"
  ]
}

# ✅ Export a COMPILED app variable at module level
# LangGraph Platform imports this variable directly

import os
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool


def _build_graph(checkpointer):
    g = StateGraph(MessagesState)
    g.add_node("responder", responder)
    g.add_edge(START, "responder")
    g.add_edge("responder", END)
    return g.compile(checkpointer=checkpointer)


# Platform provides POSTGRES_URI at build/run time via the langgraph.json env
_pool = AsyncConnectionPool(
    os.environ["POSTGRES_URI"],
    min_size=2, max_size=20,
    kwargs={"prepare_threshold": None, "autocommit": True},
    open=False,  # opened lazily
)
_checkpointer = AsyncPostgresSaver(_pool)

# THE variable Platform expects — top-level, compiled
app = _build_graph(_checkpointer)


# ❌ ANTI-PATTERN — Platform can't call a factory
# def build_app():
#     return _build_graph(_checkpointer)
# Platform will fail with: "chat.py:app is not a compiled graph"

# ✅ Declare every env var; Platform injects these at runtime
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...
POSTGRES_URI=postgresql://user:pass@host:5432/langgraph
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_PROJECT=my-agent-prod
LOG_LEVEL=INFO