引言

在「潇楠WEB3哨兵」这个多链监控交易系统里,EVM/SOL 双链监控负责“眼睛”,电报 Bot 负责“嘴巴”,而 Agent 是最后一个也是最重要的拼图——它是整个软件的“大脑”。

它不是一个独立的聊天机器人。它能读取本地监控数据库里的历史交易,能调用合约分析工具实时解读市场,能通过右下角弹窗主动联系用户,甚至能从你们的长期对话中提炼你的交易风格。

本文将从架构层面,逐一拆解 Agent 的六个核心技术模块:Mixin 混入架构、多模型智能路由、五层记忆体系、身份固化与后悔药、零代码技能扩展、定时自驱与主动感知。每个模块都有代码、有原理、有坑与反思。


一、Mixin 混入架构:Agent 如何嵌入桌面应用

Agent 不是一个独立进程。它和主窗口、电报助手、SWAP 模块共享同一个 Python 环境,通过 Mixin 模式混入到 Api 类中。

为什么不用独立进程?三个原因:

  1. Agent 需要直接访问主进程的 window 对象,用来弹出右下角通知。

  2. Agent 需要访问同一个 SQLite 数据库连接,用来读写记忆。

  3. Agent 的定时任务需要在同一个 asyncio 事件循环里执行,不能和 WSS 监控抢循环。

混入链

main.py 里的 Api 类继承链是这样的:

        class Api(ActivationMixin, AIMixin, CMCMixin, CryptoRankMixin,
             DexScreenerMixin, GeckoMixin, GoPlusMixin, StatsMixin,
             ConfigMixin, OtherMixin, TagsMixin, SwapMixin, AgentMixin):

AgentMixin 排在最后,但它不是最不重要的——恰恰相反,它是唯一一个需要延迟初始化的 Mixin。

延迟初始化的坑

Agent 需要在前端窗口完全渲染后才能启动,否则 window.evaluate_js 会失败。但 Api.__init__ 在窗口创建之前就执行了。所以 AgentMixin.__init__ 不能在 Api.__init__ 里直接调用。

解决方案是延迟初始化:

python

def after_start():
    window.maximize()
    api._inject_bridge()

    def delayed_agent_start():
        time.sleep(3)  # 等主界面完全加载
        if not api._agent_ready:
            AgentMixin.__init__(api)  # 手动触发 Agent 初始化
            api._agent_ready = True
        agent_cfg = api._load_agent_config()
        if agent_cfg.get("enabled", False):
            api._start_agent()

    threading.Thread(target=delayed_agent_start, daemon=True).start()

这里的 _agent_ready 标记防止重复初始化——因为 AgentMixin.__init__ 里会创建数据库表、启动后台线程,重复执行会导致线程泄漏。

打包环境下的路径自适应

Agent 的数据库和配置文件需要在开发环境和打包环境下都能正确定位。核心方法是判断 sys.frozen

python

if getattr(sys, 'frozen', False):
    self.base_path = os.path.dirname(sys.executable)
    self.bundle_dir = sys._MEIPASS
else:
    self.base_path = os.path.dirname(os.path.abspath(__file__))
    self.bundle_dir = self.base_path

所有文件路径——agent_memory.dbagent_config.jsonskills/ 目录——都通过 self.base_path 拼接,保证在任何环境下都能找到。


二、多模型智能路由:为不同任务分配不同大脑

「潇楠WEB3哨兵」里集成了三个 AI 模型:DeepSeek(无联网)、Moonshot(联网搜索)、Gemma(降级备用)。Agent 需要根据任务类型自动选择最合适的模型。

三模型定位

模型 定位 适合场景
DeepSeek 主力规划 代币分析、合约解读、复杂推理
Moonshot 联网搜索 新闻查询、最新项目背景
Gemma 降级兜底 前两者都失败时的备胎

路由逻辑

核心在 agent_process_input 方法里:

python

def agent_process_input(self, user_input: str):
    # ... 保存用户消息到 chat_history ...
    
    # 联网搜索判定
    need_web_search = any(kw in user_input.lower() 
        for kw in ["新闻", "最新", "今天", "实时", "快讯", "搜索"])
    
    # 路由选择
    primary_model = "moonshot" if need_web_search else "deepseek"
    
    # 调用
    reply = self._call_model_with_tools(messages, primary_model)
    
    # 降级
    if reply is None and primary_model != "deepseek":
        self._push_message("⚠️ Moonshot 调用失败,切换到 DeepSeek")
        reply = self._call_model_with_tools(messages, "deepseek")
    
    if reply is None:
        reply = "抱歉,AI 服务暂时不可用,请稍后重试。"

这里的降级链路设计很重要——不是简单的 try-catch,而是主动切换。Moonshot 挂了不意味着服务不可用,DeepSeek 可能还活着。

坑:模型超时后的死等

