ユーザーが使うほど成長するAIエージェントを作る（LangChain・DSPy合体編）
はじめに
こんにちは、今回ヌーラボブログリレー2025年冬、14日目Tech記事を担当させていただきます、インターン生のぽるしぇと申します。
今回はどのような内容を書いていくか迷いましたが、業務とは関係なく個人的に興味のあるAIエージェント開発について書いていこうと思います。
というのも、個人的にAIエージェント関連でやってみたかったことがあったのですが、ずっと先延ばしにしていたものがあったのを1週間前に思い出してしまい、いてもたってもいられずチャレンジしてしまいました。
そのため他のネタを全く用意できていないのですが、幸運なことに書く内容が多くなりそうなため、この奮闘記を２,３回に分けて書いていこうと思います。今回は記念すべき第１回目ということで、ぜひ最後まで読んでくださるとありがたいです。
全体の概要
私は今までAIエージェントを個人で作っていた際、とあることに悩まされ、苦手意識を持っていました。それは「プロンプトエンジニアリング」です。
そこで、プロンプト自身を改善していくAIエージェントシステムを作ってみたいと思い、システムを考案、開発していき、忘れないために記録していきます。
第1回の概要
AIエージェント開発でほど避けては通れないのが「プロンプトエンジニアリング」です。
「プロンプトエンジニアリング」というのは、それ専用のエンジニアがいるほどに奥が深く、また必ずと言って良いほどメンテナンスが必要で、属人化しやすく、個人の技術力に依存してしまう側面があると考えています。
この課題はコードベースでのAIエージェント作成フレームワークであるLangChainでもよくみられる課題だと考えています。
そしてこのLangChain自身も個人の技術力にそこそこ依存すると私は思います。LangChainは豊富なエコシステムによる書き方の柔軟さがあるため、システムをスケーリングしていくほど複雑化していきやすいからです。
少し前、プロンプトエンジニアリングの負担を軽減させることができる、DSPyという新しいAIエージェントフレームワークが流行しました。
これは機械学習のアイデアに基づいた設計思想を持っており、従来のLangChainよりもよりプログラマティックにAIを制御できる、個人的に面白いフレームワークです。
しかし、現時点では実製品への適用という点ではいささかエコシステムが未熟であると言われていたりします。
また、LangChainとDSPyはそれぞれ違う思想のもと生まれたAIエージェントフレームワークであり、ネットの受け売りですが組み合わせるには互換性がないという評価らしいです。2つは根本的に異なるフレームワークなのです。
そこで、従来問題とされていたLangChainとDSPyの互換性のなさを、LangChainのグラフの構造を見直すことで2つを組み合わせ、
LangChainの使いやすさとDSPyの自動プロンプト最適化を組み合わせることで、「使うほどユーザの期待にあった回答を出せるようになるAIエージェント」を手軽に作ってみることにしました。
今回はその記録を第一歩としてメモにして残していきます。
自身が感じたAIエージェントの課題
属人化するプロンプトと高まるメンテナンスコスト
私の従来のAIエージェント開発では、以下のような課題がありました。
- プロンプトエンジニアリングの属人性: 良いプロンプトを作るには専門知識と時間が必要
- 脆弱性: データの微妙な差異やモデルの更新によって、期待する精度が容易に低下してしまう
- スケーラビリティの問題: システムが複雑になるほどメンテナンスが困難に
私はプロンプトエンジニアリングに対して、
- 期待した回答を得られるまでひたすら試行錯誤を繰り返す泥臭さ
- それでもモデルやデータが変わるだけで精度が簡単に変わってしまう不安定さ
- ユーザが実際に投げる文を想定しても実際には予想とずれたりする時の虚無感
- そして実際に使用されるユーザー入力の多様性に対応させる必要があることへの適応コストの高さ
を感じていました。（ほとんどめんどくさがり屋なのが原因な気がしますが...）
これは、実際に私が体験したことで、仲間と苦労の末にやっとうまくできたプロンプトを、半年後に別のAIエージェントで転用すると名無しのゴンベさん（またはJohn Doeさん）が大量出現して絶望した記憶が頭から離れないからだと思います。
また、私が過去に構築したAIエージェントを見てみると、LangChainの豊富なエコシステムがゆえに、依存関係の規則が曖昧になりがちで、無理矢理な書き方でも実装できてしまい、結果複雑化してしまうといった所感がありました。
これらは私の技術力が足りないからではあるのですが、これこそAIエージェント作成は個人の技術力に大きく依存すると考えています。
そこで、LangChainによるAIエージェントの構造を見直しつつ、プロンプトをシステム自身に改善させることでAIを改善させるアプローチを取ろうと思います。
なぜDSPyなのか
なぜファインチューニングではないのか
ファインチューニングというのは、特定のタスクやデータセットに合わせて、事前学習済みモデルを再学習させることで、そのタスクに対する性能を高める（調整する）手法です。これだけ聞くとファインチューニングで良さそうに聞こえますが、今回はファインチューニングの手法だと実運用上、以下の問題が考えられます。
- 大規模な教師データの不足
多くの場合、ファインチューニングのデータセットには最低でも1000以上のデータが必要とされています。これは、すでに大量のパラメータを持つ事前学習モデルの性能を効果的に調整しようとするためです。
このため、AIエージェントにおいては、ユーザーのリアルな対話データを十分な量収集することが課題になってきます。 - 正解データの作成コスト
ユーザのリアルな質問データに対し、それと同数の高品質な正解データを用意する必要があります。これはいったい誰が用意するのでしょうか（すっとぼけ）
それに対しDSPyによるプロンプト最適化のアプローチは、比較的小量のデモデータで済みます。
もちろん、ファインチューニングも精度向上の面では効果的ですし、どうやらプロンプト最適化と相性も良いらしいです。
実際にプロンプト最適化とファインチューニングを交互に行うことが効果的だという研究結果もあるようです。
プロンプトエンジニアリングの本質
まず、プロンプトエンジニアリングとは何かを理解する必要があります。
まず、今回我々が使うLLM（Large Language Model）とは、自然言語モデルの一種です。
過去いくつもの自然言語モデルが開発され、今はLLMをはじめとした自己回帰モデル(Transformer)が主流ですが、実は取られていたアプローチはずっと変わりません。
それは、入力されたコンテキストに基づき、確率的に次に来る単語を予測して出力することです。
LLMも同様で、ざっくり言うと、大規模なテキストデータから単語間の関係性を分散表現として保存し、このベクトルを用いて次に最も来やすい単語を選んでいるのです。
プロンプトとは、この出力を望ましい方向へ誘導する（促す=prompt）ための入力テキストです。
プロンプトを重みとして捉える
DSPyの重要な洞察は、プロンプトをニューラルネットワークの重みと同様に扱うことです。従来の機械学習では、学習データとメトリクスを与えてモデルの重みを最適化します。DSPyは同様のアプローチで、学習データとメトリクスからプロンプトを自動最適化します。
参考論文：DSPy: Compiling Declarative Language Model Calls into Self-Improving Pipelinesでは、この手法により従来の手動プロンプティングを25-65%上回る性能を実現したと報告されています。
DSPyについて、個人的にわかりやすい記事がいくつかありましたのでここで紹介させていただきます。
LangChainとは
LangChain vs DSPy の対立構造を超えて
インターネット上では「LangChain vs DSPy」といった対立構造の記事も見られますが、実際には両者の強みを組み合わせることで、より良いソリューションを構築できます。
LangChainで快適な開発をしながらDSPyにプロンプトを任せたい
- DSPyによる宣言的な回答生成と自動プロンプト最適化
- LangChainの豊富なエコシステムによる柔軟な開発
- 両者の利点を活かした持続可能なシステム設計
これからこれを実現していこうと思います。
LangChainとDSPyの互換性問題
LangChainは豊富なエコシステムと開発利便性を提供する一方、DSPyは自動的なプロンプト最適化を実現します。
しかし、これら2つのフレームワークは従来、互換性の問題から同時に使用することが困難でした。
そこで、LangChainの構造を自分なりに見直し、LangChain作成時の個人的にルールを作ってみました。
これで、個人的にLangChainとDSPyの互換性問題がかなり解消できたと感じています。
LangGraphの4層アーキテクチャ
具体的には、以下のLangGraphの4層アーキテクチャを定義しました。
┌─────────────────────────┐
│ Service Layer │ ← 外部システムとの接続
├─────────────────────────┤
│ Agent Layer │ ← AIの推論ロジック (DSPy導入可能)
├─────────────────────────┤
│ Node Layer │ ← システムの状態管理
├─────────────────────────┤
│ Graph Layer │ ← フロー制御
└─────────────────────────┘
具体的にみていきましょう。
4層アーキテクチャによる責務の分離
最初に
At its core, LangGraph models agent workflows as graphs.
LangGraphはエージェントのワークフローをグラフとしてモデル化するLangChainの互換フレームワークです。LangGraph以下3つの要素から構成されます。
- State
アプリケーションの現在のスナップショットを表す共有データ構造。 - Nodes
エージェントのロジックをエンコードする関数です。現在の状態を入力として受け取り、何らかの計算や副作用を実行し、更新された状態を返します。 - Edges
現在の状態に基づいて次に実行するものを決定する関数。条件分岐または固定遷移のいずれかになります。
これを以下のように責任を分離させました。
各層の責務：
- Service Layer: 外部システム（API、データベース等）への干渉
- Agent Layer: 入力に対する推論処理
- Node Layer: 状態の管理・サービス/エージェントの呼び出し
- Graph Layer: グラフ構造の管理。システムそのものの統制。
このAgent層にDSPyで作成したエージェントを配置することで、LangChainのエコシステムを残しながら、DSPyの自動プロンプト最適化の恩恵を受けることができます。
さらに詳しくみていきましょう。
サービス層
このレイヤーは唯一システムの外部へのアクセスできます。
しかし、単体エージェントの記憶やLangChainによるシステムが一貫して保持する状態（記憶）を一切知りません。
要するに、このレイヤーの役割は、呼ばれたた指定の処理をするだけの関数のようなものです。
できること
- システム外部への干渉
- エージェントへの応答
- ノードへの応答
できないこと
- エージェントの呼び出し
- ノードの呼び出し
- グラフの構造への干渉
- グラフの状態への干渉
エージェント層
システムのビジネスロジック、つまり価値を生んでいる２層のうちの1つです。
しかしレイヤーの役割はすごく単純で、ノードから受け取った文章に回答するだけです。
ただし、AIによる自律的な動作により、従来のプログラムのように完全に制御することは難しいです。完全に制御したいならノード層に任せるべきです。
このレイヤーはシステムの記憶を一切知りません。
また、システムの外部の情報も、ツールを介して得られた情報しか知りません。
そして、ツールはノード層からの指定があるものしか使えません。
要するに、このレイヤーの役割は、ただ質問文を受け取り、回答文を返すだけの単純なLLMです。
できること
- ノードへの応答
- LLMによる回答生成
- ツールの利用（ノードからの許可時のみ）
できないこと
- システム外部への干渉
- ノードの呼び出し
- グラフの構造への干渉
- グラフの状態への干渉
ノード層
システムのビジネスロジック、つまり価値を生んでいる２層のうちの1つです。私は最も大事だと考えています。
このレイヤーはシステムの状態を、任意の単位で操作ができます。
各ノードはプログラム通りにAgent,Serviceを駆使して情報を整理し、システムの状態（記憶）を構築・更新していきます。
Graph全体の状態は、この各Nodeが作った小さな状態の結合したものと捉えることができます。
できること
- グラフの状態への干渉
- エージェントの呼び出し
- サービスの利用
できないこと
- システム外部への干渉
- グラフの構造への干渉
グラフ層
AIエージェントシステムそのものです。
システムの分岐ロジックはここで制御します。
Nodeに状態を割り当てますが、Graph自体は状態を操作しません。
記憶は見えるけど、中身には干渉しないのです。
できること
- ノードの呼び出し
- グラフの状態の結合・分離
- グラフの構造への干渉
できないこと
- システム外部への干渉
- エージェントの呼び出し
- サービスの呼び出し
- グラフの状態への干渉
まずはLangChainで実装していきましょう。
サービス層
サービス層はただの関数なので例は割愛します。
LangChainが用意したToolだったり、外部APIを叩く関数だったり、DBのリポジトリ層でのモジュールを操作する関数でもなんでも良いです。
大事なのは外部システムに干渉する処理を全てこの層に閉じ込めること、
そしてグラフ内部のロジックや状態には一切干渉しないことです。
エージェント層
エージェント層は、与えられた入力テキストに対し回答を生成する権利のみを持っています
次回の記事では変わりますが、まずはシステムプロンプトもインスタンス作成時に指定しましょう。
ツール（サービス層）の利用可否は、上位のノード層に制御させます。
class BaseAgent:
def __init__(
self,
agent_type: AgentType,
model: Callable,
system_prompt: str,
) -> None:
self.AgentType = agent_type
self.Model = model
self.SystemPrompt = system_prompt
def call(
self, input_text: str,
tool_list: list[Callable] | None = None
) -> dict[str, Any]:
self.Agent = create_agent(
name=self.AgentType,
model=self.Model(),
system_prompt=self.SystemPrompt,
tools=tool_list
)
response = RetryWrapper(self.Agent).invoke({"user_input": input_text})
return {
"output": response["messages"][-1].content,
"assignee": self.AgentType
}
class ScheduleAgent(BaseAgent):
"""
ScheduleAgent
-------------
- Agent for Suggesting Time Schedules which are necessary for user's visit.
"""
def __init__(self) -> None:
super().__init__('Schedule',
basic_openai,
SCHEDULE_PROMPT)
このように、入力に対して回答を生成するだけにします。
注意
LangChainにはMessagesStateとCommandという非常に強力な機能があります。
MessagesStateは、グラフ内で
{"messages": ~}
の形のデータを返すと、グラフ全体の状態をに追加できるのです。
Commandは、
Command(
update=~
goto=~
)
このように書くと、updateでグラフ全体の状態をに追加できますし、gotoで次のノードの行き先も定義できるのです。
当然サービスやエージェントでもretunに続けて書くだけで使うことができます。
例えばエージェントでCommandを使いエージェントに直接グラフの状態を更新することもできますし、エージェントにgotoコマンドを返すツールを使わせてグラフの遷移を操作させることもできるでしょう。
しかしこれはLangChain特有の機能です。DSPyにはありません。
それにAgent層にグラフの構造を書くと、グラフの構造が複雑になるほど全体像が隠されてしまします。
エージェント層では使わない方が良いでしょう。(今のところ使っても良いとしたらError時に決めたエージェントに返すぐらいだと考えていますが、ここは書き方次第でノードでできると考えています。)
ノード層
Node層ではエージェントとサービスを呼び出せます。
そしてグラフの状態を更新できます
class BaseNode():
"""
BaseNode
--------
- Provides GraphState handling.
- You can Extend this BaseNode for specific Agent Nodes.
- at Extension, you should Update process Method.
Methods
-------
- extract_messages: Extract messages from GraphState
- update_state: Update State with Agent output
- process: you should Update process Method in Extended Node
"""
def extract_messages(self, state: GraphState) -> str:
return str([message.content for message in state["messages"]])
def update_state(self, output: str) -> Command:
return Command(
update={
"messages": [AIMessage(content=json.dumps(output, ensure_ascii=False))]
}
)
def process(self, state: GraphState) -> GraphState:
pass
class ScheduleNode(BaseNode):
"""
ScheduleNode
------------
- Node for Schedule Purpose Tasks
- Based on BaseNode
- Call ScheduleAgent Process.
Methods
-------
- process: Call ScheduleAgent with extracted user input and update GraphState
"""
def __init__(self) -> None:
self.node_type: NodeType = 'Schedule'
self.Agent = ScheduleAgent()
def process(self, state: GraphState) -> Command:
try:
user_input = self.extract_messages(state)
agent_output = self.Agent.call(user_input, [tavily_research_tool])
return self.update_state(agent_output)
except Exception as e:
error_message = f"Error in {self.node_type} Node processing: {str(e)}"
return self.update_state(error_message)
このように、エージェントを呼び出すこともできますし、サービスを呼び出すこともできます。
エージェントとサービスを駆使し、システムの状態を更新していきます。
グラフ層
グラフ層では、ノードを接続し、システム全体の分岐ロジックと遷移を定義することで、AIエージェントシステムを完成させます。
グラフの構造を全てここに集約することで、グラフ全体の構造をこの層だけで把握・管理できるようになります。
今回はシンプルなSupervisor型エージェントを例示します。
Supervisor型エージェントは、タスクをどの専門エージェントに割り振るかを管理する構造を持ち、動的なエージェントルーティングを可能にします。
昔はここにわかりやすい図があったのですが今はなくなってそうです。幸い他の記事でも使っていたので参照させていただきます。
なお、LangChain公式では今はこれが一番わかりやすそうです。他にもマルチエージェント構造は種類がありますので、詳しく知りたい方は、ぜひ以下のリンクをご参照ください。
それでは早速実装しましょう。
class TouristAgentGraph:
"""
TouristAgentGraph
-----------------
Graph for Multi-Agent Collaboration about Supporting Tourist
"""
def __init__(self) -> None:
self.State = GraphState
def assign_node_types(self, state: GraphState) -> NodeType:
messages = state.get("messages", None)
if not messages:
return 'General'
next_agent = json.loads(messages[-1].content).get("next_agent", None)
if not next_agent:
return 'General'
if next_agent == 'Trip':
return 'Trip'
elif next_agent == 'Schedule':
return 'Schedule'
elif next_agent == 'Task':
return 'Task'
return 'General'
def build_graph(self) -> CompiledStateGraph:
# --- Init Graph ---
graph = StateGraph(self.State)
# --- Nodes ---
graph.add_node('Supervisor', SupervisorNode().process)
graph.add_node('General', GeneralNode().process)
graph.add_node('Trip', TripNode().process)
graph.add_node('Schedule', ScheduleNode().process)
graph.add_node('Task', TaskNode().process)
# --- Edges ---
# from START to Supervisor
graph.add_edge(START, "Supervisor")
# from Supervisor to General / Trip / Schedule / Task
graph.add_conditional_edges(
"Supervisor",
self.assign_node_types,
{
"General": 'General',
"Trip": 'Trip',
"Schedule": 'Schedule',
"Task": 'Task'
}
)
# from Agents to END
graph.add_edge("General", END)
graph.add_edge("Trip", END)
graph.add_edge("Schedule", END)
graph.add_edge("Task", END)
# --- Compile Graph ---
compiled_graph = graph.compile()
return compiled_graph
def execute(self, input: str):
self.agent_graph_logger = AgentGraphLogger()
start_time = time.time()
try:
# Build
compiled_graph = self.build_graph()
# Execute
input = json.dumps({"output": input, "assignee": "Human"}, ensure_ascii=False)
result = compiled_graph.invoke(
GraphState(messages=[HumanMessage(content=input)])
)
# Logging Graph
self.agent_graph_logger.log_graph(
input_data=input,
output_data=result,
execution_time=(time.time() - start_time)
)
return result
except Exception as e:
execution_time = time.time() - start_time
# Logging error
self.agent_graph_logger.log_graph(
input_data=input,
error=e,
execution_time=execution_time
)
raise e
Supervisorが次に遷移すべきエージェントを返したら、Graphに書いた条件文で分岐させます。
今回は担当エージェントに分岐したらSupervisorに戻りもせず回答を生成して終了する簡易的なルーティングにしてます。
余談
Supervisorでどうやって次のエージェントを出せるように制御できるんやと思う方がいられるでしょう。
LangChainにはこれを解決する方法があります。
LangChainにはwith_structured_outputというコンポーネントがあり、指定した型の回答を返してくれます。
仕組みとしてはllmモデルに制限をかけれるようにし、エージェントコンポーネントで制限を指定させることでできます。
AgentType = Literal['Supervisor', 'Trip', 'Schedule', 'Task', 'General']
class NextAgentDecision(BaseModel):
"""
NextAgent
---------
- Next Agent with structured output
- this is for Supervisor Agent to decide the next agent
"""
next_agent: AgentType
class SupervisorAgent(BaseAgent):
"""
SupervisorAgent
---------------
- Agent for Supervising and Delegating Tasks
- Based on BaseAgent (for LangChain)
- Uses structured output openai model to decide next agent
"""
def __init__(self) -> None:
super().__init__('Supervisor',
structured_openai(NextAgentDecision),
SUPERVISOR_PROMPT)
今回の目的の本質と外れるので説明はここまでにします。詳しく知りたい方はぜひ以下のリンクをみられてください。
え？DSPyではできないんじゃないかって？？
DSPyではこれと違う、LangChainよりも一貫したアプローチで実現できるのでご安心ください！
少々書き疲れましたが、この後を読めばわかるように頑張って記事を書きます...
AIエージェントシステムのアーキテクチャ
具体的な実装の指針は以上の通りです。このような書き方を徹底すれば、コードの可読性は大きく向上するはずです。
次に、4層アーキテクチャの責務を物理的に分離する、ディレクトリアーキテクチャを示します。
ai
├── __init__.py
├── agent
│ ├── __init__.py
│ ├── _shared
│ ├── evaluation
│ ├── general
│ ├── schedule
│ ├── supervisor
│ ├── task
│ └── trip
│
├── graph
│ ├── __init__.py
│ ├── self_improve_graph.py
│ └── tourist_agent_graph.py
│
├── node
│ ├── __init__.py
│ ├── _shared
│ ├── general
│ ├── optimization
│ ├── schedule
│ ├── states.py
│ ├── supervisor
│ ├── task
│ └── trip
│
└── service
├── __init__.py
├── database
├── dataset
├── tools
└── utils
4層レイヤーに合わせてディレクトリも4つの層に分けます。
ここで依存関係のルールをきめます。
- サービス層はエージェント・ノード・グラフに依存してはいけない
- エージェント層はノード・グラフに依存してはいけない
- ノード層はグラフに依存してはいけない
- グラフ層はサービス・エージェントに直接依存してはいけない
importlinterなどで制限をかけても良いと思います。
このように厳しめの設定をすることで、各レイヤーが単一責任の原則に集中でき、システムのメンテナンス性とスケーラビリティが向上します。
なお、これはレイヤードアーキテクチャに似てると思いますが、グラフ層にはAIエージェントシステム特有の役割を担うための制限の追加があり、必ずしもレイヤードアーキテクチャと同じではないと個人的に考えています。
これでai/ディレクトリ内に全部書くことができました。
最後に、ai/init.pyには、作成したグラフのみを入れて、システムをパッケージ化しましょう。
外部サービスがこのシステムを使う場合、ai/の中身は知らなくて良いはずです。
これでai/ディレクトリ内にAIエージェントシステムをパッケージ化できました。
AIエージェントシステムをアプリケーションに組み込む
それでは、構築したAIエージェントシステムを外部に提供するためのアプリケーションを構築します。
アプリケーションがすることはただ一つ、AIエージェントシステムを呼び出すだけです。
それでは、aiと同じ階層にappディレクトリを作りましょう。
src
├── ai
│
├── app
│ ├── __init__.py
│ ├── main.py
│ ├── schemas
│ │ ├── __init__.py
│ │ └── chat.py
│ └── service
│ ├── __init__.py
│ ├── ai_chat_service.py
│ ├── optimization_service.py
│ └── save_chat_data.py
│
├── config
│
├── infra
│
├── log
│
└── optimizer
では、service層でAIエージェントシステムを呼び出すコードを書きましょう。
from ai import TouristAgentGraph
class AIChatService:
"""
AIChatService
-------------
AI chat service
Use multi-agent-graph from ai/
"""
def __init__(self):
self.ai_agent = TouristAgentGraph()
def chat(self, message: str) -> tuple[str, list]:
"""
Chat with AI agents and return last message and full messages.
Returns:
tuple: (response_content, full_result)
"""
result = self.ai_agent.execute(message)
if result and "messages" in result and result["messages"]:
last_message = result["messages"][-1]
return last_message.content, result["messages"]
else:
raise Exception("No response from agents.")
あとはこれをFastAPIの根幹で呼び出せば完成です！
main.pyを編集しましょう。
# Application
app = FastAPI(title="tourist-ai", version="1.0.0")
# Services
ai_chat_service = AIChatService()
# celery_service = CeleryService()
# API Endpoints
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
try:
# Get AI chat response
response_content, full_result = ai_chat_service.chat(request.message)
# Save Messages in Background
background_tasks.add_task(
save_chat_data,
full_result
)
return ChatResponse(response=response_content)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
DSPy製エージェントの組み込み
それでは、早速DSPy製のエージェントを作成しましょう
DSPyとは、プロンプトエンジニアではなく、宣言的なプログラミングによりLLMを生業することを目的とします。
DSPyによるエージェントは以下の２要素で作られます。
- シグネチャ
特定のタスクの入出力の構造を定義するものです。
必要な入力フィールドと出力フィールドを明確に定義することで、LLMは何をするかではなく、何が必要かをLLMにパラメータとして保持させることができます。 - モジュール
モジュールは、シグネチャとLLMを使って、推論ロジックを実行します。
シグネチャで定義された入出力の構造を使って、LLMを呼び出します。
これにより、シグネチャでさえ型を定義していれば、エージェントはその型で回答を出力します。
（先ほどLangChainではモデル自体に制限をかけ、エージェントはそれを使う的なことをしていましたが、DSPyではモジュールに対しシグネチャを割り当てることで、このモジュール自体に制限をかけることができます。）
DSPyについては次回で話すほうが都合が良いので今回はこれぐらいにしますが、詳しく知りたい方は以下のリンクを見られてください。
それでは早速実装してみましょう。スケジュールエージェントとノードはこのようになるはずです
class ScheduleSignature(dspy.Signature):
"""
Schedule Signature
------------------
- Defines Signature for Schedule AI Agent using DSPy
"""
input = dspy.InputField(
type=str
)
response = dspy.OutputField(
desc=SCHEDULE_PROMPT,
type=str,
)
class DeclarativeScheduleAgent(dspy.Module):
"""
Declarative Schedule Agent
----------------------
- Agent for Suggesting Schedules which are necessary for achieving the user's goal.
- Using DSPy framework
- Using Chain of Thought
- Returns structured JSON format with output and assignee fields
"""
def __init__(self):
super().__init__()
self.lm = dspy_openai()
self.agent = dspy.ChainOfThought(ScheduleSignature)
def forward(self, input: str) -> Dict[str, AgentType]:
with dspy.context(lm=self.lm):
result = self.agent(input=input)
return {
"output": result.response,
"assignee": 'Schedule'
}
class DeclarativeScheduleNode(BaseNode):
"""
DeclarativeScheduleNode
-----------------------
- Node for Schedule Purpose Tasks using Declarative Agent
- Based on ScheduleNode
- Call Declarative Schedule Agent
"""
def __init__(self) -> None:
self.node_type: NodeType = 'Schedule'
self.Agent = DeclarativeScheduleAgent()
def process(self, state: GraphState) -> Command:
try:
user_input = self.extract_messages(state)
agent_output = self.Agent(input=user_input)
return self.update_state(agent_output)
except Exception as e:
error_message = f"Error in {self.node_type} Node processing: {str(e)}"
return self.update_state(error_message)
あとはこれをグラフで呼び出すだけです。ここを変えるだけで完成です！
graph.add_node('Schedule', DeclarativeScheduleNode().process)
LangChain製とDSPy製の比較
これでDSPy製のエージェントをLangChainに組み込むことができました！
実は今実装したDSPy製のエージェントは、LangChainより少しリッチな機能を組み込んでおきました。
それがCoT(Chain of Thought)です。
Chain of Thoughtとは、LLMが単に最終的な答えを出す前に、その推論過程を段階的に言語化させる手法です。
人間が複雑な問題を解くとき、一度に解かずに複数の段階に分けて考えるように、LLMにもそれぞれの段階ごとに自然言語で中間的な推論をさせます。
これはLLMが「なぜこの答えに至ったのか？」という中間的な思考プロセスを踏むことを擬似的に再現できていそうですね。
この「思考の連鎖（Chain）」を生成させることで、モデルはより論理的かつ正確に問題を解決できるようになるそうです。
それでは早速動かしてみましょう。
今回は以下のシステムプロンプトと質問文を使います。
Act as a schedule AI agent.
You can assist with suggesting schedules for achieving what the user wants.
Please Search the web using the TavilyResearch tool if necessary.
Sometimes user may ask you to suggest schedules ambiguously.
This means user does not know exactly what they want.
You should search them yourself.
You must always suggest schedules in a numbered list format.
京都の金閣寺と三十三間堂に観光に行きたい。
かなり抽象的な（適当な）質問ですが、どんな回答が返ってくるでしょうか？
LangChain製のエージェントは以下の回答でした。
Sure! Please let me know what you would like to achieve or what kind of schedule you need assistance with, and I\'ll be happy to help!
システムプロンプトの英語文に釣られていますね。また指示も抽象的だったので、ユーザの要求に対し、目的に固執してしまっているように思えます。
正直ここまで政府度が悪いのは予想外でした。
それではDSPy製ののChain-of-Thoughtエージェントはどうでしょうか？
以下は金閣寺と三十三間堂を観光するためのスケジュールです。
1. **08:30** - 京都駅を出発
京都駅から金閣寺へ向かいます。
2. **09:00** - 金閣寺に到着
金閣寺を観光し、美しい庭園を散策します。
3. **10:30** - 金閣寺を出発
次の目的地である三十三間堂へ向かいます。
4. **11:00** - 三十三間堂に到着
三十三間堂を観光し、千手観音像を鑑賞します。
5. **12:30** - 昼食
近くのレストランで地元の料理を楽しみます。
6. **13:30** - 観光を続ける
近隣の他の観光スポット（例えば、清水寺や東山地区）に訪れることも検討できます。
7. **17:00** - 京都駅に戻る
1日の観光を終え、京都駅へ戻ります。
このスケジュールに従えば、効率的かつ楽しい観光ができるかと思います。
驚きですよね。私も驚きました。抽象的な指示なのにしっかりと役割を認識し、かつnumbered list formatという指示も遵守してます。
もちろん、これはDSPyが推論過程でLLMへの問い合わせ回数を増やしているからなので、決してLangChainが劣っているというわけではないのでご了承ください。
また、何度も推論させている分、回答が若干遅いです。
DSPyには他にも推論の手法を提供しています。興味のある方はぜひ調べてみてください。
今回は、CoTにより、推論を最適化させることができました。しかし、今回の目的のプロンプトの最適化はまだ行えていません。
長くなりましたので、いよいよ本命のプロンプトの最適化については、次回にしようと思います。
まとめ
本記事では、LangChainとDSPyを組み合わせた自律改善型AIエージェントシステムを紹介しました。
実現したこと
- 開発効率: LangChainのエコシステムによる豊富なツール群
- メンテナンス性: DSPyによる自動プロンプト最適化
- 拡張性: 4層アーキテクチャによる関心の分離
- 継続的改善: ユーザーが使うほど性能が向上するシステム
技術的なポイント
- Agent層にDSPyを配置: LangChainの構造を崩さずDSPyを導入
- 宣言的エージェント設計: プロンプト詳細はDSPyに委譲
- データドリブンな最適化: 対話ログから自動的にプロンプト改善
- 統一インターフェース: 既存コードを変更せずエージェント切り替え可能
このアプローチにより、技術者の手動チューニングに依存しない、自律的に成長するAIエージェントシステムの
基盤
を構築できました。
次回は実際の運用データを蓄積し、DSPyの最適化機能を実装していきます。
次回ではこのようなシステムに発展させていきます（現在はこの構成で作っておりますが、最適化の機能改善しだいでは変わるかもしれません）
┌────────────────────────────────────────────────────────┐
│ Tourist AI System │
├────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌─────────────────┐ │
│ │ tourist-ai │──────│ tourist-ai-db │ │
│ │ (FastAPI) │ │ (PostgreSQL) │ │
│ └──────┬───────┘ └─────────────────┘ │
│ │ │
│ │ ┌─────────────────┐ │
│ └──────────────│ tourist-ai-redis│ │
│ ┌───────│ (Redis) │ │
│ ┌──────────────┐ └─────────────────┘ │
│ │ tourist-ai- │ │
│ │ optimizer │ │
│ │ (Celery) │ │
│ └──────────────┘ │
│ │
└────────────────────────────────────────────────────────┘
最後に、ここまで読んでくださり本当にありがとうございました。
もし私の知識が誤っていることがあれば、申し訳ありません。
ただ、私の持つアイデアだったり少ない知識が、少しでも読んでくれた皆様の、何か良いことのきっかけにでもなればと願っています。
また、しばらく私は卒論で忙しくなりますが、次回に記事を書いたら、また読んでいただけると嬉しいです。
Discussion