GenericAgent 源码解析:3000 行代码构建的极简自主代理

GenericAgent 源码解析:3000 行代码构建的极简自主代理

深入剖析 GenericAgent——一个仅用 3000 行核心代码 + 9 个原子工具 + 100 行 Agent Loop,赋予任意 LLM 对本地计算机的系统级控制能力的自主代理框架

项目简介

GenericAgent 是一个极简的、可自我进化的自主代理框架。在 AI Agent 框架越来越复杂的今天(动辄几万行代码、数十个模块),GenericAgent 反其道而行——核心代码仅约 3000 行,却实现了强大的自主代理能力。

核心理念:不预装技能,而是进化技能。每次完成任务后,自动将执行路径结晶为可复用的 Skill。

核心能力

  • 🌐 浏览器控制:真实浏览器注入,保留登录状态
  • 💻 终端执行:在本地执行任意命令
  • 📁 文件系统操作:读写、搜索、删除文件
  • ⌨️ 键盘鼠标输入:模拟用户操作
  • 👁️ 屏幕视觉:截屏并理解屏幕内容
  • 📱 移动设备控制:通过 ADB 控制 Android 设备

整体架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
┌─────────────────────────────────────────────────────────────┐
│ 前端层 (Frontends) │
│ TUI │ GUI │ Web │ CLI │ Desktop │ DingTalk │ Discord │ ... │
└────────────────────┬────────────────────────────────────────┘

┌────────────────────▼────────────────────────────────────────┐
│ Agent 主程序 (agentmain.py) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ GenericAgent - 代理实例管理 │ │
│ │ - 配置加载、LLM 切换、任务队列 │ │
│ └──────────────────────────────────────────────────────┘ │
└────────────────────┬────────────────────────────────────────┘

┌────────────────────▼────────────────────────────────────────┐
│ Agent 循环引擎 (agent_loop.py) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ agent_runner_loop - 核心循环 │ │
│ │ - 消息处理、工具调用、流式响应 │ │
│ └──────────────────────────────────────────────────────┘ │
└────────────────────┬────────────────────────────────────────┘

┌────────────────────▼────────────────────────────────────────┐
│ 核心处理器 (ga.py) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ GenericAgentHandler - 工具执行与状态管理 │ │
│ │ - 工具注册、执行、结果处理 │ │
│ │ - 记忆管理、技能进化 │ │
│ └──────────────────────────────────────────────────────┘ │
└────────────────────┬────────────────────────────────────────┘

┌────────────────────▼────────────────────────────────────────┐
│ LLM 核心抽象层 (llmcore.py) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ ToolClient / NativeToolClient - LLM 调用抽象 │ │
│ │ - 多模型支持(Claude/Gemini/Kimi/MiniMax 等) │ │
│ │ - 流式响应、工具调用协议适配 │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

目录结构总览

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
GenericAgent/
├── ga.py # Agent 核心处理器
├── agent_loop.py # Agent 循环引擎
├── agentmain.py # Agent 主程序
├── llmcore.py # LLM 核心抽象层
├── simphtml.py # 简单 HTML 解析器
├── TMWebDriver.py # 浏览器驱动

├── assets/ # 资源文件
│ ├── tools_schema.json # 工具模式定义
│ ├── sys_prompt.txt # 系统提示词
│ └── configure_mykey.py # 配置向导

├── frontends/ # 前端实现
│ ├── tui_v3.py # TUI (推荐)
│ ├── qtapp.py # PyQt5 GUI
│ ├── stapp.py # Streamlit Web
│ ├── dcapp.py # Discord
│ ├── dingtalkapp.py # 钉钉
│ └── ...

├── memory/ # 记忆存储
│ ├── global_mem.txt # 全局记忆
│ └── global_mem_insight.txt

├── plugins/ # 插件目录
├── reflect/ # 反思脚本
├── temp/ # 临时文件
│ └── model_responses/ # 模型响应日志

