pangukitsappdev.api.agent package¶

Submodules¶

pangukitsappdev.api.agent.base module¶

class pangukitsappdev.api.agent.base.AbstractAgent(llm: LLMApi)¶

基类：Agent

add_listener(agent_listener: AgentListener)¶: 添加一个Agent监听器 :param agent_listener: Agent监听器

add_tool(tool: AbstractTool)¶: 为Agent增加工具类 :param tool: Tool

clear_tool()¶: 删除所有工具

static convert_message_to_dict(conversation_messages: List[ConversationMessage]) → List[Dict]¶

get_system_prompt(agent_session: AgentSession)¶

get_tool(tool_id: str)¶

init_tool_from_retriever(messages: List[ConversationMessage])¶

static levenshtein_distance(tool_id: str, tool_map_key: str) → int¶

need_interrupt(agent_session: AgentSession) → bool¶

notice_session_end(agent_session: AgentSession, action: AgentAction)¶

notice_session_iteration(agent_session: AgentSession, action: AgentAction)¶

notice_session_start(agent_session: AgentSession)¶

abstract react(agent_session: AgentSession)¶: 迭代解决问题 :param agent_session: 历史迭代几率

static remove_start(origin_str: str, remove: str)¶

remove_tool(tool_id: str)¶: 删除tool :param tool_id: 工具id

run(user_input: Union[List[ConversationMessage], str, AgentSession]) → AgentSession¶: 执行agent :param user_input: 用户的输入, 支持多轮List[ConversationMessage]或str或AgentSession :return: 计划的结果

run_step(agent_session: AgentSession) → AgentSession¶: 单步执行Agent，提供干预能力 :param agent_session: 初始状态或运行中状态的agentSession，包括历史消息及其执行步骤，可以使用AgentSessionHelper类辅助处理 :return: Agent执行的结果

set_max_iterations(iterations: int)¶: 设置最大迭代次数 :param iterations: 次数

set_stream_callback(text_stream_callback: StreamCallbackHandler, tool_stream_callback: StreamCallbackHandler)¶: 设置流式输出回调函数 :param text_stream_callback: Agent文本输出的StreamCallback :param tool_stream_callback: Agent工具输出的StreamCallBack

set_tool_retriever(tool_retriever: ToolRetriever)¶: 设置工具检索器 :param tool_retriever: 工具检索器

static sub_str_after(origin_str: str, separator: str)¶

static sub_str_before(origin_str: str, separator: str)¶

static sub_str_between(origin_str: str, start_str: str, end_str: str)¶

tool_execute(tool: AbstractTool, tool_input: Union[str, dict], agent_session: AgentSession)¶

class pangukitsappdev.api.agent.base.Agent¶

基类：ABC

abstract add_listener(agent_listener: AgentListener)¶: 添加一个Agent监听器 :param agent_listener: Agent监听器

abstract add_tool(tool: AbstractTool)¶: 为Agent增加工具类 :param tool: Tool

abstract clear_tool()¶: 删除所有工具

abstract remove_tool(tool_id: str)¶: 删除tool :param tool_id: 工具id

abstract run(user_input: Union[List[ConversationMessage], str, AgentSession]) → AgentSession¶: 递归执行Agent，直到完成所有步骤的执行 :param user_input: 用户的输入, 支持多轮List[ConversationMessage]或str或AgentSession :return: 计划的结果

abstract run_step(agent_session: AgentSession) → AgentSession¶: 单步执行Agent，提供干预能力 :param agent_session: 初始状态或运行中状态的agentSession，包括历史消息及其执行步骤，可以使用AgentSessionHelper类辅助处理 :return: Agent执行的结果

abstract set_max_iterations(iterations: int)¶: 设置最大迭代次数 :param iterations: 次数

abstract set_stream_callback(text_stream_callback: StreamCallbackHandler, tool_stream_callback: StreamCallbackHandler)¶: 设置流式输出回调函数 :param text_stream_callback: Agent文本输出的StreamCallback :param tool_stream_callback: Agent工具输出的StreamCallBack

abstract set_tool_retriever(tool_retriever: ToolRetriever)¶: 设置工具检索器 :param tool_retriever: 工具检索器

class pangukitsappdev.api.agent.base.AgentListener¶

基类：ABC

Agent监听，允许对Agent的各个阶段进行处理

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

on_session_end(agent_session: AgentSession)¶: Session结束时调用 :param agent_session: AgentSession

on_session_iteration(agent_session: AgentSession)¶: Session迭代过程中调用 :param agent_session: AgentSession

on_session_start(agent_session: AgentSession)¶: Session启动时调用 :param agent_session: AgentSession

class pangukitsappdev.api.agent.base.AgentSessionHelper¶

基类：object

AgentSession辅助类

FINAL_ACTION = 'FINAL_ANSWER'¶

static init_agent_session(user_message: Union[str, List[ConversationMessage]]) → AgentSession¶: 使用用户消息初始化一个AgentSession :param user_message: 用户消息 :return: agent_session

static print_plan(agent_session: AgentSession) → str¶: 打印AgentSession :param agent_session: AgentSession :return: AgentSession的字符串输出

static set_tool_output(agent_session: AgentSession, observation: str)¶: 向Agent的当前步骤设置工具返回结果 :param agent_session: AgentSession :param observation: 工具返回结果

static set_user_feedback(agent_session: AgentSession, user_feedback: str)¶: 向Agent的当前步骤设置用户反馈 :param agent_session: AgentSession :param user_feedback: 用户反馈

static update_assistant_message(agent_session: AgentSession, override: bool)¶: 使用当前的action更新AssistantMessage的内容 :param agent_session: AgentSession :param override: 是否覆盖已有的内容

class pangukitsappdev.api.agent.base.AgentStreamCallBack(text_stream_callback: StreamCallbackHandler, tool_stream_callback: StreamCallbackHandler, llm_module_config: LLMModuleConfig)¶

基类：StreamCallbackHandler

on_chat_model_start(serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any) → Any¶: 对话模型开始时运行 :param serialized: 序列化model :param messages: 对话信息

on_llm_end(response: LLMResult, **kwargs: Any) → Any¶: 对话模型结束时运行 :param response: 模型结果

on_llm_error(error: BaseException, **kwargs: Any) → Any¶: 对话模型出错时运行 :param error: 异常信息

on_llm_new_token(token: str, **kwargs: Any) → Any¶: 流式每个新token生成时调用 :param token: 流式生成token

Module contents¶