早期版本没有 timeout 机制,一旦 Moonshot 卡住,整个 Agent 就僵死了。后来在每个模型调用里加了 timeout 参数,超时后抛出异常,触发降级链路。


三、五层记忆体系:Agent 如何记住你是谁

这是 Agent 最核心的能力——没有记忆的 AI 只是玩具。网页版 AI 的“金鱼记忆”在专业投资场景下是致命的:你今天告诉它你喜欢 Solana 上的新币,明天它就忘了。

「潇楠WEB3哨兵」的 Agent 通过 SQLite 构建了五层记忆架构。

五张核心表

sql

-- 1. 通用记忆(KV 存储)
CREATE TABLE agent_memory (
    category TEXT,       -- master_profile / agent_profile / chat_summary
    key TEXT UNIQUE,
    content TEXT,        -- JSON 字符串
    timestamp DATETIME
);

-- 2. 聊天历史
CREATE TABLE chat_history (
    session_id TEXT,
    role TEXT,           -- user / assistant
    content TEXT,
    timestamp DATETIME
);

-- 3. 偏好统计
CREATE TABLE preference_stats (
    category TEXT,       -- chain / interest
    value TEXT,
    count INTEGER,
    last_updated DATETIME
);

-- 4. 系统事件日志
CREATE TABLE system_events (
    event_type TEXT,
    content TEXT,
    timestamp DATETIME
);

-- 5. 固化身份备份
CREATE TABLE locked_profile (
    profile_type TEXT UNIQUE,   -- master / agent
    content TEXT,               -- JSON
    locked_at DATETIME
);

记忆的分层逻辑

短期记忆:直接取最近 20 轮对话放进上下文窗口,这是最直接的“记住刚才在聊什么”。

中期记忆:当累计 50 轮对话后,触发 _compress_history_if_needed。它不是简单截断,而是让 AI 把 50 轮对话压缩成一段 300 字的摘要,存入 agent_memory 表。同时保留最近 5 条原始消息,确保上下文不丢失。

python

def _compress_history_if_needed(self, session_id: str, max_rounds: int = 50):
    # 只压缩 50 轮以上的对话
    # 保留包含合约地址的消息、用户的明确指令、AI 的关键结论
    # 压缩结果存入 agent_memory 表

长期记忆:主人卡片和 Agent 自我认知永久存储在 agent_memory 表里,除非用户手动修改,否则不会被压缩或删除。

反思记忆:这是最“高级”的记忆层。Agent 每 6 小时自动回顾最近 60 条对话,提炼出主人的交易偏好——喜欢做多还是做空、偏好哪些链、持仓周期是短线还是长线。

python

def _reflect_and_remember(self):
    while self.agent_running:
        time.sleep(21600)  # 6小时

        # 读取最近 60 条聊天记录
        rows = conn.execute(
            "SELECT role, content FROM chat_history ORDER BY timestamp DESC LIMIT 60"
        ).fetchall()

        # 拼接对话,发给 AI 提炼
        summary = self._call_ai(prompt, max_tokens=300)

        # 覆盖写入 memory
        self._agent_remember("master_insight", "auto_reflection", summary)

坑与反思:反思结果必须覆盖写入,不能追加。否则 30 天后你的 Agent 会有 120 条反思记录,上下文窗口直接撑爆。

统计记忆preference_stats 表会自动记录你提到过的链和兴趣领域,用于 Agent 主动搭讪时的话题选择。


四、身份固化与“后悔药”:深度个性化的安全机制

记忆写入是动态的,而“理想人格”应该是可备份、可恢复的。你花了三个月打磨出来的完美 Agent 人格,可能因为一次错误的对话就被污染了。

「潇楠WEB3哨兵」设计了一套“后悔药”机制。

固化身份

用户在 Agent 窗口输入 Admin: 固化当前身份,Agent 会把当前的主人信息和自我认知序列化为 JSON,存入 locked_profile 表:

python

if user_input.strip() == "Admin: 固化当前身份":
    master = self._get_master_info()
    agent = self._get_agent_profile()
    conn = sqlite3.connect(self.agent_memory_db)
    conn.execute("INSERT OR REPLACE INTO locked_profile (profile_type, content) VALUES (?,?)",
                 ("master", json.dumps(master, ensure_ascii=False)))
    conn.execute("INSERT OR REPLACE INTO locked_profile (profile_type, content) VALUES (?,?)",
                 ("agent", json.dumps(agent, ensure_ascii=False)))

恢复身份

输入 Admin: 恢复固化身份,Agent 从 locked_profile 表读取备份,覆盖当前记忆:

python

if user_input.strip() == "Admin: 恢复固化身份":
    master_row = conn.execute("SELECT content FROM locked_profile WHERE profile_type = 'master'").fetchone()
    if master_row:
        self._agent_remember("master_profile", "owner_info", master_row[0])

清洗污染

如果 Agent 被某个话题带偏了,可以用 Admin: 删除关于 XXX 的所有对话记录 一键清洗:

python

