AI Agent:tools:takeover_memory.py:ソースコード


"""
takeover_memory

短期記憶メモリ
履歴情報を削除する前に子のメモリに一時保存して、次に引きつぐのに使う。
"""



from langchain_core.tools import tool
from pydantic import BaseModel, Field
g_memory = ""
from tools.program_called_command_list import wait_safety
class Save2TakeoverMemoryInput(BaseModel):
    data: str = Field(..., description="保存するデータ")

@tool(args_schema=Save2TakeoverMemoryInput)
def save2takeover_memory(data:str) -> bool:
    """
    次のエージェントに引き継ぐための情報を一時的に記憶するメモリに書き込みます。
    作業を再開するのに必要な情報を書き込んでください。
    ユーザーからの大元のタスク達成のために行う作業リスト。
    ユーザーからの大元のタスク達成のために行う作業リストの、現在の作業状況。
    行った作業。
    行った作業の目的。
    行った作業の作業状況。
    行った作業の結果。
    次に行うべき作業。
    次に行うべき作業の目的。
    次に行うべき作業の作業状況。
    そのほか必要な情報。
    
    
    :param data: 次のエージェントに引き継ぐための情報を一時的に記憶する情報。詳細は上にある。
    :type data: str
    :return: True
    :rtype: bool
    """
    print("save2takeover_memory called")
    global g_memory
    wait_safety()  # リクエストの制限を越えないために必要
    g_memory = data
    return True



def _get_from_takeover_memory():
    """
    前のエージェントが書き込んだ引き継ぎに必要な情報を取得する。

    """
    global g_memory
    return g_memory

def _clear_takeover_memory():
    """
    前のエージェントが書き込んだ引き継ぎに必要な情報をクリアする。

    """
    global g_memory
    g_memory = ""

g_end_of_task = False

class EndOfTaskInput(BaseModel):
    flag: bool = Field(..., description="ユーザーから要求されたタスクが終了したかどうか。Trueで終了したことを示す。")
@tool(args_schema=EndOfTaskInput)
def end_of_task(flag:bool) -> bool:
    """
    ユーザーから要求されたタスクが終了したかどうかを設定します。

    :param flag: ユーザーから要求されたタスクが終了したかどうか。 Trueで終了したことを示す
    :type flag: bool
    :return: True
    :rtype: bool
    """
    print("end_of_task called",flag)
    global g_end_of_task
    wait_safety()  # リクエストの制限を越えないために必要

    g_end_of_task = flag
    return True

def _is_end_of_task():
    global g_end_of_task
    return g_end_of_task  

def get_tools_list():
    return [save2takeover_memory, end_of_task]