Skip to content

Agent智能体

Agent

Agent 是所有 Agent 的抽象基类 (ABC),是整个框架的核心骨架。它的设计理念是 "Agent 持有不变配置,Session 持有可变状态"。主要职责如下:

核心职责

职责说明
配置管理持有 LLM、system_prompt、Config、ToolRegistry 等不变配置
Session 生命周期管理创建、切换、销毁、保存/恢复 Session,_sessions 字典 + RLock 保证线程安全
钩子系统两层钩子:Agent 级钩子 + Session 级钩子
消息历史委托add_messageget_historyclear_history 等操作委托给当前活跃的 Session
工具状态管理每轮对话前后自动保存/恢复工具状态到 Session 中
线程安全使用 contextvars 实现每个执行上下文独立的 active_session 和历史记录

Agent基类构造函数参数说明

参数类型默认值说明
namestrNoneAgent实例的名称
llmVioletAgentsLLMVioletAgentsLLM()LLM(大语言模型)调用接口,封装了 OpenAI 兼容 API 的客户端
system_promptOptional[str]None系统提示词
configOptional[Config]Config()Agent的配置参数,不赋值则使用默认配置参数
tool_registryOptional[ToolRegistry]ToolRegistry()管理工具的注册表

关键方法

  • run() — 主运行入口,解析 session → 切换 → 触发 TurnStart 钩子 → 调用 do_run → 触发 TurnEnd 钩子
  • do_run()抽象方法,子类必须实现,定义具体的 Agent 处理逻辑
  • create_session() — 创建新 Session,触发 SessionInit 钩子
  • switch_session() — 切换 Session,保存当前工具状态,恢复新 Session 的工具状态
  • session() — 上下文管理器,在指定 Session 内执行代码块
  • save_session() / restore_session() — Session 序列化与反序列化

钩子事件

  • Agent 级别SessionInitPreSessionSwitchPostSessionSwitch
  • Session 级别TurnStartTurnEnd

SubAgent

SubAgentAgent子类,代码非常简洁。它的唯一特殊之处在于:使用独立的环境变量配置 LLM

具体来说,它从环境变量 SUB_AGENT_LLM_API_KEYSUB_AGENT_LLM_BASE_URLSUB_AGENT_LLM_MODEL 中读取配置来创建 LLM 实例,而不是使用主 Agent 的默认 LLM 配置。这使得 SubAgent 可以使用不同的模型(比如更便宜或更快的模型)来执行子任务。

ReactAgent类

ReactAgent 是一个基于 ReAct(Reasoning + Acting)框架 设计的 AI Agent,核心作用是让 LLM 能够自主调用工具来完成复杂任务

使用示例:

python
from violet_agents import ReactAgent
from violet_agents import VioletAgentsLLM
from violet_agents import ToolRegistry
from violet_agents import SkillsTool, TerminalTool
from violet_agents import DefaultApprovalTool

tool_registry = ToolRegistry()
tool_registry.register_tools(SkillsTool(), TerminalTool())
agent = ReactAgent(
    name="VioletAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    tool_registry=tool_registry,
    max_steps=10
)
response = agent.run("今天北京多少度?")

ReactAgent独有的构造参数说明

参数类型默认值说明
max_stepsint5最大思考步骤次数

run()方法

run()方法有两个参数:

  • input_text:用户输入文本
  • session_id:可选,指定使用的 session ID。不存在则自动创建。

run()的返回类型为Message

调用方式

WARNING

在使用run()方法时,若在使用时未指定session_id,那么在下一次调用时所使用的session_id默认为最后一次调用的session_id,比如调用方式如下:

python
agent = SimpleAgent(
    name="helloAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    system_prompt="你是一个简单的助手,直接回答用户的问题,不进行复杂的思考和计划。"
)

# 第一次agent.run()
response = agent.run("你好,我叫Violet", session_id="user-123")

# 第二次agent.run()
response = agent.run("你好,我叫Violet")


with agent.session("user-456"):
	# 第三次agent.run()
	response = agent.run("你好,我叫ByteKnight")

# 第四次agent.run()
response = agent.run("你好,我叫什么")

第一次、第二次、第四次使用的session_id都为"user-123"

第三次使用的session_id为"user-456"

方式一:指定session_id运行

创建完agent实例后,可以直接指定session_id运行(agent实例里若没有对应Session则会自动创建)

python
agent = SimpleAgent(
    name="helloAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    system_prompt="你是一个简单的助手,直接回答用户的问题,不进行复杂的思考和计划。"
)

response = agent.run("你好,我叫Violet", session_id="user-123")

方式二:上下文管理器

你可以使用上下文管理器更优雅的管理会话

python
with agent.session("user-456"):
	response = agent.run("你好,我叫ByteKnight")
    response = agent.run("还记得我叫什么吗?")

方式三:手动管理

