この記事は every Tech Blog Advent Calendar 2025 の 13 日目の記事です。
こんにちは、開発1部に所属している25卒の岩﨑です。
本記事では、プロダクションで稼働しているAI機能のフレームワークをLangGraphに完全移行したことについてお話しします。
背景
デリッシュキッチンでは、会話形式でレシピ検索できるデリッシュAIという機能を提供しています。 推論にはWorkflow型を採用しており、LLMのAPIをフルスクラッチの関数パイプラインから呼び出すような実装となっています。
機能を公開して1年以上が経過し、様々な機能追加や改善サイクルが回された結果、ワークフローが複雑になりいくつかの問題が発生しました。
1つはアーキテクチャの責務が混同している問題があります。
コアとなるビジネスロジックやLLMの呼び出し処理は、再利用可能な形で関数化されています。
しかしビジネスロジックを呼び出すワークフローや、処理の過程で変化するデータ構造の状態管理の責務が明確に分かれていなかったため、追加・修正実装の負荷が高い状態に陥っていました。
特にワークフローに関してはレスポンス速度の向上のために処理を並列化していたりするので、流れが特に追いにくい状態でした。 またユーザ入力や処理の出力に対して分岐処理をする実装もあり、ワークフロー自体がロジックを持ってしまうことも問題としてありました。
2つ目は、変更の柔軟性がないことです。
アーキテクチャが明確に責務分けされていないことによって、特定の処理順序を置き換える場合や並列化を行う場合に、どの状態をどのように置き換えるべきかがわかりづらい状態になっていました。
新しい施策や精度改善のために頻繁にワークフローを変更する場面が増えてきたこともあってこの辺りの変更負荷が高いことは実装する上でもストレスに感じるようになりました。
したがって課題をひとことでまとめると、フルスクラッチで実装する限界がきたということになります。
今後、WorkflowとAgentの混合型を実装するであったり、コンテキストエンジニアリングに関わる実装を追加する場合に、現状の実装では複雑になりすぎてリリースとともに開発スピードが低下していくことは自明でした。
これらの課題に対処するためにワークフローを可視化したものをドキュメントとして用意していましたが、精度を改善する過程では様々な角度で試してよかったものを採用するようなサイクルを回すことが多い都合上、更新のたびにワークフローが変化していき、ちょっとでもドキュメントを放置したら陳腐化してしまうという課題もありました。
課題解決のために
これらの課題解決を考える上で、技術観点はもちろんですが、それだけでなくプロダクト観点でも長期的に運用できるようなアーキテクチャであることが理想です。
ここからは先ほどあげた課題を解決するための要素をいくつか並べていきます。
明確な責務分離が可能
課題から、ワークフロー・状態管理・ビジネスロジックの責務を分離できるような仕組みにする必要がありました。
責務が明確に分かれることで変更に強いアーキテクチャになると考えたためです。
ワークフローの変更柔軟性、拡張性
責務分離の話と重複しますが、ワークフローの責務が明確化されることによって状態を意識することなく（ワークフローの処理動作の依存関係は考慮する必要があります）順序の組み替え、並列化などに対応できることが望ましいです。
また、ReActであったりWorkflowとAgentの混合型、コンテキストエンジニアリングといった追加要素に関する要件が出てきた場合も、各処理が疎結合であることによって容易に拡張できることも期待しています。
プロバイダに依存しない
今日最高性能だったモデルが数日後にはそうではなくなっている時代で、特定のプロバイダに依存するような仕組みはできるだけ避けたいところです。
フレームワークによっては、ある機能が特定のサービスプロバイダに依存している実装など見られたため、次に示す置き換えやすさという観点でも依存しない仕組みを採用することが長期的に運用する上で大事なのではないかと思います。
プロバイダの置き換えやすさ
ここはどちらかというと必須というよりは理想に近い話です。
プロバイダ依存せずとも置き換えが困難だった場合、置き換えやすくするにはプロバイダに依存しないインターフェースを別で定義しなければなりません。
しかしながらここは独自にインターフェースを定義すれば済む話ではあるので必須とはしませんでした。
型制約（Structured Outputのため）
構造化データをLLMに出力させるためには型制約が必須です。
しかしながら、型制約はOpenAIやGoogleが提供するAPIではかなり前から対応されています。
Anthropicにおいても2025-11-14にClaudeのStructured Outputに対応したため、主要プロバイダが提供するAPIにおいてはPydanticと併用することによって型安全な出力が可能となりました。
移行前の実装でもPydanticを用いて型安全な出力になるよう実装していたため、ここは引き続き必須要件となります。
テストのしやすさ
LLMによる出力は決定論的に評価できないため、別途LLM as a Judgeによる品質評価やプロダクションから構築したデータセットを用いた評価・改善のループを回す必要があります。
しかしLLMのテストに関してはソースコードとは切り離して考える必要があると思うので、LLMに関わる処理をMock化することでLLM以外の処理に関してはテスト可能となります。
責務を明確に分け、LLMをMock化することでルールベースの処理や状態の流れなどに対するユニットテストが比較的簡単に書けるようになるため、テストのしやすさも要素として取り入れました。
LangGraphの採用
現時点において、これらの全ての要素にマッチするのがLangChainおよびLangGraphでした。
2025-10-22のタイミングでv1.0がリリースされ、マイナーバージョンやパッチバージョンでは破壊的変更が行われないことが明記されていることも決め手の1つです。
ディレクトリ構成
ここからはLangGraph適用後のディレクトリ構成について紹介します。
/ ├── docs ├── src │ ├── const │ ├── nodes │ ├── schema │ │ └── filters │ ├── types │ │ └── state │ ├── utils │ ├── workflows │ │ ├── main_workflow.py │ │ └── sub_workflows │ └── evaluation └── tests
※ 一部わかりやすさのために省略している部分があります
ここではworkflows, nodes, types/state の3つについて取り扱います。
LangGraphでは、LLMの処理を表すNodeとLLMの処理同士をどう繋ぐかを表すEdgeで整理され、大域的に扱う変数をStateとして定義することで、Workflowを組むことができます。
Stateの定義
Stateはワークフロー全体で共有される状態を表しています。
TypedDictを継承して定義することで、型安全な状態管理が可能になります。
例えば、入出力のStateは以下のように定義できます。
# types/state/input_output.py from typing import TypedDict class InputState(TypedDict, total=False): """入力スキーマ""" user_input: str options: dict class OutputState(TypedDict, total=False): """出力スキーマ""" result: str metadata: dict
Node の定義
Nodeは単一責務の処理ロジックを担当します。
ビジネスロジックは基本的にこのNodeに記述するようにしています。
class NodeA: """ノードA - 特定の処理を担当""" def __init__(self): self.llm = ChatOpenAI(model="gpt-5-mini", timeout=5.0, max_retries=2) def execute(self, input_data: str) -> dict: """処理を実行""" processed = input_data.upper() return {"field_a": processed}
サブワークフローの定義
サブワークフローは関連するNodeをまとめたグラフです。
処理の論理的なまとまりごとにサブワークフローを作成することで、再利用性とテスト容易性が向上します。
from langgraph.graph import END, START, StateGraph from langgraph.graph.state import CompiledStateGraph from my_app.nodes.phase1 import NodeA, NodeB from my_app.types.state import State class Phase1Workflow: """Phase1サブワークフロー""" def __init__(self): self.node_a = NodeA() self.node_b = NodeB() def step_a(self, state: State) -> dict: """ステップA""" user_input = state["user_input"] result = self.node_a.execute(user_input) return {"field_a": result["field_a"]} def step_b(self, state: State) -> dict: """ステップB""" user_input = state["user_input"] result = self.node_b.execute(user_input) return {"field_b": result} def build_graph(self) -> CompiledStateGraph: """グラフを構築""" g = StateGraph(State) g.add_node("step_a", self.step_a) g.add_node("step_b", self.step_b) # 並列実行 g.add_edge(START, "step_a") g.add_edge(START, "step_b") g.add_edge("step_a", END) g.add_edge("step_b", END) return g.compile()
各stepは状態管理を責務としてもち、build_graph メソッドがワークフローの責務を持っています。
build_graphでは、StateGraph を使ってノードを追加し、add_edge でノード間の依存関係を定義しています。
上記の例では START から step_a と step_b の両方にエッジを作成することで、2つの処理が並列実行されます。
このように、並列化は単にエッジの向き先を変えるだけで実現できるところも魅力的な部分です。
メインワークフローの定義
メインワークフローは複数のサブワークフローを組み合わせることで作成されるエンドツーエンドのパイプラインです。
# workflows/main_workflow.py from langgraph.graph import END, START, StateGraph from langgraph.graph.state import CompiledStateGraph from my_app.types.state import InputState, State from my_app.workflows.sub_workflows import ( Phase1Workflow, Phase2Workflow, Phase3Workflow, ) class MainWorkflow: """メインワークフロー""" def __init__(self): self.phase1 = Phase1Workflow() self.phase2 = Phase2Workflow() self.phase3 = Phase3Workflow() def build_graph(self) -> CompiledStateGraph: """メインワークフローを構築""" g = StateGraph(State, input=InputState) # サブワークフローをノードとして追加 g.add_node("phase1", self.phase1.build_graph()) g.add_node("phase2", self.phase2.build_graph()) g.add_node("phase3", self.phase3.build_graph()) # 順次実行: phase1 → phase2 → phase3 g.add_edge(START, "phase1") g.add_edge("phase1", "phase2") g.add_edge("phase2", "phase3") g.add_edge("phase3", END) return g.compile() def run(self, user_input: str, options: dict = None) -> dict: """ワークフローを実行""" initial_state: State = { "user_input": user_input, "options": options or {}, "field_a": "", "field_b": [], "field_c": [], "field_d": 0.0, "field_e": "", "result": "", "metadata": {}, } graph = self.build_graph() return graph.invoke(initial_state)
StateGraph に input=InputState を渡すことで、外部から渡される入力の型を制限できます。
また、サブワークフローの build_graph() の戻り値をそのままノードとして追加できるため、メインのワークフローが肥大化することなく実装できます。
アーキテクチャの責務分離
以上のような設計により、以下のように責務を分離することができました。
| 層 | 責務 |
|---|---|
| types/state/ | 状態の型定義 |
| nodes/ | ビジネスロジック |
| workflows/sub_workflows/ | サブワークフローのグルーピング |
| workflows/main_workflow.py | 全体のフロー制御 |
ワークフローの順序変更や並列化は workflows/ 内のエッジの向き先を変えるだけで完結し、ビジネスロジック（nodes/）に影響を与えません。
逆に、特定の nodes/ の処理内容を変更しても、workflows/ には影響しません。
以上により、明確な責務の分離と変更にも耐えうる柔軟なコードベースが構築できたのではないかと思います。
またプロバイダの変更においても、LangChainではParter packagesとしてプロバイダごとのpackageを公開しているため、適用したい箇所のライブラリを変更するだけで置き換えが可能です。
- OpenAI：langchain-openai
- Anthropic：langchain-anthropic
- Google：langchain-google-genai
おわりに
本記事ではフルスクラッチの関数型パイプラインによって稼働していたコードベースをLangChain, LangGraphに移行する意思決定の過程について説明しました。
最初からLangGraphにすべきだったかという問いに対しては必ずしもそうとは言えません。
フレームワークは実際に使ってみないと知見が蓄積しないので試す姿勢は大事だと思います。
とはいえ内部処理をちゃんと理解しないままフレームワークに依存するのは危険なので、どのようなフレームワーク思想の元で、なにを解決したいのかを考えるべきなのかなと今回のリファクタリングを通して思いました。
エブリーでは、ともに働く仲間を募集しています。
テックブログを読んで少しでもエブリーに興味を持っていただけた方は、ぜひ一度カジュアル面談にお越しください！
