導入

以下の記事で、LangGraph+ローカルLLMを使ってAgentの実装をしました。

同様の仕組で、Human in the Loopをしてみます。

Human in the Loop?

ここではAgentの処理ループの中に、人間の承認・判断を組み込むことを言います。

例えば、ツールを実行する前に人が処理内容をチェックし、環境に影響を及ぼす処理の場合は止めるような判断をいれることができます。

今回は以下の公式Exampleのウォークスルーを改変して実行します。
また、プロンプトテンプレートもカスタマイズしたものを利用してみます。

実践・検証はDatabricks on AWS上で実施しました。

Step1. パッケージのインストール

使うパッケージをインストール。
今回もExLlama V2を使って推論します。
また、langgraphを追加でインストールしています。

%pip install -U transformers accelerate "exllamav2>=0.0.11" langchain langchainhub duckduckgo-search

dbutils.library.restartPython()

Step2. モデルのロード

LLMのモデルを読み込みます。
今回もこちらで作成した、langchainのカスタムチャットモデルを使って読み込みました。

使っているモデルはOpenChat 3.5の0106版をGPTQで量子化した以下のモデルです。
事前にダウンロードしておいたものを使います。

from exllamav2_chat import ChatExllamaV2Model

model_path = "/Volumes/training/llm/model_snapshots/models--TheBloke--openchat-3.5-0106-GPTQ"

chat_model = ChatExllamaV2Model.from_model_dir(
    model_path,
    cache_max_seq_len=8192,
    system_message_template="GPT4 Correct User: {}<|end_of_turn|>GPT4 Correct Assistant: OK.<|end_of_turn|>",    
    human_message_template="GPT4 Correct User: {}<|end_of_turn|>GPT4 Correct Assistant: ",
    ai_message_template="{}",
    temperature=0.0001,
    top_p=0.0001,
    max_new_tokens=1024,
    repetition_penalty = 1.15,
    low_memory=True,
    cache_8bit=True,
)

# TheBloke/openchat-3.5-0106-GPTQはtokenizerのEOS設定に不具合があるので修正
chat_model.exllama_tokenizer.eos_token_id = 32000

Step3. ツールの準備

LLM Agentの中で呼び出して実行するツールを定義します。

今回はDuckDuckGoとPython REPLを実行するツールの2種を用意しました。


from langchain_core.tools import tool
from typing import Annotated
from langchain_experimental.utilities import PythonREPL
from langchain_community.tools.ddg_search.tool import DuckDuckGoSearchResults, DuckDuckGoSearchRun

search = DuckDuckGoSearchRun(max_results=5)

repl = PythonREPL()

@tool
def python_repl(
    code: Annotated[str, "The python code to execute to generate your chart."]
):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    return f"Succesfully executed:\n```python\n{code}\n```\nStdout: {result}"

tools = [search, python_repl]

Step4. Agent用Chainの作成

エージェント用のプロンプトテンプレートを取得し、create_json_chat_agent関数を使ってAgentのChainを作成します。

少しツールを使われやすくするために、hwchase17/react-chat-jsonのプロンプトテンプレートを微修正したプロンプトテンプレートを用意し、create_json_chat_agentでエージェントを作成します。

from typing import List, Union
from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    ChatMessage,
    SystemMessage,
    FunctionMessage,
    ToolMessage,
)
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, PromptTemplate, MessagesPlaceholder, HumanMessagePromptTemplate
from langchain.agents import create_json_chat_agent