├── mykey_template.py # 配置模板
├── mykey.py # 用户配置(不提交到 Git)
├── pyproject.toml # 项目配置
└── README.md # 项目说明

关键文件说明

文件 大小 说明
ga.py 37KB Agent 核心处理器,工具执行
agent_loop.py 6.7KB Agent 循环引擎,核心循环逻辑
agentmain.py 18KB Agent 主程序,配置管理
llmcore.py 68KB LLM 核心抽象层,多模型支持
simphtml.py 42KB HTML 解析器,网页处理
TMWebDriver.py 15KB 浏览器驱动,真实浏览器控制

核心模块详解

1. Agent Loop:100 行代码的”心脏”

agent_loop.py 是整个系统最精简也最重要的模块——仅约 100 行核心代码,却实现了完整的 Agent 循环。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def agent_runner_loop(agent, task_queue, display_queue):
"""
Agent 主循环

参数:
- agent: GenericAgent 实例
- task_queue: 任务队列(输入)
- display_queue: 显示队列(输出)
"""
while True:
# 1. 从任务队列获取输入
task = task_queue.get()

# 2. 构建消息
messages = build_messages(agent, task)

# 3. 调用 LLM
response = agent.llmclient.chat(messages, tools=agent.tools)

# 4. 处理响应(文本/工具调用)
if response.has_tool_calls:
# 执行工具调用
results = execute_tool_calls(agent, response.tool_calls)
# 将结果加入历史
agent.history.append_tool_results(results)
# 继续循环(不获取新任务)
continue
else:
# 纯文本响应,输出给用户
display_queue.put(format_response(response))
# 等待下一个任务

这个循环的精妙之处在于:

  • 简洁到不能再简洁:没有复杂的中间件、没有插件系统、没有花哨的编排
  • 职责清晰:获取任务 → 构建消息 → 调用 LLM → 处理响应 → 循环
  • 生产者-消费者模式:通过 task_queuedisplay_queue 实现前后端解耦

循环状态机

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
┌─────────┐
│ 空闲 │ ← 等待任务
└────┬────┘
│ 获取任务

┌─────────┐
│ 思考 │ ← 调用 LLM
└────┬────┘

├─→ 工具调用?──┐
│ │
│ 是 │ 否
▼ ▼
┌─────────┐ ┌─────────┐
│ 执行 │ │ 输出 │
└────┬────┘ └────┬────┘
│ │
└──────┬───────┘


继续循环

关键特性

  1. 流式处理:支持流式响应,实时显示思考过程
  2. 多轮对话:自动维护对话历史
  3. 工具调用链:支持连续多次工具调用
  4. 中断恢复:支持用户中断后恢复执行
  5. 错误处理:工具执行失败时自动重试或降级

2. GenericAgentHandler:核心处理器

ga.py 中的 GenericAgentHandler 是整个系统的核心处理器,负责:

  1. 工具注册与管理

    • 维护工具模式(TOOLS_SCHEMA)
    • 工具执行与结果返回
    • 工具调用日志记录
  2. 状态管理

    • 对话历史维护
    • 工具调用状态追踪
    • 中断与恢复机制
  3. 记忆系统

    • 全局记忆(global_memory)读写
    • 工作记忆(working_memory)管理
    • 记忆导入/导出
  4. 技能进化

    • 任务执行路径结晶
    • 技能模板生成
    • 技能库管理

工具执行核心

1
2
3
4
5
6
7
8
9
10
11
12
13
def execute_tool(self, tool_name, tool_args):
"""执行单个工具并返回结果"""

# 智能格式化输出
def smart_format(text, mode='auto'):
"""根据上下文智能格式化输出文本"""

# 全局记忆管理
def get_global_memory():
"""获取全局记忆内容"""

def update_global_memory(content):
"""更新全局记忆"""

工具执行流程

1
用户输入 → 解析工具调用 → 参数验证 → 执行工具 → 格式化结果 → 返回 LLM

3. GenericAgent:主程序

