LangGraph Fundamentals

LangGraph では、エージェントのワークフローを**有向グラフ**としてモデル化します。

• StateGraph：ステートフルなグラフを構築するためのメインクラス
• ノード：処理を実行し、状態を更新する関数
• エッジ：実行順序を定義します（静的または条件付き）
• START/END：開始点と終了点を示す特別なノード
• Reducer を使用した状態：状態の更新をどのように統合するかを制御します

グラフは実行前に compile() する必要があります。

LangGraph を使用する場合	代替手段を使用する場合
エージェントのオーケストレーションをきめ細かく制御する必要がある	手早いプロトタイピング → LangChain agents
分岐やループを含む複雑なワークフローを構築する	シンプルなステートレスワークフロー → LangChain direct
ヒューマン・イン・ザ・ループや永続化が必要	すぐに使える機能が必要 → Deep Agents

---

状態管理

必要な処理	解決策	例
値を上書きする	Reducer なし（デフォルト）	カウンターなどの単純なフィールド
リストに追加する	Reducer（operator.add / concat）	メッセージ履歴、ログ
カスタムロジック	カスタム Reducer 関数	複雑な統合

リストの累積と整数の合計に Reducer を使用する状態スキーマを定義します。 ```python from typing_extensions import TypedDict, Annotated import operator

class State(TypedDict): name: str # Default: overwrites on update messages: Annotated[list, operator.add] # Appends to list total: Annotated[int, operator.add] # Sums integers

</python>
<typescript>
配列の累積に ReducedValue を使用する StateSchema を使います。
```typescript
import { StateSchema, ReducedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  name: z.string(),  // Default: overwrites
  messages: MessagesValue,  // Built-in for messages
  items: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (current, update) => current.concat(update) }
  ),
});

Reducer がない場合、リストを返すと以前の値が上書きされます。 ```python # WRONG: List will be OVERWRITTEN class State(TypedDict): messages: list # No reducer!

Node 1 returns: {"messages": ["A"]}

Node 2 returns: {"messages": ["B"]}

Final: {"messages": ["B"]} # "A" is LOST!

CORRECT: Use Annotated with operator.add

from typing import Annotated import operator

class State(TypedDict): messages: Annotated[list, operator.add]

Final: {"messages": ["A", "B"]}

</python>
<typescript>
ReducedValue がない場合、配列は追加されずに上書きされます。
```typescript
// WRONG: Array will be overwritten
const State = new StateSchema({
  items: z.array(z.string()),  // No reducer!
});
// Node 1: { items: ["A"] }, Node 2: { items: ["B"] }
// Final: { items: ["B"] }  // A is lost!

// CORRECT: Use ReducedValue
const State = new StateSchema({
  items: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (current, update) => current.concat(update) }
  ),
});
// Final: { items: ["A", "B"] }

ノードは状態を変更して全体を返すのではなく、部分的な更新を返す必要があります。 ```python # WRONG: Returning entire state object def my_node(state: State) -> State: state["field"] = "updated" return state # Don't mutate and return!

CORRECT: Return dict with only the updates

def my_node(state: State) -> dict: return {"field": "updated"}

</python>
<typescript>
状態オブジェクト全体ではなく、部分的な更新のみを返します。
```typescript
// WRONG: Returning entire state
const myNode = async (state: typeof State.State) => {
  state.field = "updated";
  return state;  // Don't do this!
};

// CORRECT: Return partial updates
const myNode = async (state: typeof State.State) => {
  return { field: "updated" };
};

---

グラフの構築

必要な処理	エッジの種類	使用する場面
常に同じノードに進む	add_edge()	固定された決定的なフロー
状態に基づいて分岐する	add_conditional_edges()	動的な分岐
状態を更新し、経路も選択する	Command	1つのノードでロジックをまとめる
複数のノードにファンアウトする	Send	動的な入力による並列処理

