How to Integrate Process Rewards in LangGraph Agents and Handle Node Execution Failures?

オープン
  • 何か問題がありますか？
#387 コメント 4 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 3 日以内に返信

@beanie00 がすでに取り組んでいます。

2025年12月10日 から。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
25/100
issue の種類
ドキュメント
明瞭さ
説明が足りない
活発さ
停滞
技術スタック
python
領域
ai, machine-learning

調査の方向性

agl.emit_reward の呼び出しと LitAgent.rollout のエントリーポイントから始め、次に LangGraph StateGraph の呼び出しと、issue に示されている条件付きエッジの例を調査してください。この統合において、中間報酬とノードの失敗がどのように扱われる想定なのかを判断してください。完了条件は、文書化された明確な回答、または実装に関する範囲を絞った追加依頼です。

索引モデルが issue の本文から書いたものです。

説明

examples question v0.x

Hello, I’m exploring how to combine LangGraph with process rewards and handle node failures in sequential agents. I have two main questions that aren’t covered in existing tutorials:

Process reward + LangGraph

Is it possible to call emit_reward directly within an agent’s node to emit intermediate rewards during training? For example:

import agentlightning as agl
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.graph.state import CompiledStateGraph

class Agent:
        def ___init___(...):
                 ...

        def node_1(self, state: State) -> State:
                result = self.invoke_prompt(prompt)
                reward_1 = ...
                agl.emit_reward(reward_1)
                new_state = ...
                return new_state

        def node_2(self, state: State) -> State:
                result = self.invoke_prompt(prompt)
                reward_2 = ...
                agl.emit_reward(reward_2)
                new_state = ...
                return new_state

        def graph(self) -> CompiledStateGraph[State]:
            builder = StateGraph(State)
            builder.add_node(self.node1)
            builder.add_node(self.node2)

            builder.add_edge(START, "node1")
            builder.add_edge("node1", "node2")
            builder.add_edge("node2", END)

            return builder.compile()


class LitSQLAgent(agl.LitAgent[Dict[str, Any]]):

    def __init__(...):
        ...

    def rollout(
        self,
        task: Dict[str, Any],
        resources: agl.NamedResources,
        rollout: agl.Rollout,
    ) -> None:
        agent = Agent(...).graph()
        try:
                # Required to make the langchain tracing work
                handler = self.tracer.get_langchain_handler()
                result = agent.invoke( 
                    {"question": question},
                    {"callbacks": [handler] if handler else [], "recursion_limit": 100},
                )
        except Exception as e:
                logger.exception(f"[Rollout {rollout_id}] Error during agent invocation: {e}")
                return
        return

Handle node failures

If I should add a confitional edge for each intermediate node?

class Agent:
    ...  # nearly the same as the agent above

        def check_1(self, state: State):
            if is_training and not is_success(state):
                    return END
            else:
                    return "node2"

        def graph(self) -> CompiledStateGraph[State]:
            builder = StateGraph(State)
            builder.add_node(self.node1)
            builder.add_node(self.node2)

            builder.add_edge(START, "node1")
            builder.add_edge("node2", END)
            builder.add_conditional_edges("node1", self.check_1)

            return builder.compile()
主要言語
Python
スター
18.9k
フォーク
1.7k
平均マージ
4日 11時間
マージ済み PR（30日）
9

環境構築

このプロジェクトには開発コンテナ、Dockerfile、コントリビューションガイドがありません。まず README を読み、一般的な手順ははじめてのコントリビューションガイドを参照してください。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

microsoft/agent-lightning のほかの issue

microsoft/agent-lightning の issue をすべて見る

似ている issue

Python の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。