prompt2 = ChatPromptTemplate(
    input_variables=["agent_scratchpad", "input", "tool_names", "tools"],
    input_types={
        "chat_history": List[
            Union[
                AIMessage,
                HumanMessage,
                ChatMessage,
                SystemMessage,
                FunctionMessage,
                ToolMessage,
            ]
        ],
        "agent_scratchpad": List[
            Union[
                AIMessage,
                HumanMessage,
                ChatMessage,
                SystemMessage,
                FunctionMessage,
                ToolMessage,
            ]
        ],
    },
    messages=[
        SystemMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=[],
                template="Assistant is a large language model trained by OpenAI.\n\nAssistant is designed to be able to assist with a wide range of tasks, from answering simple questions to providing in-depth explanations and discussions on a wide range of topics. As a language model, Assistant is able to generate human-like text based on the input it receives, allowing it to engage in natural-sounding conversations and provide responses that are coherent and relevant to the topic at hand.\n\nAssistant is constantly learning and improving, and its capabilities are constantly evolving. It is able to process and understand large amounts of text, and can use this knowledge to provide accurate and informative responses to a wide range of questions. Additionally, Assistant is able to generate its own text based on the input it receives, allowing it to engage in discussions and provide explanations and descriptions on a wide range of topics.\n\nOverall, Assistant is a powerful system that can help with a wide range of tasks and provide valuable insights and information on a wide range of topics. Whether you need help with a specific question or just want to have a conversation about a particular topic, Assistant is here to assist.",
            )
        ),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        HumanMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=["input", "tool_names", "tools"],
                template='TOOLS\n------\nAssistant can ask the user to use tools to look up information that may be helpful in answering the users original question. The tools the human can use are:\n\n{tools}\n\nRESPONSE FORMAT INSTRUCTIONS\n----------------------------\n\nWhen responding to me, please output a response in one of two formats:\n\n**Option 1:**\nUse this if you want the human to use a tool.\nMarkdown code snippet formatted in the following schema:\n\n```json\n{{\n    "action": string, \\ The action to take. Must be one of {tool_names}\n    "action_input": string \\ The input to the action\n}}\n```\n\n**Option #2:**\nUse this if you can respond directly to the human after tool execution. Markdown code snippet formatted in the following schema:\n\n```json\n{{\n    "action": "Final Answer",\n    "action_input": string \\ You should put what you want to return to use here\n}}\n```\n\nUSER\'S INPUT\n--------------------\nHere is the user\'s input (remember to respond with a markdown code snippet of a json blob with a single action, and NOTHING else):\n\n{input}',

            )
        ),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ],
)

agent_runnable = create_json_chat_agent(chat_model, tools, prompt2)

Step5. LangGraphで処理フローを定義

まずは前回同様、処理中の状態を保管するためのクラスを定義します。

from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
   input: str
   chat_history: list[BaseMessage]
   agent_outcome: Union[AgentAction, AgentFinish, None]
   intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

次にノード用関数を定義。
execute_tools関数の中で、ユーザにツールを実行してよいかどうかの判断を求める処理を入れます。
これによって、ツールを実行する前に実行確認を得ることができます。

その他は以前の記事と同様です。

from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor
import re

tool_executor = ToolExecutor(tools)

def run_agent(data):
    agent_outcome = agent_runnable.invoke(data)
    return {"agent_outcome": agent_outcome}

def execute_tools(data):

    agent_action = data['agent_outcome']

    tool = agent_action.tool
    tool_input = agent_action.tool_input

    # Human in the Loop: ユーザへツールの実行確認
    response = input(prompt=f"[y/n] continue with: {tool} with {tool_input}?")
    if response == "n":
        raise ValueError

    output = tool_executor.invoke(agent_action)
    return {"intermediate_steps": [(agent_action, str(output))]}

# 条件分岐でどちらのエッジを進むかを決定するために使用されるロジックを定義する
def should_continue(data):
    if isinstance(data['agent_outcome'], AgentFinish):
        return "end"
    else:
        return "continue"

最後にグラフを定義します。

from langgraph.graph import END, StateGraph

# 新しいグラフを定義
workflow = StateGraph(AgentState)

# 循環する二つのノードを定義する
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)

# エントリポイントを設定
workflow.set_entry_point("agent")

# 次に条件付きエッジを追加する
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END,
    },
)

workflow.add_edge("action", "agent")

app = workflow.compile()

ここまでで実行する準備ができました。

Step6. 実行

では、実行してみましょう。
"what is weather in sf"を聞いてみます。

inputs = {"input": "what is weather in sf", "chat_history": []}
result = app.invoke(inputs)

print(result["agent_outcome"].return_values["output"])

実行すると、duckduckgo_serchを使ってよいか確認する入力欄が表示されます。

yを入れると処理が継続され、ツールの実行結果としての回答を得ることができます。

nを入力すると例外が発生して終了します。
入力時の処理はexecute_tools関数の中でハンドリングできるため、単純に例外発生で止めるのではなく、別のツールの利用を促すなど、ユースケースによって使い分けることができます。

まとめ

LangGraphを使ったHuman in the Loopを組み込んだAgentの実装を試してみました。

今回、ツールでPython REPLを入れたのですが、なかなかツール選択の対象にならず。。。
このあたり、NexsusRavenなど、FunctionCallingに適したモデルを使う方がよさそうです。

※　以前試してみた記事はこちら。