2つのノードを線形エッジでつなぐシンプルなグラフです。 ```python from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict

class State(TypedDict): input: str output: str

def process_input(state: State) -> dict: return {"output": f"Processed: {state['input']}"}

def finalize(state: State) -> dict: return {"output": state["output"].upper()}

graph = ( StateGraph(State) .add_node("process", process_input) .add_node("finalize", finalize) .add_edge(START, "process") .add_edge("process", "finalize") .add_edge("finalize", END) .compile() )

result = graph.invoke({"input": "hello"}) print(result["output"]) # "PROCESSED: HELLO"

</python>
<typescript>
ノードを addEdge で連結し、呼び出す前にコンパイルします。
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  input: z.string(),
  output: z.string().default(""),
});

const processInput = async (state: typeof State.State) => {
  return { output: `Processed: ${state.input}` };
};

const finalize = async (state: typeof State.State) => {
  return { output: state.output.toUpperCase() };
};

const graph = new StateGraph(State)
  .addNode("process", processInput)
  .addNode("finalize", finalize)
  .addEdge(START, "process")
  .addEdge("process", "finalize")
  .addEdge("finalize", END)
  .compile();

const result = await graph.invoke({ input: "hello" });
console.log(result.output);  // "PROCESSED: HELLO"

条件付きエッジを使い、状態に基づいて異なるノードにルーティングします。 ```python from typing import Literal from langgraph.graph import StateGraph, START, END

class State(TypedDict): query: str route: str result: str

def classify(state: State) -> dict: if "weather" in state["query"].lower(): return {"route": "weather"} return {"route": "general"}

def route_query(state: State) -> Literal["weather", "general"]: return state["route"]

graph = ( StateGraph(State) .add_node("classify", classify) .add_node("weather", lambda s: {"result": "Sunny, 72F"}) .add_node("general", lambda s: {"result": "General response"}) .add_edge(START, "classify") .add_conditional_edges("classify", route_query, ["weather", "general"]) .add_edge("weather", END) .add_edge("general", END) .compile() )

</python>
<typescript>
addConditionalEdges は、関数の戻り値に基づいてルーティングします。
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  query: z.string(),
  route: z.string().default(""),
  result: z.string().default(""),
});

const classify = async (state: typeof State.State) => {
  if (state.query.toLowerCase().includes("weather")) {
    return { route: "weather" };
  }
  return { route: "general" };
};

const routeQuery = (state: typeof State.State) => state.route;

const graph = new StateGraph(State)
  .addNode("classify", classify)
  .addNode("weather", async () => ({ result: "Sunny, 72F" }))
  .addNode("general", async () => ({ result: "General response" }))
  .addEdge(START, "classify")
  .addConditionalEdges("classify", routeQuery, ["weather", "general"])
  .addEdge("weather", END)
  .addEdge("general", END)
  .compile();

Command を使うと、1つの return で状態を更新し、次のノードを選択できます。 ```python from langgraph.types import Command from typing import Literal

class State(TypedDict): count: int result: str

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]: """Update state AND decide next node in one return.""" new_count = state["count"] + 1 if new_count > 5: return Command(update={"count": new_count}, goto="node_c") return Command(update={"count": new_count}, goto="node_b")

graph = ( StateGraph(State) .add_node("node_a", node_a) .add_node("node_b", lambda s: {"result": "B"}) .add_node("node_c", lambda s: {"result": "C"}) .add_edge(START, "node_a") .add_edge("node_b", END) .add_edge("node_c", END) .compile() )

</python>
<typescript>
update と goto を含む Command を返すことで、状態の変更とルーティングをまとめます。
```typescript
import { StateGraph, StateSchema, START, END, Command } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  count: z.number().default(0),
  result: z.string().default(""),
});

const nodeA = async (state: typeof State.State) => {
  const newCount = state.count + 1;
  if (newCount > 5) {
    return new Command({ update: { count: newCount }, goto: "node_c" });
  }
  return new Command({ update: { count: newCount }, goto: "node_b" });
};

const graph = new StateGraph(State)
  .addNode("node_a", nodeA, { ends: ["node_b", "node_c"] })
  .addNode("node_b", async () => ({ result: "B" }))
  .addNode("node_c", async () => ({ result: "C" }))
  .addEdge(START, "node_a")
  .addEdge("node_b", END)
  .addEdge("node_c", END)
  .compile();

