Kasi Yaswanth

私は最近、LangGraphとMCPで構築したサポートボットのデバッグに何時間も費やしましたが、問題はエージェントの推論自体ではなく、その推論をユーザーにどのように提示するかという点にあることに気づきました。ボットはしばしばローディングスピナーで止まっているように見え、ユーザーはクエリの処理がフリーズしているのか、それともまだアクティブに処理中なのかを疑問に思っていました。問題は、フロントエンドがエージェントの思考プロセス全体が完了した時点でしか更新されず、複雑なクエリでは数秒から数分かかる可能性があることでした。

この経験から、エージェントの推論をリアルタイムでフロントエンドにストリーミングし、よりインタラクティブで魅力的なユーザー体験を提供する方法を探求するに至りました。そうすることで、ユーザーはエージェントが何を行い、なぜそうするのかをよりよく理解できるようになり、特に透明性と信頼が重要なアプリケーションでは重要です。

これを実現するには、LangGraphのStateGraphとadd_node機能を活用して、エージェントの推論プロセスを表すグラフを構築できます。次に、MCPのtoolsとresourcesを使用して、エージェントが思考プロセスを進める際にフロントエンドに送信できる更新ストリームを作成できます。

以下は、Pythonを使用したこの実装方法の例です：

import langgraph as lg
from mcp import tools, resources

# Create a new StateGraph to represent the agent's reasoning process
graph = lg.StateGraph()

# Define a node that represents the agent's current thought process
def thought_process_node(graph, thought_process):
    node = graph.add_node(thought_process)
    # Add conditional edges to represent the possible next steps in the thought process
 graph.add_conditional_edges(node, [
        (thought_process + " -> considering option A", 0.7),
        (thought_process + " -> considering option B", 0.3)
    ])
    return node

# Create a stream of updates that can be sent to the frontend
def stream_reasoning(graph, initial_thought_process):
    current_node = thought_process_node(graph, initial_thought_process)
    while True:
        # Get the next node in the graph based on the current node and the agent's reasoning
        next_node = graph.get_next_node(current_node)
        # Send an update to the frontend with the current thought process and the next possible steps
        yield {
            "thought_process": next_node.value,
            "next_steps": [edge.value for edge in graph.get_outgoing_edges(next_node)]
        }
        # Update the current node
        current_node = next_node

# Create a checkpoint to save the agent's reasoning process
checkpoint = resources.Checkpoint("agent_reasoning")

# Start the agent's reasoning process and stream the updates to the frontend
initial_thought_process = "considering user query"
for update in stream_reasoning(graph, initial_thought_process):
    # Send the update to the frontend
    print(update)
    # Save the current state of the agent's reasoning process
    checkpoint.save(graph)

Enter fullscreen mode Exit fullscreen mode

このコードは、エージェントの推論プロセスを表すStateGraphを作成し、現在の思考プロセスを表すノードを定義します。次に、エージェントが思考プロセスを進める際にフロントエンドに送信できる更新ストリームを作成します。stream_reasoning関数は、各ステップで更新をyieldし、現在の思考プロセスと次の可能なステップを含みます。

このアプローチを実装する際に注意すべき実用的な落とし穴は、エージェントの推論プロセスが高度に分岐する可能性があり、指数関数的な数の可能な次のステップにつながることです。これを軽減するには、ビームサーチやプルーニングなどの手法を使用して、可能な次のステップの数を制限し、グラフが大きくなりすぎるのを防ぐことができます。

エージェント型AIの可能性を探求し続ける中で、エージェントの推論の複雑さと、LangGraphとMCPを使用してより透明性が高くインタラクティブなシステムを構築する方法について、さらに深く掘り下げていきます。明日、複数のエージェントとモデルを統合して、さらに強力で柔軟なAIアプリケーションを作成する方法を見ていきます。