プロダクションで稼働しているAI機能のフレームワークをLangGraphに完全移行しました
株式会社エブリー / uho-wq
メンバー / バックエンドエンジニア
導入の背景・解決したかった問題
導入背景
ツール導入前の課題
導入前はフレームワークを使用せず、フルスクラッチでワークフローを実装していました。
これにより大きく2つの課題を感じていました。
1つ目は、アーキテクチャの責務が混同している点です。
コアとなるビジネスロジックやLLMの呼び出し処理は、再利用可能な形で関数化されています。しかし、ビジネスロジックを呼び出すワークフローや、処理の過程で変化するデータ構造の状態管理の責務が明確に分かれていなかったため、追加・修正実装の負荷が高い状態に陥っていました。
特にワークフローに関してはレスポンス速度の向上のために処理を並列化していたりするので、流れが特に追いにくい状態でした。またユーザー入力や処理の出力に対して分岐処理をする実装もあり、ワークフロー自体がロジックを持ってしまうことも問題となっていました。
2つ目は、変更の柔軟性がないことでした。
アーキテクチャが明確に責務分けされていないことによって、特定の処理順序を置き換える場合や並列化を行う場合に、どの状態をどのように置き換えるべきかが分かりづらい状態になっていました。
ワークフローを変更する場面が増えている状況では、この辺りの変更負荷の高さが実装上のストレスにもなっていました。
したがって課題をひとことでまとめると、フルスクラッチで実装する限界がということになります。
どのような状態を目指していたか:
明確な責務分離が可能な状態
課題から、ワークフロー・状態管理・ビジネスロジックの責務を分離できるような仕組みにする必要がありました。
責務が明確に分かれることで変更に強いアーキテクチャになると考えたためです。
ワークフローの変更柔軟性・拡張性
責務分離の話と重複しますが、ワークフローの責務が明確化されることによって状態を意識することなく(ワークフローの処理動作の依存関係は考慮する必要があります)順序の組み替え、並列化などに対応できることが望ましいです。
プロバイダに依存しない
フレームワークによっては、ある機能が特定のサービスプロバイダに依存している実装など見られたため、次に示す置き換えやすさという観点でも依存しない仕組みを採用することが長期的に運用する上で大事だと考えます。
プロバイダの置き換えやすさ
プロバイダ依存せずとも置き換えが困難だった場合、置き換えやすくするにはプロバイダに依存しないインターフェースを別で定義しなければなりません。
しかしながら、ここは独自にインターフェースを定義すれば済む話ではあるので必須とはしませんでした。
型制約(Structured Outputのため)
構造化データをLLMに出力させるためには型制約が必須です。
移行前の実装でもPydanticを用いて型安全な出力になるよう実装していたため、ここは引き続き必須要件としました。
テストのしやすさ
LLMによる出力は決定論的に評価できないため、別途LLM as a Judgeによる品質評価やプロダクションから構築したデータセットを用いた評価・改善のループを回す必要があります。
しかし、LLMのテストに関してはソースコードとは切り離して考える必要があると思うので、LLMに関わる処理をMock化することでLLM以外の処理に関してはテスト可能となります。
責務を明確に分け、LLMをMock化することでルールベースの処理や状態の流れなどに対するユニットテストが比較的簡単に書けるようになるため、テストのしやすさも要素として取り入れました。
比較検討したサービス
- ADK
- Genkit
- LangChain
- Pydantic AI
比較した軸
目指したい状態に対して、どれだけマッチしているかです。
導入の成果
改善したかった課題はどれくらい解決されたか
LangGraphは目指している状態のすべての条件にマッチしていたので、改善したかった課題はすべて解決できました。
どのような成果が得られたか
ワークフローとビジネスロジックで明確な責務分離が可能となったことで、ビジネスロジックを修正する場合にワークフローを意識しなくてよくなったのは開発生産性に大きく寄与しています。
また、プロバイダをまたいだモデルの変更なども移行したことによって簡単に行うことができるようになりました。
導入に向けた社内への説明
上長・チームへの説明
フレームワークはOSSなので実質かかるコストは人件費だけでした。また別案件ですでに導入実績があった背景もありました。
過去の導入を踏まえた上での実装にすることで0→1よりも早い移行が可能であること、現状のままだとなぜだめなのか、LangGraphを導入するとなにが良いかを説明したことで結果的に移行のための時間を確保することができました。
活用方法
よく使う機能
1. StateGraph
グローバルに適用する状態の登録と、グラフを構築するためのNode、Edgeを登録するクラス
from langgraph.graph import END, START, StateGraph from langgraph.graph.state import CompiledStateGraph
from my_app.nodes import NodeA, NodeB
from my_app.types.state import State
class MainWorkflow:
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:
"""
グラフを構築する。
LangGraphでは、ノード(処理ステップ)とエッジ(遷移)を定義することで
ワークフローの実行順序を決定する。
本グラフの構造:
START ─┬─> step_a ─┬─> END
└─> step_b ─┘
- START: グラフの開始点
- END: グラフの終了点
- 同じノードから複数のエッジを張ることで並列実行が可能
"""
g = StateGraph(State)
# ノードの登録
# 各ノードは状態(State)を受け取り、更新する辞書を返す関数
g.add_node("step_a", self.step_a)
g.add_node("step_b", self.step_b)
# エッジの定義(実行順序の決定)
# STARTから両ノードへエッジを張ることで、step_aとstep_bが並列実行される
g.add_edge(START, "step_a")
g.add_edge(START, "step_b")
# 両ノードからENDへエッジを張り、全ノード完了後にグラフが終了する
g.add_edge("step_a", END)
g.add_edge("step_b", END)
# グラフをコンパイルして実行可能な形式に変換
return g.compile()
2. The graph API - PNG
Mermaid.Ink の API を使用してグラフを可視化する場合に使用
from IPython.display import Image, display
from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles
display(Image(app.get_graph().draw_mermaid_png()))
ツールの良い点
- 明確な責務分離が可能
- ワークフローの変更柔軟性、拡張性
- プロバイダに依存しない
- プロバイダを置き換えやすい
- 型制約(Structured Outputのため)
- テストがしやすい
ツールの課題点
- フレームワークに癖があるのでキャッチアップが大変
株式会社エブリー / uho-wq
メンバー / バックエンドエンジニア
よく見られているレビュー
株式会社エブリー / uho-wq
メンバー / バックエンドエンジニア
レビューしているツール
目次
- 導入の背景・解決したかった問題
- 活用方法



