Agent智能体
Agent 类
Agent 是所有 Agent 的抽象基类 (ABC),是整个框架的核心骨架。它的设计理念是 "Agent 持有不变配置,Session 持有可变状态"。主要职责如下:
核心职责
| 职责 | 说明 |
|---|---|
| 配置管理 | 持有 LLM、system_prompt、Config、ToolRegistry 等不变配置 |
| Session 生命周期管理 | 创建、切换、销毁、保存/恢复 Session,_sessions 字典 + RLock 保证线程安全 |
| 钩子系统 | 两层钩子:Agent 级钩子 + Session 级钩子 |
| 消息历史委托 | add_message、get_history、clear_history 等操作委托给当前活跃的 Session |
| 工具状态管理 | 每轮对话前后自动保存/恢复工具状态到 Session 中 |
| 线程安全 | 使用 contextvars 实现每个执行上下文独立的 active_session 和历史记录 |
Agent基类构造函数参数说明
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name | str | None | Agent实例的名称 |
llm | VioletAgentsLLM | VioletAgentsLLM() | LLM(大语言模型)调用接口,封装了 OpenAI 兼容 API 的客户端 |
system_prompt | Optional[str] | None | 系统提示词 |
config | Optional[Config] | Config() | Agent的配置参数,不赋值则使用默认配置参数 |
tool_registry | Optional[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 级别:
SessionInit、PreSessionSwitch、PostSessionSwitch - Session 级别:
TurnStart、TurnEnd
SubAgent 类
SubAgent 是 Agent 的子类,代码非常简洁。它的唯一特殊之处在于:使用独立的环境变量配置 LLM。
具体来说,它从环境变量 SUB_AGENT_LLM_API_KEY、SUB_AGENT_LLM_BASE_URL、SUB_AGENT_LLM_MODEL 中读取配置来创建 LLM 实例,而不是使用主 Agent 的默认 LLM 配置。这使得 SubAgent 可以使用不同的模型(比如更便宜或更快的模型)来执行子任务。
ReactAgent类
ReactAgent 是一个基于 ReAct(Reasoning + Acting)框架 设计的 AI Agent,核心作用是让 LLM 能够自主调用工具来完成复杂任务。
使用示例:
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_steps | int | 5 | 最大思考步骤次数 |
run()方法
run()方法有两个参数:
input_text:用户输入文本session_id:可选,指定使用的 session ID。不存在则自动创建。
run()的返回类型为Message
调用方式
WARNING
在使用run()方法时,若在使用时未指定session_id,那么在下一次调用时所使用的session_id默认为最后一次调用的session_id,比如调用方式如下:
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则会自动创建)
agent = SimpleAgent(
name="helloAgent",
llm=VioletAgentsLLM(provider="deepseek", model="deepseek-v4-flash"),
system_prompt="你是一个简单的助手,直接回答用户的问题,不进行复杂的思考和计划。"
)
response = agent.run("你好,我叫Violet", session_id="user-123")方式二:上下文管理器
你可以使用上下文管理器更优雅的管理会话
with agent.session("user-456"):
response = agent.run("你好,我叫ByteKnight")
response = agent.run("还记得我叫什么吗?")方式三:手动管理
你也可以手动管理会话
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作为参考
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 工具中,以下展示如何简单使用这个功能
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, PostToolCall等Session级别Hook钩子事件。
- Agent级别(作用对象为整个
Agent实例)SessionInit:Session初始化时触发PreSessionSwitch:Session切换前触发PostSessionSwitch:Session切换后触发
- Session级别(作用对象为
Agent实例的具体对话轮次)TurnStart:对话轮次开始前(即执行do_run()前)TurnEnd:对话轮次结束后(即执行完do_run()后)
如何使用Hook钩子
如果想要使用Agent提供的Hook钩子,推荐以下使用方式
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数据
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实例中
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()方法即可,比如:
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[]来修改它,比如
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钩子事件,推荐采用以下方式:
- 在
Agent实现类中定义Hook钩子事件
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.
"""
...在对应时机触发自定义Hook钩子事件
pythondef 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注册自定义Hook钩子事件,有两种注册方式
供内部注册,推荐在构造函数中进行注册
pythondef __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}"))供外部注册
pythonagent.register_session_hook("UserPromptSubmit", lambda text: print(f"User submitted input: {text}"))
一般都不会自定义Agent级别的Hook钩子,因为Agent级别Hook钩子作用对象为整个Agent实例,旨在Agent实例触发某个动作后调用对应的钩子函数。