Send によるファンアウト：条件付きエッジから `[Send("worker", {...})]` を返すと、並列ワーカーを起動できます。結果フィールドにはリデューサーが必要です。複雑なワークフローでは、Send と実行パターンを深く理解することが不可欠です。 無限ループを防ぐため、条件付きエッジで END に進むようにします。 ```python from langgraph.graph import StateGraph, START, END

class State(TypedDict): count: int max_iterations: int

def increment(state: State) -> dict: return {"count": state["count"] + 1}

def should_continue(state: State) -> str: if state["count"] >= state["max_iterations"]: return END return "increment"

graph = ( StateGraph(State) .add_node("increment", increment) .add_edge(START, "increment") .add_conditional_edges("increment", should_continue) .compile() )

result = graph.invoke({"count": 0, "max_iterations": 5}) print(result["count"]) # 5

</python>
<typescript>
条件関数から END を返してループを終了します。
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  count: z.number().default(0),
  maxIterations: z.number(),
});

const increment = async (state: typeof State.State) => {
  return { count: state.count + 1 };
};

const shouldContinue = (state: typeof State.State) => {
  if (state.count >= state.maxIterations) return END;
  return "increment";
};

const graph = new StateGraph(State)
  .addNode("increment", increment)
  .addEdge(START, "increment")
  .addConditionalEdges("increment", shouldContinue)
  .compile();

const result = await graph.invoke({ count: 0, maxIterations: 5 });
console.log(result.count);  // 5

実行可能なグラフを取得するには、compile() が必要です。 ```python # WRONG builder.invoke({"input": "test"}) # AttributeError!

CORRECT

graph = builder.compile() graph.invoke({"input": "test"})

</python>
<typescript>
実行可能なグラフを取得するには、compile() が必要です。
```typescript
// WRONG
await builder.invoke({ input: "test" });

// CORRECT
const graph = builder.compile();
await graph.invoke({ input: "test" });

無限ループを避けるため、END への条件付き経路を用意します。 ```python # WRONG: Loops forever builder.add_edge("node_a", "node_b") builder.add_edge("node_b", "node_a")

CORRECT

def should_continue(state): return END if state["count"] > 10 else "node_b" builder.add_conditional_edges("node_a", should_continue)

</python>
<typescript>
END を返す条件付きエッジを使ってループを終了します。
```typescript
// WRONG: Loops forever
builder.addEdge("node_a", "node_b").addEdge("node_b", "node_a");

// CORRECT
builder.addConditionalEdges("node_a", (state) => state.count > 10 ? END : "node_b");

その他のよくある間違い： ```python # Router must return names of nodes that exist in the graph builder.add_node("my_node", func) # Add node BEFORE referencing in edges builder.add_conditional_edges("node_a", router, ["my_node"])

Command return type needs Literal for routing destinations (Python)

def node_a(state) -> Command[Literal["node_b", "node_c"]]: return Command(goto="node_b")

START is entry-only - cannot route back to it

builder.add_edge("node_a", START) # WRONG! builder.add_edge("node_a", "entry") # Use a named entry node instead

Reducer expects matching types

return {"items": ["item"]} # List for list reducer, not a string

```typescript
// Always await graph.invoke() - it returns a Promise
const result = await graph.invoke({ input: "test" });

// TS Command nodes need { ends } to declare routing destinations
builder.addNode("router", routerFn, { ends: ["node_b", "node_c"] });

### 設定できること

• TypedDict/StateSchema でカスタム状態スキーマを定義する
• リデューサーを追加して、状態の更新をどのようにマージするか制御する
• ノード（任意の関数）を作成し、静的エッジ／条件付きエッジを追加する
• Command を使って状態の更新とルーティングをまとめる
• 条件付きの終了処理を備えたループを作成する

設定できないこと

• START/END の動作を変更する
• ノード関数の外部から状態にアクセスする
• 状態を直接変更する（部分更新の辞書を返す必要があります）