agentmain.py 中的 GenericAgent 是面向用户的代理实例,封装了底层 GenericAgentHandler 的复杂性:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class GenericAgent:
def __init__(self):
# 初始化配置
self.config = load_config()

# 初始化 LLM 客户端
self.llmclient = ToolClient(self.config.llm)

# 初始化工具模式
self.tools = load_tool_schema()

# 初始化处理器
self.handler = GenericAgentHandler(self)

# 任务队列
self.task_queue = Queue()
self.display_queue = Queue()

def run(self):
"""启动代理主循环"""
threading.Thread(
target=agent_runner_loop,
args=(self, self.task_queue, self.display_queue)
).start()

def submit_task(self, task):
"""提交任务"""
self.task_queue.put(task)

def abort(self):
"""中断当前任务"""
self.handler.abort()

启动参数

1
2
3
4
5
6
7
8
9
python agentmain.py [选项]

选项:
--nolog 禁用日志
--llm_no N 指定 LLM 编号
--verbose 详细输出
--task "任务" 直接执行任务
--history FILE 加载历史记录
--reflect SCRIPT 启用反思模式

4. llmcore.py:LLM 核心抽象层

这是整个系统中最复杂的模块(约 68KB),负责与各种 LLM API 的交互。

核心类层次结构

1
2
3
4
5
6
7
8
9
ToolClient (基类)
├── NativeToolClient (原生工具调用客户端)
│ ├── NativeClaudeSession (Claude 会话)
│ ├── NativeOAISession (OpenAI 兼容会话)
│ └── NativeGeminiSession (Gemini 会话)
└── BaseSession (基础会话类)
├── ClaudeSession
├── OAISession
└── GeminiSession

ToolClient:统一接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class ToolClient:
"""统一的 LLM 调用接口"""

def __init__(self, backend):
self.backend = backend # 具体的会话实现

def chat(self, messages, tools=None):
"""
调用 LLM

参数:
- messages: 消息列表 [{"role": "user", "content": "..."}]
- tools: 工具模式列表

返回:
- 响应对象(包含文本和/或工具调用)
"""
return self.backend.chat(messages, tools)

def set_system(self, prompt):
"""设置系统提示词"""
self.backend.system = prompt

NativeToolClient:原生工具调用

专门处理支持原生工具调用的 LLM(如 Claude、GPT-4):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class NativeToolClient:
def __init__(self, backend):
self.backend = backend
self.backend.system = self._thinking_prompt()
self._pending_tool_ids = []

def chat(self, messages, tools=None):
# 1. 合并消息
merged = self._merge_messages(messages)

# 2. 调用后端
resp = self.backend.ask(merged)

# 3. 处理工具调用
if resp.tool_calls:
self._pending_tool_ids = [tc.id for tc in resp.tool_calls]

return resp

多模型支持

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# Claude (Anthropic)
class NativeClaudeSession:
def ask(self, messages):
response = self.client.messages.create(
model=self.model,
messages=messages,
tools=self.tools,
stream=True
)

# OpenAI 兼容 (GPT-4, Kimi, MiniMax 等)
class NativeOAISession:
def ask(self, messages):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools,
stream=True
)

# Gemini (Google)
class NativeGeminiSession:
def ask(self, messages):
response = self.client.generate_content(
contents=messages,
tools=self.tools,
stream=True
)

消息格式转换

不同 LLM 的消息格式不同,llmcore.py 负责统一转换:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 统一格式
[
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好!有什么可以帮助你的吗?"},
{"role": "user", "content": "打开浏览器"}
]

# Claude 格式
[
{"role": "user", "content": [{"type": "text", "text": "你好"}]},
{"role": "assistant", "content": [{"type": "text", "text": "你好!"}]},
]

# OpenAI 格式
[
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好!"},
]

工具系统

GenericAgent 的核心能力来自 9 个原子工具,这些工具通过 assets/tools_schema.json 定义。

9 个原子工具

1. browser - 浏览器控制