你也可以手动管理会话

python
agent.create_session("user-789")
agent.switch_session("user-789")
response = agent.run("你好,我叫Violet")
print(response.content)
agent.create_session("user-456")
agent.switch_session("user-456")
response = agent.run("你好,我叫ByteKnight")
print(response.content)
agent.switch_session("user-789")
response = agent.run("你好,我叫什么")
print(response.content)

为Agent智能体提供Tool工具

如果你想为Agent实例提供Tool工具,你需要在ToolRegistry实例中注册Tool,并将ToolRegistry实例传给Agent实例。以下代码将使用注册内置Tool作为参考

python
from violet_agents import ReactAgent
from violet_agents import VioletAgentsLLM
from dotenv import load_dotenv
from violet_agents.tools import ToolRegistry
from violet_agents.tools import SkillsTool, TerminalTool

load_dotenv()
tool_registry = ToolRegistry()
tool_registry.register_tools(SkillsTool(), TerminalTool())
agent = ReactAgent(
    name="VioletAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    tool_registry=tool_registry
)
response = agent.run("帮我看看当前目录下的文件,并告诉我有哪些.py文件")
print(response.content)

ToolRegistry具有审批工具调用功能,具体信息请移步到Tools 工具中,以下展示如何简单使用这个功能

python
from violet_agents import ReactAgent
from violet_agents import VioletAgentsLLM
from dotenv import load_dotenv
from violet_agents import ToolRegistry
from violet_agents import SkillsTool, TerminalTool
from violet_agents import DefaultApprovalTool

load_dotenv()
tool_registry = ToolRegistry(DefaultApprovalTool(require_approval_tools=[TerminalTool],
                                                max_attempts=5,
                                                auto_approve_if_no_rules=True))
tool_registry.register_tools(SkillsTool(), TerminalTool())
agent = ReactAgent(
    name="VioletAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    tool_registry=tool_registry
)
response = agent.run("帮我看看当前目录下的文件,并告诉我有哪些.py文件")
print(response.content)

Hook钩子事件

Agent基类有以下Hook钩子事件,会在适当时机触发对应钩子,Agent实现类可根据需求拓展Hook钩子事件,比如内置的ReactAgent拓展了PreToolCall, PostToolCallSession级别Hook钩子事件

  • Agent级别(作用对象为整个Agent实例)
    • SessionInit:Session初始化时触发
    • PreSessionSwitch:Session切换前触发
    • PostSessionSwitch:Session切换后触发
  • Session级别(作用对象为Agent实例的具体对话轮次)
    • TurnStart:对话轮次开始前(即执行do_run()前)
    • TurnEnd:对话轮次结束后(即执行完do_run()后)

如何使用Hook钩子

如果想要使用Agent提供的Hook钩子,推荐以下使用方式

python
from violet_agents import SimpleAgent
from violet_agents import VioletAgentsLLM

agent = SimpleAgent(
    name="helloAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    system_prompt="你是一个简单的助手,直接回答用户的问题,不进行复杂的思考和计划。"
)

agent.registry_agent_hook("SessionInit", lambda session: print(f"Session {session.session_id} has been initialized."))
agent.register_session_hook("TurnStart", lambda text: print(f"Turn started with input: {text}"))
agent.register_session_hook("TurnEnd", lambda response: print(f"Turn ended with response: {response.content}"))
agent.run("你好,我叫Violet")

register_agent_hook()register_session_hook()方法为主流编辑器(如VS CODE)提供了良好的代码提示,你可以顺滑的使用它们

Session会话管理

设计理念:一个Agent实例可以管理多个Session

DANGER

注意:如果不将Agent实例中的Session数据持久化,在Agent实例销毁后Session会话也会跟着消失!

取出Session数据

当你需要你的Agent支持持久化,你需要使用save_session()方法来获取Agent实例管理的指定Session数据

python
from violet_agents import ReactAgent
from violet_agents import VioletAgentsLLM
from dotenv import load_dotenv
from violet_agents import ToolRegistry
from violet_agents import SkillsTool, TerminalTool
from violet_agents import DefaultApprovalTool

load_dotenv()
tool_registry = ToolRegistry(DefaultApprovalTool(require_approval_tools=[TerminalTool],
                                                max_attempts=5,
                                                auto_approve_if_no_rules=True))
tool_registry.register_tools(SkillsTool(), TerminalTool())
agentA = ReactAgent(
    name="VioletAgent",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    tool_registry=tool_registry
)
response = agentA.run("帮我看看当前目录下的文件,并告诉我有哪些.py文件", session_id="user-123")
session_data = agentA.save_session("user-123")

获取到数据后,你可以选择落库(比如 MySQL, SQLite等关系型数据库),或单纯导出到.json文件统一管理,待需要使用对应的Session数据再取出即可

加载Session数据

使用restore_session()方法,你可以将Session数据加载到对应的Agent实例中

