Decompose ReAct

1. Overview

Custom ReActをベースにReActの内部で起こっている部分をすべて一つずつ分解して見る。

含まれる内容

LLMに送られるPrompt (Prompt Template): Agentにセットされている
LLMから帰ってきたOutput: Agent内
Output Parser: Agent内
AgentActionとAgentFinish: Agent.planの結果がAgentActionかAgentFinishでAgentFinishの場合にはAgentExecutor内のWhileを抜けて終了する

詳細:

AgentExecutorはagentとtoolsで初期化される
agent_executor(inputs=question, return_only_outputs=False, callbacks=None) で実行されるが、 AgentExecutorはChainを継承しているので Chain.__call__が呼ばれる事となる。
Chain.__call__の内部からは、AgentExecutor._callが呼ばれる
AgentExecutor._call内ではwhileによって終了条件or _take_next_stepの結果がAgentFinishになるまで_take_next_stepを行う
_take_next_step内では、agent_action = agent.plan(intermediate_steps, callbacks, inputs)を実行
1. agent_action がAgentActionの場合は、対応するツールを呼んで結果を得る observation = tool.run()をとり、 result.append((agent_action, observation))としてresultを返す (基本len(result)は1)
2. agent_actionがAgentFinishの場合は agent_actionをそのまま帰す
_take_next_stepの結果を取り
1. AgentFinishの場合はそのままreturnして終了 (return_valuesにintermediate_stepsを入れて返す)
2. AgentActionの場合は、intermediate_stepsにintermediate_steps.extend(result) 前回のresultを追加して、whileループを続ける

2. Implementation

_take_next_step Stepの中身

output = self.agent.plan(
    intermediate_steps,
    callbacks=run_manager.get_child() if run_manager else None,
    **inputs,
)

intermediate_steps: agent.plan内で実行したAgentActionとResultの履歴(tuple)のlist (intermediate_steps)
run_manager <- callbackManager.on_chain_startの返り値 (特に不要であればNoneでいい)
inputsは Chain.callのなかのinputs = self.prep_inputs(inputs)で変換されている

2.1. run_managerの出処CallbackManagerに関して

CallbackManager.configure:

AgentExecutor(Chain).callbacks はfrom_agent_and_toolsで作成するとNone (ref)

callback_manager = CallbackManager.configure(
    inheritable_callbacks=None, local_callbacks=None, verbose=True, inheritable_tags=None, local_tags=None
)
run_manager = callback_manager.on_chain_start( # _callにrun_managerがあるのでrun_managerを設定する
    dumpd(self), # dumpd(obj: Any) = json.loads(dumps(obj))
    inputs, # self.prep_inputsでDictになったもの
)

2.2. intermediate_stepsに関して

最初の状態は [] (_call()内)

intermediate_steps: List[Tuple[AgentAction, str]] = []

毎回agent.planでどのツールを使うか決め、Toolを実行した後に result.append((agent_action, observation))がextendされる

2.3. inputs & input_keys

Chain.__call__内でinputsは変形されているので注意
```
inputs = {list(_input_keys)[0]: inputs}
```
1. _input_keysは各 self.agent.input_keys (AgentExecutor.input_keys) 1. 今回のAgentの input_keysは
```
agent.input_keys
['input']
```

2.4. callbacks

agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    callback_manager=None, # default: None (explicitly write None)
    verbose=True,
)

以下の定義よりcls(agent=agent, tools=tools, callback_manager=callback_manager, **kwargs) AgentExecutorを初期化する。

class AgentExecutor(Chain):
    ...
    @classmethod
    def from_agent_and_tools(
        cls,
        agent: Union[BaseSingleActionAgent, BaseMultiActionAgent],
        tools: Sequence[BaseTool],
        callback_manager: Optional[BaseCallbackManager] = None,
        **kwargs: Any,
    ) -> AgentExecutor:
        """Create from agent and tools."""
        return cls(
            agent=agent, tools=tools, callback_manager=callback_manager, **kwargs
        )

callback_managerはNoneで、verboseがTrue.
CallbackManagerを初期化してからrun_managerを取り出すがcallback manager自体は特に使われていない。
Chain.__call__内で run_managerがcallback_manager.on_chain_start()で取り出され、self._call(inputs, run_manager=run_manager)で渡される

3. Tips