1
2
3
4
5
6
7
8
9
10
11
def browser(action, url=None, selector=None, text=None):
"""
控制真实浏览器(通过 TMWebDriver)

action:
- open: 打开 URL
- click: 点击元素
- type: 输入文本
- scroll: 滚动页面
- screenshot: 截图
"""

实现原理

  • 使用 TMWebDriver 注入到真实浏览器进程
  • 保留浏览器的登录状态和 Cookie
  • 支持 JavaScript 执行

2. terminal - 终端执行

1
2
3
4
5
6
7
8
9
10
11
12
13
def terminal(command):
"""
在终端执行命令

返回:命令输出(stdout + stderr)
"""
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True
)
return result.stdout + result.stderr

3. file - 文件操作

1
2
3
4
5
6
7
8
9
10
def file(action, path, content=None):
"""
文件操作

action:
- read: 读取文件
- write: 写入文件
- delete: 删除文件
- list: 列出目录
"""

4. keyboard - 键盘输入

1
2
3
4
5
6
7
8
9
def keyboard(action, key=None, text=None):
"""
键盘控制

action:
- press: 按键
- type: 输入文本
- hotkey: 组合键
"""

5. mouse - 鼠标控制

1
2
3
4
5
6
7
8
9
10
def mouse(action, x=None, y=None):
"""
鼠标控制

action:
- move: 移动
- click: 点击
- drag: 拖拽
- scroll: 滚动
"""

6. screen - 屏幕视觉

1
2
3
4
5
6
def screen(action="screenshot", region=None):
"""
屏幕截图

返回:Base64 编码的图片
"""

7. adb - 移动设备控制

1
2
3
4
5
6
7
def adb(command, device=None):
"""
ADB 命令(控制 Android 设备)

command: ADB 命令字符串
device: 设备序列号(可选)
"""

8. ask_user - 询问用户

1
2
3
4
5
6
def ask_user(question):
"""
向用户提问(用于需要人类输入的场景)

返回:用户的回答
"""

9. start_long_term_update - 长期记忆

1
2
3
4
5
6
def start_long_term_update(content):
"""
将信息写入长期记忆

用于持久化重要的上下文信息
"""

工具执行流程

1
2
3
4
5
6
7
8
9
10
11
12
13
LLM 输出工具调用

解析工具名称和参数

验证参数合法性

执行工具函数

格式化结果

返回给 LLM

LLM 继续推理

前端系统

GenericAgent 提供了极其丰富的前端界面,这是它的一个亮点——一个 Agent 核心,多种接入方式。

前端目录结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
frontends/
├── tuiapp.py # 终端 TUI (Textual)
├── tuiapp_v2.py # 增强版 TUI
├── tui_v3.py # 第三代 TUI(推荐)
├── qtapp.py # PyQt5 桌面 GUI
├── stapp.py # Streamlit Web UI
├── dcapp.py # Discord 机器人
├── dingtalkapp.py # 钉钉机器人
├── qqapp.py # QQ 机器人
├── wechatapp.py # 微信机器人
├── fsapp.py # 飞书机器人
├── desktop_pet_v2.pyw # 桌面宠物
├── conductor.py # 多代理管理器
└── desktop/ # Tauri 桌面应用
├── src-tauri/
└── src/

TUI (终端界面)

使用 Textual 框架构建的终端图形界面:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class GenericAgentTUI(App):
def __init__(self):
super().__init__()
self.agent = GenericAgent()

def compose(self):
"""构建 UI 组件"""
yield Header()
yield ChatView(id="chat") # 聊天视图
yield Input(placeholder="输入消息...") # 输入框
yield Footer()

def on_mount(self):
"""启动代理"""
self.agent.run()
self.start_message_listener()

def start_message_listener(self):
"""监听代理消息"""
def listener():
while True:
msg = self.agent.display_queue.get()
self.call_from_thread(self.update_chat, msg)

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

GUI (桌面界面)

使用 PyQt5 构建的完整桌面应用:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.agent = GenericAgent()

# UI 组件
self.chat_display = QTextEdit() # 聊天显示
self.input_box = QLineEdit() # 输入框
self.send_button = QPushButton("发送")

