监听Agent(Python SDK)

一次Agent的响应如果涉及到多个任务的分解，往往会执行比较长的时间，此时可以对agent的执行过程进行监听，输出中间步骤。

AgentListener的定义如下：

class AgentListener(ABC):
    """Agent监听，允许对Agent的各个阶段进行处理
    """
    def on_session_start(self, agent_session: AgentSession):
        """
        Session启动时调用
        :param agent_session: AgentSession
        """

    def on_session_iteration(self, agent_session: AgentSession):
        """
        Session迭代过程中调用
        :param agent_session: AgentSession
        """

    def on_session_end(self, agent_session: AgentSession):
        """
        Session结束时调用
        :param agent_session: AgentSession
        """

    def on_check_interrupt_requirements(self, agent_session: AgentSession):
        """
        onSessionIteration调用结束后，检查Agent是否需要终止，如果需要终止，则返回true，默认不终止
        可以在终止前对agentSession进行修改，如：修改agent的finalAnswer
        :param agent_session: AgentSession
        :return: bool类型结果
        """
        return False

from pangukitsappdev.agent.agent_session import AgentSession
from pangukitsappdev.api.agent.base import AgentListener

class TestListener(AgentListener):
    def on_session_start(self, agent_session: AgentSession):
        print(agent_session)

    def on_session_iteration(self, agent_session: AgentSession):
        print(agent_session)

    def on_session_end(self, agent_session: AgentSession):
        print(agent_session)

上述代码分别对应了Agent的开始、中间过程和结束阶段。

from pangukitsappdev.agent.react_pangu_agent import ReactPanguAgent
from pangukitsappdev.api.llms.factory import LLMs

agent = ReactPanguAgent(LLMs.of("pangu"))
agent.add_listener(TestListener())

listener会在Agent运行时生效。

class AgentSession(BaseModel):
    """
    Agent运行Session，包含历史Action，当前Action，状态
    Attributes:
        messages: 本次session的用户的输入
        session_id: UUID，在一个session内唯一
        current_action: 当前Action
        agent_session_status: Agent状态
        is_by_step: 是否是逐步执行
        current_message: 当前的AssistantMessage
    """
    messages: List[ConversationMessage]
    session_id: str
    current_action: Optional[AgentAction]
    agent_session_status: str
    is_by_step: Optional[bool]
    current_message: Optional[ConversationMessage]

AgentAction包含Agent的工具选择、工具执行结果、思考等信息，AgentSessionStatus为Agnet的执行状态。

class InterruptListener(AgentListener):
    def on_check_interrupt_requirements(self, agent_session: AgentSession):
        if "capital" == agent_session.current_action.action:
            # 如果当前的action为captial，则返回capital工具调用的原始返回值
            agent_session.final_answer = agent_session.current_action.observation
            return True
        return False
agent.add_listener(InterruptListener())