3.1. 準備

poetry run python

from src.langchain.react_decompose import *

3.2. 全部を実行

llm = OpenAI()
agent = ReActTestAgent.from_llm_and_tools(
    llm,
    tools,
)
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    callback_manager=None, # default: None (explicitly write None)
    verbose=True,
)
question = "How much is the difference between the total of company C, F and the total of company A, E ?"
agent_executor(inputs=question, return_only_outputs=False, callbacks=None)

3.3. Agent.planのStepを1つ実行 (_take_next_step前半)

実際の実行では、AgentExecutor._take_next_stepの中のループで行われている
OutputParserがParseに失敗することもある
返り値は、 AgentAction or AgentFinish

intermediate_steps = []
agent_action = agent.plan(intermediate_steps, callbacks=None, **{"input":"How much is the difference between the total of company C, F and the total of company A, E ?"})
AgentAction(tool='GetInvoice', tool_input='C', log='思考: I need to get invoice amount of company C.\n行動: GetInvoice[C]')

AgentActionにはtool, tool_input, logがある

3.4. Agent.plan結果からToolを呼ぶ (_take_next_step後半)

上のagent.planの結果がAgentActionだった場合、_take_next_stepの結果を以下のように取れる

name_to_tool_map = {tool.name: tool for tool in tools} # https://github.com/hwchase17/langchain/blob/b95002289409077965d99636b15a45300d9c0b9d/langchain/agents/agent.py#L939C9-L939C25
agent_action # AgentAction(tool='GetInvoice', tool_input='C', log='思考: I need to get invoice amount of company C.\n行動: GetInvoice[C]')
result = []
tool = name_to_tool_map[agent_action.tool]
return_direct = tool.return_direct # false
tool_run_kwargs = agent.tool_run_logging_kwargs() # {'llm_prefix': '思考: ', 'observation_prefix': '観察: '}
if return_direct:
    tool_run_kwargs["llm_prefix"] = ""
observation = tool.run(
    agent_action.tool_input,
    verbose=True,
    callbacks=None,
    **tool_run_kwargs,
) # observation = 20000 <- toolから返された結果
result.append((agent_action, observation))
result # _take_next_stepで返される結果 [(AgentAction(tool='GetInvoice', tool_input='C', log='思考: I need to get invoice amount of company C.\n行動: GetInvoice[C]'), 20000)]

3.5. AgentExecutor._callのwhile内で_take_next_stepの結果を受け取った後

next_step_output = self._take_next_step(...)

next_step_outputにより - AgentFinish -> _returnを呼んで終了 - その他→続行

intermediate_steps.extend(result) # resultは前のステップでのresult

前回とほぼ同様に intermediate_stepsだけが変わった状態でもう一度agent.planを呼ぶ

agent_action = agent.plan(intermediate_steps, callbacks=None, **{"input":"How much is the difference between the total of company C, F and the total of company A, E ?"})
agent_action # AgentAction(tool='GetInvoice', tool_input='F', log=' I need to get invoice amount of company F.\n行動: GetInvoice[F]')

agent_actionの処理を3.4.と同様にするとresultはいかのようになる

result
[(AgentAction(tool='GetInvoice', tool_input='F', log=' I need to get invoice amount of company F.\n行動: GetInvoice[F]'), 20000), (AgentAction(tool='GetInvoice', tool_input='F', log=' I need to get invoice amount of company F.\n行動: GetInvoice[F]'), 4100)]

そしてintermediate_stepsを更新する

intermediate_steps.extend(result) # resultは前のステップでのresult

さらにagent.plan(intermediate_steps, callbacks, inputs) -> agent_action or agent_finish -> tool.run -> observation -> resultにappend -> intermediate_steps更新 -> …を繰り返す

結局一番大事なのはintermediate_stepsに結果を貯めていくこと

3.6. AgentFinishとなった場合

_returnでreturn_valuesを返す

AgentExecutor._return

def _return(
    self,
    output: AgentFinish,
    intermediate_steps: list,
    run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
    if run_manager:
        run_manager.on_agent_finish(output, color="green", verbose=self.verbose)
    final_output = output.return_values
    if self.return_intermediate_steps:
        final_output["intermediate_steps"] = intermediate_steps
    return final_output

4. Run

poetry run python src/langchain/react_decompose.py