# 布局
layout = QVBoxLayout()
layout.addWidget(self.chat_display)
layout.addWidget(self.input_box)

# 信号槽
self.send_button.clicked.connect(self.send_message)

def send_message(self):
"""发送消息"""
text = self.input_box.text()
self.agent.submit_task(text)
self.input_box.clear()

Web UI

使用 Streamlit 构建的 Web 界面:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import streamlit as st

# 初始化代理
if 'agent' not in st.session_state:
st.session_state.agent = GenericAgent()
st.session_state.agent.run()

# 聊天界面
st.title("GenericAgent")

# 消息历史
for msg in st.session_state.messages:
st.chat_message(msg['role']).write(msg['content'])

# 输入框
if prompt := st.chat_input("输入消息..."):
st.session_state.messages.append({"role": "user", "content": prompt})
st.session_state.agent.submit_task(prompt)

# 监听响应
with st.chat_message("assistant"):
response = st.write_stream(response_stream())
st.session_state.messages.append({"role": "assistant", "content": response})

聊天机器人集成

以钉钉机器人为例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class DingTalkApp:
def __init__(self):
self.agent = GenericAgent()
self.agent.verbose = False

async def on_message(self, content, sender_id, sender_name):
"""处理钉钉消息"""
task = asyncio.create_task(
self.run_agent(sender_id, content)
)

async def run_agent(self, chat_id, content):
"""运行代理并返回结果"""
display_queue = Queue()

# 提交任务
self.agent.submit_task(content)

# 收集响应
response = ""
while True:
msg = display_queue.get()
if msg.get('done'):
break
response += msg.get('next', '')

# 发送回复
await self.send_text(chat_id, response)

多代理管理器

frontends/conductor.py 管理多个并行的代理实例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class SubAgentPool:
def __init__(self, agent):
self.agent = agent
self.subagents = {}

def create_subagent(self, task, llm_no=None):
"""创建子代理"""
sid = short_id()
subagent = GenericAgent()
subagent.next_llm(llm_no or self.agent.llm_no)

# 启动子代理
threading.Thread(
target=subagent.run,
daemon=True
).start()

# 注册到池
self.subagents[sid] = SubAgentInfo(
id=sid,
agent=subagent,
task=task,
status="running"
)

return sid

def input_subagent(self, sid, message, llm_no=None):
"""向子代理发送消息"""
subagent = self.subagents[sid]
subagent.agent.submit_task(message)

def abort_subagent(self, sid):
"""中断子代理"""
self.subagents[sid].agent.abort()

记忆系统

GenericAgent 实现了多层记忆系统,支持短期和长期记忆。

记忆层次

1
2
3
4
5
6
7
8
9
┌─────────────────────────────────────┐
│ 工作记忆 (Working Memory) │ ← 当前对话上下文
├─────────────────────────────────────┤
│ 短期记忆 (Short-term) │ ← 近期对话历史
├─────────────────────────────────────┤
│ 长期记忆 (Long-term) │ ← 持久化的重要信息
├─────────────────────────────────────┤
│ 全局记忆 (Global Memory) │ ← 跨会话的知识库
└─────────────────────────────────────┘

全局记忆

文件位置memory/global_mem.txt

1
2
3
4
5
6
7
8
9
10
11
12
13
# [Global Memory - L2]

## 用户偏好
- 喜欢使用中文交流
- 常用的开发工具:VSCode, Git

## 重要信息
- 项目名称:GenericAgent
- 创建时间:2024-06-24

## 技能库
- 网页浏览:已掌握
- 文件操作:已掌握

读写方式

1
2
3
4
5
6
7
8
9
10
11
# 读取全局记忆
def get_global_memory():
mem_path = os.path.join(script_dir, 'memory', 'global_mem.txt')
with open(mem_path, 'r', encoding='utf-8') as f:
return f.read()