python
response = agentA.run("帮我看看当前目录下的文件,并告诉我有哪些.py文件", session_id="user-123")
session_data = agentA.save_session("user-123")
agentB = ReactAgent(
    name="VioletAgentB",
    llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
    tool_registry=tool_registry
)
agentB.restore_session(session_data)
agentB.switch_session("user-123")
response = agentB.run("之前我让你干嘛来着?我的名字叫什么")
print(response.content)

自定义Agent实现类

如果你想创建自己的Agent实现类,你只需要实现do_run()方法即可,比如:

python
class SimpleAgent(Agent):
    """简单Agent,一次对话立即响应,不进行复杂的思考和计划"""

    def __init__(self,
                 name: str,
                 llm: VioletAgentsLLM,
                 system_prompt: Optional[str] = None,
                 config: Optional[Config] = None):
        super().__init__(name, llm, system_prompt, config)

    def do_run(self, 
               input_text: str, 
               session: "Session") -> Message:
        sess = session

        user_message = Message(content=input_text, role="user")
        history = sess.get_history()
        history.append(user_message)
        messages = history

        response = self.llm.chat(messages=messages)
        response_text = response.choices[0].message.content
        response_message = Message(content=response_text, role="assistant")

        sess.add_message(user_message)
        sess.add_message(response_message)
        return response_message

需要注意的是,在do_run()中你需要修改Session实例内部属性来实现Session状态的持久化维护,例如对话状态、工具状态等,对于对话状态,推荐使用Session实例的get_history(), add_message(), clear_history()进行维护。如果你为你的Agent实现类额外添加可变状态,考虑使用Session实例的agent_state属性,并通过session.agent_state.setDefault()session.agent_state[]来修改它,比如

python
from violet_agents import Agent, VioletAgentsLLM, Message, Session, Config
from typing import Optional

class DIYAgent(Agent):
    """简单Agent,一次对话立即响应,不进行复杂的思考和计划"""

    def __init__(self,
                 name: str,
                 llm: VioletAgentsLLM,
                 system_prompt: Optional[str] = None,
                 config: Optional[Config] = None):
        super().__init__(name, llm, system_prompt, config)

    def do_run(self, 
               input_text: str, 
               session: "Session") -> Message:
        sess = session
        # 额外添加的可变状态属性
        sess.agent_state.setdefault("turn", 0)
        user_message = Message(content=input_text, role="user")
        history = sess.get_history()
        history.append(user_message)
        messages = history

        response = self.llm.chat(messages=messages)
        response_text = response.choices[0].message.content
        response_message = Message(content=response_text, role="assistant")
        # 修改额外添加的可变状态属性
        sess.agent_state["turn"] += 1
        print(f"Turn {sess.agent_state['turn']}: User input: {input_text}")
        sess.add_message(user_message)
        sess.add_message(response_message)
        return response_message

自定义Hook钩子事件

如果你想为你的Agent实现类额外添加自定义Hook钩子事件,推荐采用以下方式:

  1. Agent实现类中定义Hook钩子事件
python
    if TYPE_CHECKING:
        @overload
        def register_session_hook(self, event: Literal["UserPromptSubmit"],
                        callback: Callable[[str], None],
                        session_id: Optional[str] = None) -> None: ...
        
        @overload
        def _trigger_session_hooks(self, event: Literal["UserPromptSubmit"], arg: str, sess: Optional["Session"] = None) -> None: 
            """用户提交输入时触发的钩子
            
            Args:
                event (Literal["UserPromptSubmit"]): 事件类型
                arg (str): 用户输入的文本
                sess (Optional[Session], optional): 会话对象. Defaults to None.
                """
            ...
  1. 在对应时机触发自定义Hook钩子事件

    python
        def do_run(self, 
                   input_text: str, 
                   session: "Session") -> Message:
            # ...省略
            
    		# 触发自定义Hook钩子事件
            self._trigger_session_hooks("UserPromptSubmit", input_text)
            response = self.llm.chat(messages=messages)
            # ... 省略
            return response_message
  2. 注册自定义Hook钩子事件,有两种注册方式

    1. 供内部注册,推荐在构造函数中进行注册

      python
          def __init__(self,
                       name: str,
                       llm: VioletAgentsLLM,
                       system_prompt: Optional[str] = None,
                       config: Optional[Config] = None):
              super().__init__(name, llm, system_prompt, config)
              self.register_session_hook("UserPromptSubmit", lambda text: print(f"User submitted input: {text}"))
    2. 供外部注册

      python
      agent.register_session_hook("UserPromptSubmit", lambda text: print(f"User submitted input: {text}"))

一般都不会自定义Agent级别的Hook钩子,因为Agent级别Hook钩子作用对象为整个Agent实例,旨在Agent实例触发某个动作后调用对应的钩子函数。