🏷️ Label 说明（仅标注后端）：
【未实现】：文档中有能力描述，但当前后端代码未落地【部分实现】：已有基础能力，但未形成完整闭环【未接入主链路】：模块已实现，但当前 API 主调用路径未使用
| 特性 | 描述 |
|---|---|
| 智能压缩算法 | 当上下文使用率达到92%阈值时自动触发压缩，保留关键信息 |
| 重要性评分 | 基于消息角色、位置、关键词等多维度评分，智能筛选保留内容 |
| 分层存储 | Hot(活跃) → Warm(近期) → Cold(归档) 三层存储机制 |
| Token优化 | 动态上下文窗口调整，最大化利用模型上下文能力 【部分实现】（当前为固定上限 + 压缩触发，未实现按模型能力动态扩缩容） |
| 历史摘要 | 对压缩的历史对话生成智能摘要，保留上下文连贯性 |
| 特性 | 描述 |
|---|---|
| 异步核心调度器 | 基于asyncio的异步架构，支持并发处理多个会话 【未接入主链路】（当前 /chat 与 /chat/stream 直接走 ChatAgent 调用链） |
| 中断和恢复 | 支持中断正在进行的对话，可从检查点恢复执行 【部分实现】（已有中断标记与检查点写入，未实现检查点恢复执行） |
| 检查点机制 | 定期保存执行状态，异常时可恢复 【部分实现】（已保存检查点，未提供恢复 API / 自动恢复流程） |
| 多层异常处理 | 完善的错误捕获和恢复机制，保证系统稳定性 |
| 工具调用 | 支持并行工具执行，可扩展自定义工具 |
作为tools编排在工具里
| 特性 | 描述 |
|---|---|
| LLM 自主决策开启 | 通过 system prompt 注入 + manage_todo_list 工具，LLM 自行判断何时启用多步骤任务追踪 |
| 三态步骤管理 | 每个步骤支持 pending → running → completed 三态流转，同一时间最多一个 running |
| SSE 实时推送 | 每次 todo 变更自动通过 SSE 推送完整快照（type="todo_list"），前端无需轮询 |
| PostgreSQL 持久化 | 独立两张表（session_todo_lists / session_todo_items）+ 乐观锁 revision，支持页面刷新恢复 |
| REST 恢复端点 | GET /sessions/{id}/todo-list 供前端刷新时一次性拉取最新 todo 状态 |
| 与普通会话兼容 | 无 todo 的会话不受影响，todo 为可选附加能力，不改变原有对话链路 |
| 特性 | 描述 |
|---|---|
| 优先级队列 | 支持消息优先级调度，紧急消息优先处理 |
| 多后端支持 | Memory(内存) / Redis / Kafka 三种后端可选 |
| 中间件系统 | 内置日志、计时、重试、限流等中间件，可自定义扩展 |
| 消息TTL | 支持消息过期时间设置 【未实现】（配置项已存在，消费路径未执行 TTL 过期裁剪） |
| 特性 | 描述 |
|---|---|
| 思考模式展示 | AI思考过程置灰显示，思考结束后可折叠展开 |
| 流式响应 | 实时显示AI回复，打字机效果 |
| 对话管理 | 独立会话ID管理，支持多会话切换 |
| 自动标题 | 对话开始时自动生成主题标题 |
| Markdown渲染 | 支持代码高亮、表格、列表等富文本展示 |
| 响应式设计 | 适配桌面和移动端 |
| 特性 | 描述 |
|---|---|
| Docker 容器隔离 | 每次代码执行在独立 Docker 容器中运行，提供 OS 级安全隔离 |
| 资源限制 | CPU（50% 单核）、内存（256 MB）、PID（64）、执行时间（30s）严格限制 |
| 预装科学计算库 | 镜像内置 numpy / pandas / matplotlib / sympy / scipy / requests |
| AST 安全预检 | 执行前静态分析，拦截 fork bomb 等资源耗尽模式，预警高风险操作 |
| 网络隔离 | 默认禁用容器网络，可按需开启 |
| 输出捕获与截断 | 完整捕获 stdout / stderr，超过 64 KB 自动截断 |
| 即用即销 | 每次执行创建一次性容器，执行完毕立即销毁，无状态残留 |
┌─────────────────────────────────────────────────────────────────┐
│ Vue3 Frontend │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ChatWindow│ │ChatMessage│ │ChatInput │ │ChatSidebar│ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ └──────────────┼──────────────┼──────────────┘ │
│ ▼ ▼ │
│ ┌──────────────────────────┐ │
│ │ Pinia Store (State) │ │
│ └────────────┬─────────────┘ │
└───────────────────────────┼─────────────────────────────────────┘
│ HTTP/SSE
▼
┌─────────────────────────────────────────────────────────────────┐
│ FastAPI Backend │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ API Layer │ │
│ │ /chat/ /chat/stream /sessions/ /chat/title │ │
│ │ /sessions/{id}/todo-list │ │
│ └────────────────────────────┬─────────────────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ ChatAgent Core │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Agent Loop │ │Tool Executor│ │Memory Manager│ │ │
│ │ │(Async Sched)│ │ (Parallel) │ │(Compress 92%)│ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ └────────────────┼────────────────┘ │ │
│ └──────────────────────────┼───────────────────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Message Pipeline & Queue │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │Middleware│→ │Priority Q│→ │ Handler │ │ │
│ │ │(Log/Retry)│ │(Redis/Kafka)│ │ │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼───────────────────────────────┐ │
│ │ Code Execution Sandbox (Docker) │ │
│ │ ┌──────────┐ ┌───────────────┐ ┌──────────────────┐ │ │
│ │ │ Security │→ │ Container │→ │ Output Capture │ │ │
│ │ │(AST Check)│ │(CPU/Mem/PID) │ │(stdout/stderr) │ │ │
│ │ └──────────┘ └───────────────┘ └──────────────────┘ │ │
│ └──────────────────────────────────────────────────────────┘ │
└───────────────────────────┬─────────────────────────────────────┘
│ OpenAI API
▼
┌─────────────────┐
│ OpenAI / │
│ Compatible API│
└─────────────────┘
0. 环境要求
- 后端: Python 3.12+
- 前端: Node.js 18+ / Bun
- 代码沙箱: Docker Engine 20.10+
- 可选: Redis 7+ (用于消息队列), Kafka (用于大规模部署)
1. 后端启动
# 进入后端目录
cd agent-backend
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境
source venv/bin/activate # Linux/Mac
# 或
venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt
# 配置环境变量
cp .env.example .env
# 编辑 .env 文件，填入你的配置
# OPENAI_API_KEY=your-api-key-here
# OPENAI_BASE_URL=https://api.openai.com/v1
# 启动开发服务器
uvicorn app.main:app --reload --port 8000
# 或生产模式
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 42. 前端启动
# 进入前端目录
cd agent-frontend
# 安装依赖
npm install
# 或使用 pnpm
pnpm install
# 或使用 bun
bun install
# 启动开发服务器
npm run dev
# 构建生产版本
npm run build3. 访问应用
- 前端界面: http://localhost:5173
- API文档 (Swagger): http://localhost:8000/docs
- API文档 (ReDoc): http://localhost:8000/redoc
当上下文使用率达到92%阈值时自动触发压缩：
# 压缩触发条件
def should_compress(self, messages: list[Message], max_tokens: int) -> bool:
total_tokens = sum(m.token_count for m in messages)
usage_ratio = total_tokens / max_tokens
return usage_ratio >= 0.92 # 92% 阈值基于多维度评分决定消息保留优先级：
def score(self, message: Message, position: int, total: int) -> float:
# 位置因子 (最近的消息更重要)
position_factor = self.decay_factor ** (total - position - 1)
# 角色权重 (system > user > assistant > tool)
role_weights = {
MessageRole.SYSTEM: 1.0,
MessageRole.USER: 0.8,
MessageRole.ASSISTANT: 0.6,
MessageRole.TOOL: 0.5,
}
# 关键词分析
keyword_score = self._analyze_keywords(message.content)
# 工具调用加成
tool_bonus = 0.2 if message.tool_calls else 0.0
return weighted_sum(...)┌─────────────────────────────────────────────┐
│ Hot Layer │
│ - 当前活跃对话 │
│ - 完整消息内容 │
│ - 最高优先级 │
├─────────────────────────────────────────────┤
│ Warm Layer │
│ - 近期历史对话 │
│ - 可快速恢复 │
│ - 中等优先级 │
├─────────────────────────────────────────────┤
│ Cold Layer │
│ - 归档历史 │
│ - 仅保留摘要 │
│ - 低优先级 │
└─────────────────────────────────────────────┘
class AgentLoop:
async def _main_loop(self) -> None:
"""主处理循环"""
while not self._stop_event.is_set():
# 检查暂停
if self._pause_event.is_set():
await asyncio.sleep(0.1)
continue
# 获取下一条消息
message = await self.queue.dequeue(timeout=1.0)
if message:
# 异步处理
task = asyncio.create_task(
self._process_message(message)
)
self._tasks[message.session_id] = task# 中断会话
async def interrupt(self, session_id: UUID) -> bool:
if session_id in self._interrupt_events:
self._interrupt_events[session_id].set()
self._agent_states[session_id].status = MessageStatus.INTERRUPTED
return True
return False
# 检查点恢复
async def _create_checkpoint(self, session_id: UUID, state: AgentState):
checkpoint = Checkpoint(
id=state.session_id,
iteration=state.iteration,
state={"status": state.status.value},
messages=list(session.messages)
)
self._checkpoints[session_id].append(checkpoint)🏷️
【部分实现】当前后端已实现“中断标记 + 检查点写入”，但尚未实现“从检查点恢复继续执行”的完整流程（包括恢复入口、状态回放、重入执行）。
# 内置中间件
pipeline = (
MessagePipeline()
.use(LoggingMiddleware()) # 日志记录
.use(TimingMiddleware()) # 性能计时
.use(ValidationMiddleware()) # 数据验证
.use(RetryMiddleware(max_retries=3)) # 重试机制
.use(RateLimitMiddleware(rps=10.0)) # 限流
)🏷️
【未实现】QUEUE_MESSAGE_TTL已在配置中定义，但当前队列enqueue/dequeue主路径未依据 TTL 进行过期过滤。
框架启动后，工具注册链路如下：
ChatAgent.__init__调用create_default_executor()app.agent.executor.create_default_executor()转调app.agent.tools.register.create_default_executor()register.create_default_executor()创建ToolExecutor，并执行register_default_tools(executor)register_default_tools()读取settings.tools.*，调用register_discovered_tools(...)register_discovered_tools()内部调用discover_tools(...)完成发现，再逐个executor.register_tool(tool)
这意味着：工具发现发生在 Agent 初始化阶段，不是每次请求动态发现。
discover_tools(...) 同时支持两类来源：
-
Built-in：遍历
app.agent.tools.internal包- 使用
pkgutil.walk_packages递归扫描模块 - 使用
inspect.getmembers(..., inspect.isclass)收集类 - 仅保留“定义在当前模块中 +
BaseTool子类 + 非抽象类” - 通过无参构造实例化
- 使用
-
Entry Points（SPI）：读取
importlib.metadata.entry_points(group=...)- 对每个 entry point 调用
ep.load() - 允许以下返回形态（会被统一转换）：
BaseTool子类BaseTool实例- 工厂函数（返回上述对象）
- 可迭代对象（元素可递归为上述任意形态）
- 对每个 entry point 调用
- 重名工具（同
tool.name）处理：先到先得，后续同名工具会被忽略并记录 warning 日志 - 发现异常处理：
TOOLS_DISCOVERY_FAIL_FAST=false：记录 error 日志并跳过失败提供方TOOLS_DISCOVERY_FAIL_FAST=true：抛出异常，启动失败（适合强一致生产环境）
TOOLS_ENABLE_BUILTIN_DISCOVERY=true # 是否扫描内置工具包
TOOLS_ENABLE_ENTRYPOINT_DISCOVERY=true # 是否启用 entry points 发现
TOOLS_ENTRYPOINT_GROUP=chat_agent_framework.tools
TOOLS_DISCOVERY_FAIL_FAST=false # 单个 provider 失败时是否直接失败- entry point 推荐指向无状态工具类或轻量工厂函数
- 工具
name必须全局唯一（建议使用前缀，如weather.query） execute(**kwargs)中避免长阻塞；需要网络 I/O 时必须异步化- 保证
parameters与实际函数参数语义一致，避免模型构造非法参数
当工具未生效时，优先检查：
- 插件包是否安装到当前运行环境
TOOLS_ENTRYPOINT_GROUP是否与插件声明一致- 工具名是否与现有工具冲突（冲突时后加载者会被忽略）
TOOLS_DISCOVERY_FAIL_FAST是否导致启动直接失败- 启动日志中是否出现
Tool discovery completed/External tool entry point failed
代码执行沙箱允许 Agent（LLM）在对话过程中自主编写并运行 Python 代码，获取真实计算结果。这与 ChatGPT Code Interpreter / Claude Analysis 的能力对齐。
| 方案 | 隔离级别 | 安全性 | 包管理 | 生产就绪 | 代表产品 |
|---|---|---|---|---|---|
eval() / exec() |
无 | ❌ 极危险 | ❌ | ❌ | — |
| RestrictedPython | 函数级 | ❌ | ❌ | — | |
| subprocess + venv | 进程级 | ✅ | — | ||
| Docker 容器 | OS 级 | ✅ 工业级 | ✅ | ✅ | ChatGPT / Claude / Gemini |
| Firecracker microVM | VM 级 | ✅ 最强 | ✅ | ✅ | AWS Lambda |
Docker 容器是 Agent 代码执行的行业标准。ChatGPT Code Interpreter、Claude Analysis、Gemini Code Execution 底层均使用容器化方案。核心优势：
- 命名空间隔离：进程、文件系统、网络、IPC 完全隔离，即使代码尝试
os.system("rm -rf /")也只影响容器内部 - cgroup 资源限制：CPU、内存、PID 数量均可精确控制，防止资源耗尽攻击
- 即用即销：每次执行创建全新容器，执行后立即销毁，无状态泄漏
- 可复现环境：预构建镜像确保每次执行环境一致
- 生态成熟：Docker SDK、镜像管理、日志收集等工具链完善
app/sandbox/ sandbox/
├── __init__.py ├── Dockerfile # 沙箱镜像定义
├── models.py # 数据模型 └── requirements.txt # 预装包列表
├── security.py # AST 安全预检
├── manager.py # Docker 容器管理
└── executor.py # 高层执行接口
app/agent/tools/internal/
└── python_executor.py # BaseTool 实现（自动被 SPI 发现注册）
用户: "帮我计算 fibonacci(50) 的值"
│
▼
LLM 决定调用 python_executor 工具
│ {"code": "def fib(n):\n a, b = 0, 1\n for _ in range(n):\n a, b = b, a+b\n return a\nprint(fib(50))"}
│
▼
PythonExecutorTool.execute(code)
│
├─ ① SecurityChecker.validate(code)
│ ├─ AST 解析 → 语法检查
│ ├─ 模块黑名单检查（ctypes / multiprocessing / signal）
│ ├─ 危险调用预警（os.system 等 → 仅 warn，不 block）
│ └─ 通过 ✅ / 拒绝 ❌
│
├─ ② DockerSandboxManager.execute(request)
│ ├─ 创建一次性容器（image: agent-sandbox:latest）
│ │ CPU: 50% 单核 | 内存: 256MB | PID: 64 | 网络: 禁用
│ ├─ 通过 tar archive 注入代码 → /workspace/main.py
│ ├─ 启动容器, python -u /workspace/main.py
│ ├─ asyncio.wait_for(container.wait(), timeout=30s)
│ ├─ 捕获 stdout / stderr (≤ 64KB)
│ └─ force remove 容器
│
▼
ExecutionResult → 格式化返回给 LLM
│ "STDOUT:\n12586269025\n\nExecution time: 0.03s"
│
▼
LLM 组织自然语言回复给用户
┌─────────────────────────────────────────────────────────┐
│ Layer 1: AST 静态分析（SecurityChecker） │
│ • 语法检查（快速失败，避免浪费容器资源） │
│ • 阻止：ctypes / multiprocessing / signal │
│ • 预警：os.system / subprocess / eval（仅日志记录） │
├─────────────────────────────────────────────────────────┤
│ Layer 2: Docker 容器隔离 │
│ • 独立 PID / Mount / Network / IPC namespace │
│ • 非 root 用户 (sandbox) │
│ • security_opt: no-new-privileges │
├─────────────────────────────────────────────────────────┤
│ Layer 3: cgroup 资源限制 │
│ • CPU: 50% 单核 (cpu_quota / cpu_period) │
│ • 内存: 256 MB (OOM Kill) │
│ • PID: 64 (防止 fork bomb) │
│ • 执行时间: 30s (asyncio 超时 → container.kill) │
├─────────────────────────────────────────────────────────┤
│ Layer 4: 网络隔离 │
│ • 默认 network_disabled=true │
│ • 可按请求临时开启（需要 pip install 时） │
├─────────────────────────────────────────────────────────┤
│ Layer 5: 输出控制 │
│ • stdout / stderr 截断上限: 64 KB │
│ • 容器文件系统执行后销毁，无持久化 │
└─────────────────────────────────────────────────────────┘
create start wait/timeout logs remove
│ │ │ │ │
┌─────────┐ ┌─▼───────────┐ ┌─▼──────────┐ ┌─▼────────┐ ┌▼────────┐ ┌▼──────────┐
│ Request │──▶│ Container │▶│ Running │▶│ Exited / │▶│ Output │▶│ Destroyed │
│(code) │ │ Created │ │ (python │ │ Timeout │ │ Capture │ │ (force rm)│
└─────────┘ │ (detach) │ │ main.py) │ │ (killed) │ └─────────┘ └───────────┘
└─────────────┘ └────────────┘ └──────────┘
tar inject ↗
(put_archive)
代码注入方式：不使用 Volume Mount（有安全风险），而是通过 container.put_archive() 将代码打包为 tar 写入容器内 /workspace/main.py，实现零宿主机文件系统暴露。
沙箱镜像定义在 sandbox/Dockerfile：
FROM python:3.12-slim
# 非 root 用户
RUN groupadd -r sandbox && useradd -r -g sandbox -m -s /bin/bash sandbox
# 预装科学计算包
COPY requirements.txt /tmp/requirements.txt
RUN pip install --no-cache-dir -r /tmp/requirements.txt
# 工作目录
RUN mkdir -p /workspace && chown sandbox:sandbox /workspace
WORKDIR /workspace
USER sandbox
CMD ["python", "-u", "/workspace/main.py"]预装包列表（sandbox/requirements.txt）：
- 计算: numpy, pandas, scipy, sympy
- 可视化: matplotlib
- 工具: requests, python-dateutil, tabulate, pyyaml
| 场景 | 处理方式 |
|---|---|
| 使用预装包 (numpy 等) | 直接 import，零延迟 |
| 需要额外包 | LLM 传入 install_packages 参数 → 容器内 pip install -q → 需开启网络 |
| 安装失败 | stderr 返回错误信息，LLM 可自行调整 |
1. 异步 Docker 调用
Docker SDK 是同步的，通过 asyncio.to_thread() 包装保持事件循环响应：
# manager.py — 所有 Docker 操作均 offload 到线程池
container = await asyncio.to_thread(self._create_container, request)
await asyncio.to_thread(self._copy_code_to_container, container, script)
await asyncio.to_thread(container.start)
exit_info = await asyncio.wait_for(
asyncio.to_thread(container.wait),
timeout=request.timeout, # asyncio 层超时 → container.kill()
)2. 超时处理双保险
asyncio.wait_for(timeout=30s) ← 应用层超时（首选）
│ TimeoutError
▼
container.kill() ← 强制终止容器进程
container.remove(force=True) ← 清理容器（finally 块保证执行）
3. SPI 自动注册
PythonExecutorTool 放置在 app/agent/tools/internal/ 包中，框架启动时被 discover_tools() 自动扫描注册，无需手动配置。
| 集成点 | 说明 |
|---|---|
| Tool SPI | PythonExecutorTool 继承 BaseTool，自动被发现注册 |
| ToolExecutor | 通过标准 execute() 调用链执行，享受并行执行 / 超时 / 错误恢复 |
| Config | SandboxConfig 挂载到 Settings.sandbox，支持环境变量覆盖 |
| System Prompt | 可在 system prompt 中告知 LLM 拥有代码执行能力 |
| SSE 流式 | 工具执行结果作为 tool 消息回传，LLM 基于结果继续生成 |
# 1. 确保 Docker 已安装并运行
docker --version
docker info
# 2. 首次启动时自动构建沙箱镜像（约 1-2 分钟）
# 或手动构建：
docker build -t agent-sandbox:latest ./sandbox/
# 3. 安装 Python Docker SDK
pip install docker>=7.0.0如果宿主机没有安装 Docker，
CodeExecutor.initialize()会抛出明确错误并记录日志，其他工具不受影响。
Todo 功能允许 LLM 在处理多步骤任务时，自动创建并维护一份可视化的任务清单，前端可实时展示步骤进度。
Todo 功能不是由用户手动开关，而是由 LLM 自主判断是否需要：
- System Prompt 注入：
ChatAgent在每次对话时，将TODO_SYSTEM_PROMPT追加到系统消息中，告知 LLM 拥有manage_todo_list工具。 - LLM 自主调用：当 LLM 判断当前任务需要多步骤完成（如多阶段分析、批量操作等），它会自行发起
manage_todo_list工具调用。 - 工具拦截：
ChatAgent在处理工具调用时，将manage_todo_list从普通工具中分离，单独路由到TodoService，不经过ToolExecutor。
# core.py 中的 system prompt 片段
TODO_SYSTEM_PROMPT = """你拥有一个名为 manage_todo_list 的工具。
当用户的请求需要多步骤才能完成时（例如：多阶段分析、多文件操作、复杂计划执行等），
你**必须**在开始工作前先调用 manage_todo_list 来创建任务清单。
使用规则：
1. 在开始多步骤任务前，调用 manage_todo_list 创建完整的步骤清单（所有步骤 pending，第一步设为 running）。
2. 每完成一个步骤后，再次调用 manage_todo_list，将已完成的步骤标记为 completed，下一步标记为 running。
3. 每次调用都必须发送**完整列表**（不是增量更新）。
4. 同一时间最多只有一个步骤处于 running 状态。
5. 对于简单的单步任务（如简单问答、翻译等），**不要**调用此工具。
"""触发时机示意图：
用户发送消息
│
▼
ChatAgent.chat_stream()
│
├─ 注入 TODO_SYSTEM_PROMPT 到 system 消息
├─ 注册 manage_todo_list 到工具列表
│
▼
LLM 返回响应
│
├─ 若包含 manage_todo_list 调用 ──→ 拦截，路由到 TodoService
│ │
│ ├─ 写入 / 更新数据库
│ ├─ SSE 推送 todo 快照
│ └─ 返回确认给 LLM
│
└─ 其他工具调用 ──→ 正常走 ToolExecutor
| 维度 | 普通会话 | 含 Todos 的会话 |
|---|---|---|
| 数据库 | 仅 sessions + messages 表 |
额外关联 session_todo_lists + session_todo_items 表 |
| SSE 事件 | session / thinking / content / done |
额外推送 todo_list 类型事件 |
| 工具调用 | 所有工具走 ToolExecutor |
manage_todo_list 被拦截走 TodoService，其余不变 |
| System Prompt | 原有 system 消息 | 额外追加 TODO_SYSTEM_PROMPT 段落 |
| REST 端点 | 无 todo 相关 | GET /sessions/{id}/todo-list 返回快照 |
| SessionModel 关系 | todo_list 为 None |
todo_list 指向 TodoListModel 实例 |
对于前端：
- 普通会话：
StreamChunk中todo_list字段始终为null，前端不显示 todo 面板 - 含 Todos 会话：收到
type="todo_list"事件时渲染任务列表；页面刷新后通过 REST 恢复
Todo 数据存储在两张 PostgreSQL 表中，与 sessions 表通过外键关联：
sessions (1) ──── (0..1) session_todo_lists (1) ──── (N) session_todo_items
session_todo_lists 表：
| 列 | 类型 | 说明 |
|---|---|---|
id |
UUID PK | 主键 |
session_id |
UUID FK UNIQUE | 关联 sessions，一对一 |
title |
VARCHAR(255) | 任务清单标题 |
revision |
INTEGER | 乐观锁版本号，每次写操作 +1 |
status |
VARCHAR(50) | active / completed / archived |
created_at / updated_at |
TIMESTAMPTZ | 时间戳 |
session_todo_items 表：
| 列 | 类型 | 说明 |
|---|---|---|
id |
UUID PK | 主键 |
todo_list_id |
UUID FK | 关联 session_todo_lists |
label |
VARCHAR(500) | 步骤描述 |
status |
VARCHAR(50) | pending / running / completed |
order_index |
INTEGER | 排序序号 |
created_at / updated_at |
TIMESTAMPTZ | 时间戳 |
关键约束：
session_id上建立 UNIQUE 索引，保证每个会话最多一个 todo-list- 外键使用
ON DELETE CASCADE，删除会话时自动级联清除 todo 数据 revision用于并发控制：前端可用此字段判断是否需要更新渲染
每次 todo 变更，后端通过 SSE 推送完整 todo 快照（非增量）：
data: {
"session_id": "550e8400-...",
"type": "todo_list",
"delta": "",
"todo_list": {
"id": "a1b2c3d4-...",
"title": "数据分析流程",
"items": [
{"id": "...", "label": "收集数据源", "status": "completed", "order_index": 1},
{"id": "...", "label": "数据清洗", "status": "running", "order_index": 2},
{"id": "...", "label": "建模评估", "status": "pending", "order_index": 3}
],
"revision": 3,
"updated_at": "2026-02-12T10:30:00Z"
}
}ChatAgent (core.py)
│ 拦截 manage_todo_list 工具调用
▼
TodoService (services/todo_service.py)
│ 业务逻辑 + SSE 广播
▼
TodoRepository (database/todo_repository.py)
│ CRUD + revision 管理
▼
PostgreSQL (session_todo_lists + session_todo_items)
TodoService 关键方法：
| 方法 | 说明 |
|---|---|
create_todo_list(session_id, title, labels) |
创建清单，首项自动 running |
create_or_replace_with_items(session_id, title, items) |
用 LLM 给出的精确状态创建/替换清单 |
advance_step(session_id) |
running→completed，下一个 pending→running |
set_item_status(session_id, item_id, status) |
设置指定项状态 |
complete_all(session_id) |
标记所有项为 completed |
clear(session_id) |
删除整个 todo-list |
get_todo_list(session_id) |
只读查询（无广播） |
POST /api/v1/chat/
Content-Type: application/json
{
"message": "你好，请介绍一下自己",
"session_id": "uuid-or-null-for-new-session"
}响应:
{
"session_id": "550e8400-e29b-41d4-a716-446655440000",
"message": {
"id": "...",
"role": "assistant",
"content": "你好！我是Chat Agent...",
"created_at": "2024-01-01T00:00:00Z"
},
"status": "completed",
"usage": {
"prompt_tokens": 20,
"completion_tokens": 100,
"total_tokens": 120
}
}POST /api/v1/chat/stream
Content-Type: application/json
{
"message": "写一个Python函数",
"session_id": "uuid-or-null"
}SSE 响应格式:
data: {"session_id":"...","type":"session","delta":"uuid"}
data: {"session_id":"...","type":"thinking","thinking":"让我思考一下..."}
data: {"session_id":"...","type":"content","delta":"好的"}
data: {"session_id":"...","type":"content","delta":"，我来"}
data: {"session_id":"...","type":"done","is_thinking_complete":true}
POST /api/v1/chat/title
Content-Type: application/json
{
"session_id": "550e8400-e29b-41d4-a716-446655440000"
}响应:
{
"session_id": "550e8400-e29b-41d4-a716-446655440000",
"title": "Python函数编写讨论"
}GET /api/v1/sessions/?page=1&page_size=20GET /api/v1/sessions/{session_id}DELETE /api/v1/sessions/{session_id}GET /api/v1/sessions/{session_id}/todo-list响应 (200):
{
"id": "a1b2c3d4-...",
"title": "数据分析流程",
"items": [
{"id": "...", "label": "收集数据源", "status": "completed", "order_index": 1},
{"id": "...", "label": "数据清洗", "status": "running", "order_index": 2},
{"id": "...", "label": "建模评估", "status": "pending", "order_index": 3}
],
"revision": 3,
"updated_at": "2026-02-12T10:30:00Z"
}其他响应码:
204 No Content— 该会话没有 todo-list404 Not Found— session_id 不存在
此端点用于前端页面刷新或切换会话时恢复 todo 状态。实时更新通过
/chat/streamSSE 的todo_list事件获取。
# ============ OpenAI 配置 ============
OPENAI_API_KEY=your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini
OPENAI_MAX_TOKENS=4096
OPENAI_TEMPERATURE=0.7
OPENAI_TIMEOUT=60.0
OPENAI_MAX_RETRIES=3
# ============ 内存配置 ============
MEMORY_MAX_CONTEXT_TOKENS=128000
MEMORY_COMPRESSION_THRESHOLD=0.92
MEMORY_TARGET_COMPRESSION_RATIO=0.3
MEMORY_MAX_MESSAGES_IN_MEMORY=100
MEMORY_SUMMARY_MAX_TOKENS=500
MEMORY_IMPORTANCE_DECAY_FACTOR=0.95
# ============ Agent 配置 ============
AGENT_MAX_ITERATIONS=10
AGENT_ITERATION_TIMEOUT=300
AGENT_ENABLE_PARALLEL_TOOLS=true
AGENT_MAX_PARALLEL_TOOLS=5
AGENT_ENABLE_INTERRUPTION=true
# ============ 消息队列配置 ============
QUEUE_BACKEND=memory
# Redis 配置 (如果使用 Redis)
QUEUE_REDIS_URL=redis://localhost:6379/0
# Kafka 配置 (如果使用 Kafka)
QUEUE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
QUEUE_KAFKA_TOPIC_PREFIX=agent
# 消息 TTL（当前后端消费链路未启用）
QUEUE_MESSAGE_TTL=3600 # `【未实现】`
# ============ 代码沙箱配置 ============
SANDBOX_ENABLED=true # 启用代码执行沙箱
SANDBOX_IMAGE_NAME=agent-sandbox:latest # Docker 镜像名
SANDBOX_AUTO_BUILD_IMAGE=true # 镜像不存在时自动构建
SANDBOX_EXECUTION_TIMEOUT=30 # 默认执行超时(秒)
SANDBOX_MAX_EXECUTION_TIMEOUT=120 # 最大允许超时(秒)
SANDBOX_MAX_OUTPUT_SIZE=65536 # 最大输出字节数 (64KB)
SANDBOX_MEMORY_LIMIT=256m # 容器内存限制
SANDBOX_CPU_PERIOD=100000 # CPU CFS 周期(微秒)
SANDBOX_CPU_QUOTA=50000 # CPU CFS 配额(50%单核)
SANDBOX_PIDS_LIMIT=64 # 容器最大进程数
SANDBOX_NETWORK_ENABLED=false # 默认禁用网络
SANDBOX_CONTAINER_WORKDIR=/workspace # 容器工作目录
# ============ 数据库配置 ============
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/agent_db
# ============ 服务器配置 ============
SERVER_HOST=0.0.0.0
SERVER_PORT=8000
SERVER_DEBUG=true
SERVER_CORS_ORIGINS=["http://localhost:5173","http://localhost:3000"]
# ============ 应用配置 ============
ENVIRONMENT=development创建 .env.local:
VITE_API_BASE_URL=http://localhost:8000/api/v1from app.agent.executor import BaseTool
class WeatherTool(BaseTool):
"""天气查询工具"""
@property
def name(self) -> str:
return "get_weather"
@property
def description(self) -> str:
return "获取指定城市的天气信息"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
async def execute(self, city: str, unit: str = "celsius") -> str:
# 实现天气查询逻辑
weather_data = await fetch_weather(city)
return f"{city}当前温度: {weather_data['temp']}°"
# 注册工具
from app.agent.core import ChatAgent
agent = ChatAgent()
agent.register_tool(WeatherTool())框架采用“服务发现”方式加载工具，完整实现细节已收敛到 5 核心模块详解 → 4) Tool SPI 注册与发现（核心实现），本节仅保留接入示例。
[project.entry-points."chat_agent_framework.tools"]
weather = "my_plugin.weather:WeatherTool"
batch_tools = "my_plugin.bundle:provide_tools"其中 provide_tools 可以返回：
- 单个
BaseTool子类/实例 - 或工具列表（可混合类与实例）
from app.messaging.pipeline import PipelineMiddleware, PipelineContext
from typing import Callable, Awaitable
class AuthMiddleware(PipelineMiddleware):
"""认证中间件"""
@property
def name(self) -> str:
return "auth"
async def process(
self,
context: PipelineContext,
next_handler: Callable[[PipelineContext], Awaitable]
):
# 验证 token
token = context.metadata.get("token")
if not self._validate_token(token):
raise UnauthorizedError("Invalid token")
# 继续处理
return await next_handler(context)
def _validate_token(self, token: str) -> bool:
# 实现验证逻辑
return True
# 使用中间件
pipeline.use(AuthMiddleware())from app.memory.compressor import CompressionStrategy
class CustomCompressor(CompressionStrategy):
"""自定义压缩策略"""
async def compress(
self,
messages: list[Message],
target_ratio: float
) -> tuple[list[Message], str | None]:
# 实现自定义压缩逻辑
retained = []
summary = None
for msg in messages:
if self._should_keep(msg):
retained.append(msg)
else:
summary = await self._summarize(msg)
return retained, summary| 技术 | 版本 | 用途 |
|---|---|---|
| Python | 3.12+ | 运行时 |
| FastAPI | 0.115+ | Web框架 |
| Pydantic | 2.10+ | 数据验证 |
| OpenAI SDK | 1.55+ | LLM调用 |
| Tiktoken | 0.8+ | Token计数 |
| Structlog | 24.4+ | 日志系统 |
| Redis | 5.2+ | 消息队列(可选) |
| Aiokafka | 0.12+ | Kafka支持(可选) |
| Docker SDK | 7.0+ | 代码沙箱容器管理 |
| 技术 | 版本 | 用途 |
|---|---|---|
| Vue | 3.5+ | UI框架 |
| TypeScript | 5.6+ | 类型系统 |
| Pinia | 2.2+ | 状态管理 |
| Vite | 5.4+ | 构建工具 |
| Tailwind CSS | 3.4+ | 样式框架 |
| Marked | 14.0+ | Markdown解析 |
| Axios | 1.7+ | HTTP客户端 |
MIT License
Made with ❤️ by Chat Agent Framework Team