# 更新全局记忆
def update_global_memory(content):
mem_path = os.path.join(script_dir, 'memory', 'global_mem.txt')
with open(mem_path, 'w', encoding='utf-8') as f:
f.write(content)

工作记忆

在每次对话中动态构建,包含:

  1. 系统提示词:定义代理的身份和能力
  2. 全局记忆:跨会话的知识库
  3. 对话历史:当前的对话上下文
  4. 工具调用结果:最近几次工具调用的结果
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def build_messages(agent, task):
"""构建消息列表"""
messages = []

# 1. 系统提示词
messages.append({
"role": "system",
"content": get_system_prompt()
})

# 2. 全局记忆
global_mem = get_global_memory()
if global_mem:
messages.append({
"role": "system",
"content": f"### [GLOBAL MEMORY]\n{global_mem}"
})

# 3. 对话历史
messages.extend(agent.history)

# 4. 当前任务
messages.append({
"role": "user",
"content": task
})

return messages

记忆导入/导出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def _import_memory_from(source_dir: str, ga_root: str):
"""
从其他 GA 实例导入记忆

1. 备份当前 memory/ 目录
2. 复制源目录的 memory/ 文件
3. 复制 temp/model_responses/ 日志
"""
src = Path(source_dir).expanduser().resolve()
dst = Path(ga_root).resolve()

# 备份
backup_dir = dst / "temp" / f"memory_import_backup_{timestamp}"
shutil.copytree(dst / "memory", backup_dir / "memory")

# 复制
shutil.copytree(src / "memory", dst / "memory", dirs_exist_ok=True)

return {
"memoryCopied": count,
"backupDir": str(backup_dir)
}

插件与扩展

反思模式

reflect/ 目录下的脚本允许代理定期执行检查任务:

1
2
3
4
5
6
7
8
9
10
11
12
# reflect/example.py
INTERVAL = 60 # 每 60 秒执行一次
ONCE = False # 是否只执行一次

def check():
"""检查函数"""
result = "检查结果"
return result

def on_done(result):
"""完成回调"""
print(f"反思完成:{result}")

启动反思模式:

1
python agentmain.py --reflect example.py

完整启动流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
1. 用户启动前端(如 python frontends/tui_v3.py)

2. 前端初始化 GenericAgent 实例

3. GenericAgent 加载配置(mykey.py)

4. 创建 LLM 客户端(ToolClient)

5. 加载工具模式(tools_schema.json)

6. 创建 GenericAgentHandler

7. 启动 agent_runner_loop 线程

8. 前端进入消息循环

9. 用户输入消息

10. 消息进入 task_queue

11. agent_runner_loop 获取任务

12. 构建消息列表

13. 调用 LLM

14. 处理响应(文本/工具调用)

15. 如果是工具调用,执行工具并继续循环

16. 如果是文本响应,输出到 display_queue

17. 前端显示响应

18. 等待下一个任务

关键设计模式

1. 生产者-消费者模式

任务队列和显示队列使用经典的生产者-消费者模式:

1
2
3
4
5
6
7
8
9
10
11
# 生产者(前端)
task_queue.put(task)

# 消费者(agent_runner_loop)
task = task_queue.get()

# 生产者(agent_runner_loop)
display_queue.put(response)

# 消费者(前端)
response = display_queue.get()

2. 策略模式

LLM 调用使用策略模式,不同的 LLM 实现相同的接口:

1
2
3
4
5
6
7
8
9
10
11
class ToolClient:
def __init__(self, backend: BaseSession):
self.backend = backend # 策略

def chat(self, messages, tools):
return self.backend.chat(messages, tools)

# 不同的策略
claude_client = ToolClient(NativeClaudeSession())
gpt_client = ToolClient(NativeOAISession())
gemini_client = ToolClient(NativeGeminiSession())

3. 观察者模式

前端通过队列监听代理的消息:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 代理(被观察者)
class GenericAgent:
def __init__(self):
self.display_queue = Queue()

def notify(self, message):
self.display_queue.put(message)

# 前端(观察者)
class TUI:
def __init__(self, agent):
self.agent = agent