if user_input.startswith("Admin: 删除关于") and "的所有对话记录" in user_input:
    keyword = user_input.replace("Admin: 删除关于", "").replace("的所有对话记录", "").strip()
    conn.execute("DELETE FROM chat_history WHERE content LIKE ?", (f"%{keyword}%",))
    conn.execute("DELETE FROM agent_memory WHERE content LIKE ? AND category = 'chat_summary'", (f"%{keyword}%",))

这三条 Admin 指令构成了完整的“人格保险”体系——固化、恢复、清洗,三者配合,让 Agent 的人格既能进化,也有回滚的安全网。


五、零代码技能扩展:教 Agent 新能力

传统的方式是改 api_agent.py 代码,增加新的 elif tool_name == "xxx" 分支。但这样会污染核心代码,而且每次升级都需要重新打包。

「潇楠WEB3哨兵」实现了一套不依赖 Python 代码的技能扩展机制。

HTML 技能说明书

在软件根目录下新建 skills/ 文件夹,放一个 HTML 文件,比如 skills/代币解锁分析.html

html

<h1>代币解锁分析</h1>
<p>触发条件:用户询问“代币解锁”、“解锁分析”时使用此技能。</p>
<p>执行步骤:</p>
<ol>
<li>查询持币地址分布,关注前10大持仓地址的占比。</li>
<li>分析筹码集中度是否过高(超过 50% 则为高风险)。</li>
<li>输出风险等级和仓位建议。</li>
</ol>

Agent 启动时会自动扫描 skills/ 文件夹,提取 HTML 中的纯文本,拼接到系统提示词中:

python

def _load_external_skills(self) -> str:
    skills_dir = os.path.join(self.base_path, 'skills')
    for filename in os.listdir(skills_dir):
        if filename.endswith('.html'):
            # 提取纯文本
            text = re.sub(r'<[^>]+>', ' ', html_content)
            all_skills.append(f"【技能:{filename}】\n{text}")

内置技能与外部技能的关系

Agent 内置了 16 个技能(合约分析、代币检测、新闻抓取等),这些通过 Function Calling 在代码中实现。外部技能则是纯文本的“工作手册”,告诉 Agent “怎么做”。

两者的隔离设计很重要:外部技能不写入数据库,只存在于内存中。删掉 skills/ 文件夹里的文件,重启软件,Agent 就忘了这个技能。不会污染记忆,也不需要改任何代码。


六、定时自驱与主动感知:Agent 的生物钟

被动应答的 Agent 只是工具。真正有价值的 Agent,应该有自己的“生物钟”——能在设定的时间主动问候用户,能在检测到市场异常时主动弹窗预警。

定时问候与随机搭讪

_agent_loop 是 Agent 的主循环,运行在守护线程中。它管理两个核心行为:

  1. 定时问候:每天在你设定的时间(如 10:00)发送早安消息

  2. 随机搭讪:在你设定的间隔内(如 2-5 小时),随机触发一次主动搭讪

搭讪内容不是硬编码的,而是让 AI 根据你的主人卡片实时生成:

python

def _generate_random_chat(self) -> str:
    master = self._get_master_info()
    name = master.get("name", "主人")
    interests = ", ".join(master.get("interests", []))
    chains = ", ".join(master.get("preferred_chains", []))

    prompt = (
        f"你是{agent_name}。你的主人叫{name},他主要关注{interests},偏好{chains}链。"
        f"现在是{time_desc}。请用一句亲切的话跟{name}搭讪,20字左右。"
    )

    result = self._call_ai(prompt, max_tokens=100)
    return result.strip() if result else random.choice(fallbacks)

右下角弹窗的线程间通信

Agent 的定时任务跑在守护线程里,而前端 UI 跑在主线程。要让 Agent 的消息显示在右下角弹窗里,需要跨线程调用 window.evaluate_js

python

def _push_message(self, message: str):
    if self.window:
        safe_msg = json.dumps(message)
        # 让前端 JS 显示右下角弹窗
        self.window.evaluate_js(
            f"if(window.showAgentNotification){{window.showAgentNotification({safe_msg})}}"
        )

:如果在窗口还没完全渲染好时就调用 evaluate_js,会导致主线程卡死。解决方法是 Agent 初始化时延迟 3 秒启动。


总结

Agent 的本质 = 记忆 + 路由 + 技能 + 生物钟 + 身份固化 + 应用穿透

六个模块协同工作,让「潇楠WEB3哨兵」从一个被动执行监控任务的工具,进化成一个真正有记忆、有判断力、有主动意识的投研伙伴。

技术的尽头不是堆砌功能,而是让软件有人味儿。当你的 Agent 早上 10 点主动发来一条消息说“老板早,BSC 上新上了一个代币,要我帮你扫一下吗”,你就知道——它不是工具,它是你团队里的第一号数字员工。

GITHUB:https://github.com/pingdj/Web3

Logo

一站式 AI 云服务平台

更多推荐