def start_listener(self):
while True:
msg = self.agent.display_queue.get()
self.update_ui(msg)

4. 工厂模式

LLM 客户端使用工厂模式创建:

1
2
3
4
5
6
7
8
9
10
def resolve_client(config):
"""根据配置创建对应的 LLM 客户端"""
if config.type == "claude":
return NativeToolClient(NativeClaudeSession(config))
elif config.type == "openai":
return NativeToolClient(NativeOAISession(config))
elif config.type == "gemini":
return NativeToolClient(NativeGeminiSession(config))
else:
raise ValueError(f"不支持的 LLM 类型:{config.type}")

错误处理机制

1
2
3
4
5
6
7
8
9
10
11
def execute_tool_safe(self, tool_name, tool_args):
"""安全执行工具(带错误处理)"""
try:
result = self.execute_tool(tool_name, tool_args)
return result
except Exception as e:
# 记录错误
log_error(f"工具执行失败:{tool_name}: {e}")

# 返回错误信息给 LLM
return f"❌ 工具执行失败:{str(e)}"

中断与恢复

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class GenericAgentHandler:
def __init__(self):
self.aborted = False

def abort(self):
"""中断当前任务"""
self.aborted = True

def execute_tool(self, tool_name, tool_args):
"""执行工具(检查中断标志)"""
if self.aborted:
raise InterruptedError("任务已中断")

# 执行工具...

def reset(self):
"""重置中断标志"""
self.aborted = False

配置示例

mykey.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# LLM 配置
LLM_LIST = [
{
"type": "claude",
"model": "claude-3-5-sonnet-20241022",
"api_key": "your-api-key",
"base_url": "https://api.anthropic.com"
},
{
"type": "openai",
"model": "gpt-4-turbo",
"api_key": "your-api-key",
"base_url": "https://api.openai.com/v1"
}
]

# 默认 LLM 索引
DEFAULT_LLM = 0

# 其他配置
SYSTEM_PROMPT = "你是一个有用的助手..."
MAX_HISTORY = 50 # 最大历史记录条数

总结

GenericAgent 的核心设计理念是极简主义自我进化

  1. 极简架构:核心代码仅约 3000 行,Agent Loop 仅约 100 行
  2. 原子工具:9 个基础工具覆盖所有系统级操作
  3. 多模型支持:统一的 LLM 抽象层,支持主流大模型
  4. 多前端:TUI、GUI、Web、聊天机器人等多种界面
  5. 记忆系统:多层记忆,支持知识积累
  6. 自我进化:自动将任务结晶为可复用的技能

与其他框架的对比

维度 GenericAgent Codex CLI Hermes Agent Pi Agent
核心代码量 ~3000 行 ~数万行 ~数万行 ~数万行
语言 Python Rust Python TypeScript
Agent Loop ~100 行 复杂状态机 复杂状态机 复杂状态机
工具数量 9 个原子工具 内置+MCP 丰富工具集 内置+扩展
前端 10+ 种 TUI/CLI CLI/TUI/Web TUI/RPC
记忆系统 多层记忆 会话存储 17 种插件 JSONL 会话
设计理念 极简进化 工程安全 全栈平台 模块解耦

GenericAgent 适合谁?

  • 想要快速理解 AI Agent 核心原理的学习者
  • 想要构建自己 Agent 原型的开发者
  • 需要多前端接入(钉钉、Discord、微信等)的场景
  • 追求极简设计的工程师

如果你想快速理解 AI Agent 的核心原理,或者想要一个可以随意魔改的极简 Agent 框架,GenericAgent 是绝佳的学习素材。


项目地址:GenericAgent(开源项目)
核心文件:ga.py、agent_loop.py、agentmain.py、llmcore.py


GenericAgent 源码解析:3000 行代码构建的极简自主代理
https://tingfeng347.github.io/2026/08/02/GenericAgent 源码解析:3000 行代码构建的极简自主代理/
作者
Tingfeng
发布于
2026年8月2日
许可协议