lzb/agent-b1

0 Fork 0

T

lzb 84425f91e2 添加 .gitignore

2026-07-14 21:24:32 +08:00

agent

添加统一前端 Demo 服务器（端口8090）

2026-07-14 20:45:26 +08:00

.gitignore

添加 .gitignore

2026-07-14 21:24:32 +08:00

README.md

更新 README.md

2026-07-14 20:40:45 +08:00

README.md

个人模块 README

B1 模块：Agent 运行时与消息管理模块

1. 模块概述

1.1 模块名称

B1 — Agent 运行时与消息管理模块 (Agent Runtime & Message Manager)

1.2 模块说明

B1 是 Agent 系统的核心调度引擎，负责串联 B3（工具层）、B4（LLM 推理）、B5（记忆管理）三大外部模块，实现完整的 ReAct（Reasoning + Acting）Agent Loop。

本模块在完整系统中的定位是"总控中心"：

用户输入 → B1 Runtime
              ├─ B5 load_memory()      → 获取记忆上下文
              ├─ B3 get_tools_schema() → 获取工具描述
              │
              └─ Agent Loop:
                   ├─ B4 generate_ai_message()  → LLM 推理（生成 AIMessage）
                   ├─ 判断 AIMessage 是否有 tool_calls
                   │   ├─ 无 → 提取 final_answer，退出循环
                   │   └─ 有 → B3 execute_tool_calls() → 获取 ToolMessage
                   │            → 追加到 messages，回到 B4
                   │
                   └─ max_turns 超限 → 终止
              │
              └─ 输出产物 + B5 save_memory() → 保存对话记忆

核心职责：

接收并校验用户输入（runtime_input.json）
构建初始消息序列：SystemMessage + memory_context + HumanMessage
维护 Agent Loop：LLM 推理 → 工具调用 → 结果回填 → 再推理，直到获得最终回答
使用 max_turns 防止工具调用死循环
将完整消息序列、执行轨迹、最终回答写入产物文件
支持两种执行模式：fixture（独立演示，不依赖其他模块）和 integrated（全系统串联）

1.3 完成情况概览

类型	完成情况
基础要求	✅ 全部完成：接收单次用户输入、串联 B3/B4/B5、构建初始 messages、Agent Loop 循环控制、SystemMessage/AIMessage/ToolMessage 消息管理、输出 messages.json / trace.json / final_answer.md
进阶要求	✅ 5 项全部完成：多轮用户输入、断点续跑(Checkpoint/Resume)、批量任务处理(Batch Runner)、历史消息压缩(History Compression)、多 Prompt 模板切换与对比
可独立运行的演示	✅ Fixture 模式（完全不依赖 B2-B5 和 GPU）、Web Demo 系统
与团队系统集成情况	✅ 通过 `run_agent()` 函数被 `unified_demo_server.py`（统一前端，端口 8090）和 `run_full_demo.py`（一键完整演示）调用

2. 环境、模型与数据依赖

2.1 运行环境

项目	要求
Python 版本	`3.10`
必要依赖	`PyYAML`（读配置）、`sentence-transformers`（B5 向量检索，间接依赖）；Fixture 模式零额外依赖
是否需要模型	Fixture/Mock 模式不需要；Integrated + prompt_json 模式需要 Qwen3.5-4B
是否需要 GPU	Mock 模式不需要；真实模型推理建议 GPU
是否需要外部数据集	不需要

2.2 模型依赖

B1 本身不直接加载模型，而是通过 B4 层调用。当 B1 以 integrated 模式运行且 --llm_mode prompt_json 时，需要 B4 能加载模型。

模型	来源	项目内相对路径	用途
`Qwen3.5-4B`	`https://www.modelscope.cn/models/Qwen/Qwen3.5-4B`	由 `configs/model.yaml` 中 `model_name_or_path` 指定	B4 本地推理生成 AIMessage
Qwen3-1.7B	https://www.modelscope.cn/models/Qwen/Qwen3-1.7B	由 `configs/model.yaml` 中 `model_name_or_path` 指定	B4 本地推理生成 AIMessage

# 如无 GPU 或无模型，使用 mock 模式即可完整验证 B1 所有功能：
python b1_agent_runtime.py --input ../data/runtime_input.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode mock --outdir ../outputs/B1_runtime

2.3 数据集或样例数据依赖

B1 的数据依赖全部由项目内置样例覆盖，无需额外下载。

数据或文件	来源	项目内相对路径	用途
`runtime_input.json` + `runtime_input_0~5.json`	项目自带	`agent/data/`	不同任务类型的集成模式输入样例
`b1_fixture_input.json` + preset 系列	项目自带	`agent/data/b1_fixtures/`	Fixture 独立演示的预设输入
`batch_input_demo.json`	项目自带	`agent/data/`	批量任务输入样例
`prompts/*.txt`	项目自带	`agent/prompts/`	System Prompt 模板文件
`configs/tools.yaml`	项目自带（B3 配置）	`agent/configs/`	B3 工具配置
`configs/memory.yaml`	项目自带（B5 配置）	`agent/configs/`	B5 记忆配置
`configs/model.yaml`	项目自带（B4 配置）	`agent/configs/`	B4 模型配置

2.4 安装步骤

# 1. 创建 conda 环境
conda create -n b1_env python=3.10 -y
conda activate b1_env

# 2. 安装依赖
cd agent
export PYTHONNOUSERSITE=1
pip install -r requirements.txt

# 3. 验证：运行 fixture 模式（不需要 GPU，不需要 B2-B5）
cd code
python b1_agent_runtime.py \
  --input ../data/b1_fixtures/b1_fixture_input.json \
  --outdir ../outputs/B1_fixture

个人模块（B1）要求可以脱离完整系统独立运行：Fixture 模式不需要 B2-B5 也不需要 GPU，仅用预设 JSON mock 数据即可完成完整的 Agent 消息闭环验证。

3. 文件结构与接口边界

3.1 文件结构

只列出与本模块直接相关的文件。

agent/
├── code/
│   ├── b1_agent_runtime.py          # 核心：Agent 运行时主程序（728 行）
│   ├── b1_checkpoint.py             # 进阶2：断点续跑（保存/加载/检查 checkpoint）
│   ├── b1_history_compressor.py     # 进阶4：历史消息压缩（token 估算+摘要生成）
│   └── b1_batch_runner.py           # 进阶3：批量任务 CLI 入口
├── demo/
│   ├── b1_demo_server.py            # Web Demo 服务器（端口 8080）
│   └── templates/
│       └── b1_index.html            # Web Demo 前端页面
├── prompts/
│   ├── local_tool_agent.txt         # 通用工具代理 prompt
│   ├── file_expert_agent.txt        # 文件阅读专家 prompt
│   ├── calculation_agent.txt        # 计算分析专家 prompt
│   └── conversion_agent.txt         # 格式转换专家 prompt
├── data/
│   ├── runtime_input.json           # 集成模式输入样例（主任务）
│   ├── runtime_input_0~5.json       # 不同工具任务的输入样例
│   ├── b1_fixtures/                 # Fixture 模式预设数据
│   │   ├── b1_fixture_input.json
│   │   ├── preset_memory.json
│   │   ├── preset_ai_messages.json
│   │   └── preset_tool_messages.json
│   └── batch_input_demo.json        # 批量任务输入样例
├── experiments/
│   ├── compare_llm_modes.py         # mock vs prompt_json 对比实验
│   ├── compare_max_turns.py         # 不同 max_turns 对比实验
│   ├── compare_prompts.py           # 不同 system prompt 对比实验
│   ├── error_case_analysis.py       # 错误案例分析
│   ├── with_without_memory.py       # 有无 memory 对比实验
│   └── batch_experiment.py          # 端到端批量实验
├── outputs/B1_fixture/              # Fixture 模式输出
├── outputs/B1_runtime/              # Integrated 模式输出
├── B1_VERIFICATION.md               # 模块验证说明书
├── B1模块接口与集成关系.md           # B1 接口与集成关系文档
└── README.md                        # 项目总 README

3.2 接口边界

类型	来源 / 去向	数据格式	说明
输入	用户/前端 → `run_agent()`	`runtime_input.json`（JSON 对象）	含 `conversation_id`, `user_input`, `system_prompt_path`, `toolset`, `max_turns`, `save_memory` 等字段
输入	前端 → `run_agent()`	CLI 参数	`--tools_config`, `--memory_config`, `--model_config`, `--outdir`, `--llm_mode`
输入（批量）	用户 → `run_agent()`	JSON（`tasks` + `common_config`）	自动检测批量模式，内部循环调用
输出	B1 → 磁盘	`messages.json`	完整消息序列（System/Human/AI/Tool）
输出	B1 → 磁盘	`trace.json`	执行轨迹（状态、轮次、每轮详情、进阶功能记录）
输出	B1 → 磁盘	`final_answer.md`	最终回答 Markdown
输出	B1 → 磁盘	`runtime_log.jsonl`	运行时日志
调用 B5	`B1 → load_memory()`	函数调用	加载全局+指定 ID 的记忆文档
调用 B5	`B1 → save_memory()`	函数调用	保存对话记忆到磁盘
调用 B3	`B1 → get_tools_schema()`	函数调用	获取当前 toolset 的 OpenAI 风格 tools_schema
调用 B3	`B1 → execute_tool_calls()`	函数调用	传入 tool_calls 列表，获得 ToolMessage 列表
调用 B4	`B1 → generate_ai_message()`	函数调用	传入 messages + tools_schema，获得 AIMessage

4. 基础要求实现与演示

4.1 基础功能说明

依据 PPT 任务书中 B1 的基础实训要求，实现了以下 7 项基础功能：

接收单次用户输入：从 runtime_input.json 中读取 user_input 字段作为 HumanMessage
加载记忆上下文：调用 B5 load_memory() 获取全局记忆和用户指定的历史对话记忆
构建初始 messages：将 system_prompt + memory_context + user_input 构建为 [SystemMessage, HumanMessage]
获取工具 Schema：调用 B3 get_tools_schema() 获取可用工具描述，传给 B4 进行工具绑定
LLM 推理与判断：调用 B4 generate_ai_message()，判断返回的 AIMessage 是否包含 tool_calls
工具调用循环：有 tool_calls → 调用 B3 execute_tool_calls() → ToolMessage 追加到 messages → 再次调用 B4；支持 max_turns 防止死循环
消息角色管理：完整管理 SystemMessage、HumanMessage、AIMessage、ToolMessage 在整个 Agent 生命周期中的增删流转

4.2 基础功能实现路径

文件 / 函数 / 脚本	作用
`b1_agent_runtime.py:run_agent()` (第 324-668 行)	B1 主入口函数，编排整个 Agent 流程
`b1_agent_runtime.py:_validate_runtime_input()` (第 15-67 行)	校验 `runtime_input.json` 的字段完整性和合法性
`b1_agent_runtime.py:_memory_context()` (第 70-77 行)	将 B5 返回的记忆文档包装为 XML 标签格式
`b1_agent_runtime.py:generate_ai_message()` (第 85-88 行)	B4 调用代理，延迟导入避免 fixture 模式不必要加载
`b1_agent_runtime.py:_load_fixture_inputs()` (第 91-112 行)	Fixture 模式：加载预设 memory/tools_schema/AI/Tool 消息
`b1_agent_runtime.py:_fixture_tool_messages()` (第 115-127 行)	Fixture 模式：按 tool_call_id 匹配预设 ToolMessage
`b1_agent_runtime.py:_run_batch()` (第 222-321 行)	批量模式：内部分发每个 task 到 `run_agent()`

核心流程：

[读取 runtime_input.json] → [校验输入] → [加载 memory(B5)] → [生成 tools_schema(B3)]
→ [构建 system_prompt + memory_context + user_input → messages]
→ [Agent Loop] → [调用 B4] → [判断 AIMessage]
  → 有 tool_calls → [调用 B3 执行工具] → [ToolMessage 追加到 messages] → [回到 B4]
  → 无 tool_calls → [提取 content 作为 final_answer] → [退出循环]
→ [写入输出产物] → [保存 memory(B5)] → [返回结果]

4.3 基础功能输入格式与样例

字段 / 输入文件	类型 / 格式	是否必需	说明
`conversation_id`	`string`	是	对话唯一标识
`user_input`	`string`	是（与 `user_inputs` 二选一）	单次用户输入文本
`system_prompt_path`	`string`	是	System Prompt 模板文件路径（相对路径或绝对路径）
`toolset`	`string`	是	使用的工具集名称，对应 `tools.yaml` 中的 toolset
`max_turns`	`integer (>0)`	是	最大工具调用轮次，防止死循环
`save_memory`	`"none" \| "conversation" \| "global"`	是	是否保存本轮对话记忆
`selected_memory_ids`	`string[]`	否（integrated 模式默认 `[]`）	要加载的历史记忆 ID 列表
`use_global_memory`	`boolean`	否（默认 `false`）	是否同时加载全局记忆
`execution_mode`	`"integrated" \| "fixture"`	否（默认 `"integrated"`）	执行模式

样例输入：

样例文件	用途
`data/b1_fixtures/b1_fixture_input.json`	Fixture 模式：验证 B1 独立消息闭环（不依赖 B2-B5）
`data/runtime_input.json`	Integrated 模式：file_reader 主线任务（读文件+总结）
`data/runtime_input_0.json`	Integrated 模式：无工具倾向任务（验证模型直接回答）
`data/runtime_input_2~5.json`	Integrated 模式：分别测试 calculator/file_search/table/convert 四个工具

4.4 基础功能演示命令

Fixture 模式（独立演示，不依赖任何其他模块）：

cd agent/code
python b1_agent_runtime.py \
  --input ../data/b1_fixtures/b1_fixture_input.json \
  --outdir ../outputs/B1_fixture

Integrated mock 模式（串联 B3/B4/B5，LLM 使用 mock）：

cd agent/code
python b1_agent_runtime.py \
  --input ../data/runtime_input.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode mock \
  --outdir ../outputs/B1_runtime

Integrated 真实模式（串联 B3/B4/B5，LLM 真实推理）：

cd agent/code
python run_full_demo.py \
  --input ../data/runtime_input.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode prompt_json \
  --outdir ../outputs/full_demo

观察要点：

Fixture 模式：outputs/B1_fixture/ 下生成 messages.json、trace.json、final_answer.md 三个文件
消息角色顺序应为：system → user → assistant(tool_calls) → tool → assistant(final)
trace.json 中 status=success
Integrated 模式额外生成 B3/B4/B5 相关产物（tools_schema、llm_calls、selected_memory 等）

4.5 基础功能输出格式

输出文件 / 返回字段	格式	说明
`messages.json`	JSON 数组	完整消息序列，每条消息含 `role`、`content`，AIMessage 可能含 `tool_calls`
`trace.json`	JSON 对象	执行轨迹：`status`, `tool_rounds_used`, `llm_call_count`, `turns[]`(每轮详情), `error`
`final_answer.md`	Markdown	从最终 AIMessage.content 提取的纯文本回答
`runtime_log.jsonl`	JSONL（追加）	每次运行的时间戳、状态、耗时
`run_agent()` 返回值	Python dict	`conversation_id`, `status`, `final_answer`, 各产物路径, `elapsed_ms`

4.6 基础功能结果截图

[在此处插入 Fixture 模式运行截图]
[在此处插入 messages.json 消息序列截图（展示 system → user → assistant → tool → assistant 五条消息）]
[在此处插入 final_answer.md 内容截图]
[在此处插入 trace.json 中 status=success 的截图]

示例占位：

5. 进阶要求实现与演示

5.1 选择的进阶要求

依据 PPT 任务书 B1 的 5 项进阶要求，全部实现：

进阶要求	是否完成	对应文件 / 函数	简要说明
进阶1：多轮用户输入	✅ 完成	`b1_agent_runtime.py:run_agent()` 第 412-564 行	支持 `user_inputs` 数组，顺序处理多轮对话，`trace.multi_user` 记录每轮详情
进阶2：断点续跑(Checkpoint/Resume)	✅ 完成	`b1_checkpoint.py`（89 行） + `run_agent()` 第 331-358、396-401、546-550 行	每轮自动保存完整状态，`--resume` 恢复，成功后自动清理 checkpoint
进阶3：批量任务处理	✅ 完成	`b1_agent_runtime.py:_run_batch()` 第 222-321 行 + `b1_batch_runner.py`	检测 `tasks+common_config` 自动分发，生成 `batch_summary.json` + `batch_report.md`
进阶4：历史消息压缩	✅ 完成	`b1_history_compressor.py`（141 行） + `run_agent()` 第 449-459 行	超 token 上限自动将中间消息压缩为摘要，保留 system + 末尾消息
进阶5：多 Prompt 模板切换与对比	✅ 完成	`b1_agent_runtime.py:_select_prompt_template()` 第 132-146 行 + `_collect_multi_prompt_answers()` 第 149-217 行	支持 `auto`（关键词匹配）和 `compare`（并行对比多套模板）两种模式

5.2 进阶功能 1：多轮用户输入

功能说明

基础版本只支持单个 user_input，用户无法在对话中途追问。进阶 1 支持 user_inputs 数组，实现多轮对话：第一轮读取文件获得结果后，第二轮可以直接追问"刚才的内容提到了哪些组件？"，Agent 能基于完整对话历史回答。

实现路径

文件 / 函数 / 脚本	作用
`b1_agent_runtime.py:run_agent()` 第 412-415 行	检测 `user_inputs` 数组，构建多轮输入序列
`b1_agent_runtime.py:run_agent()` 第 439-564 行	外层循环：遍历每个 user_input，内层维持 Agent Loop
`b1_agent_runtime.py:_validate_runtime_input()` 第 28-35 行	校验 `user_inputs` 数组格式
`multi_user_turns` 列表 (第 423 行)	收集每轮用户输入的独立统计（工具轮次、LLM 次数、最终回答、耗时）

流程：

[第1轮 user_input] → [Agent Loop（工具调用+推理）] → [final_answer_1]
→ [第2轮 user_input 作为新 HumanMessage 追加到 messages] → [Agent Loop] → [final_answer_2]
→ ... → [所有轮次的回答合并写入 final_answer.md]

输入格式与样例

字段 / 输入文件 / 配置项	类型 / 格式	是否必需	说明
`user_inputs`	`string[]`	是（替代 `user_input`）	多轮用户输入字符串数组
`multi_turn_mode`	`"sequential"`	否（默认）	多轮模式

样例 runtime_input.json（多轮）：

{
  "conversation_id": "conv_multi_001",
  "user_inputs": [
    "帮我阅读 docs/agent_intro.txt，总结三条中文要点。",
    "刚才读到的内容中提到了哪些组件？"
  ],
  "multi_turn_mode": "sequential",
  "system_prompt_path": "../prompts/local_tool_agent.txt",
  "toolset": "basic_tools",
  "max_turns": 3,
  "save_memory": "conversation"
}

演示命令

cd agent/code
python -c "
import json
r = json.load(open('../data/runtime_input.json', encoding='utf-8'))
r['user_inputs'] = [
    '帮我阅读 docs/agent_intro.txt，总结三条中文要点。',
    '刚才读到的内容中提到了哪些组件？'
]
r.pop('user_input', None)
with open('/tmp/runtime_multi_input.json', 'w', encoding='utf-8') as f:
    json.dump(r, f, ensure_ascii=False)
"
python b1_agent_runtime.py --input /tmp/runtime_multi_input.json \
  --tools_config ../configs/tools.yaml --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml --llm_mode mock --outdir ../outputs/B1_runtime/multi_turn

输出格式

输出文件 / 返回字段	格式	说明
`trace.json` → `multi_user.user_turns`	`integer`	多轮用户输入的总轮数
`trace.json` → `multi_user.details[]`	数组	每轮输入的工具轮次、LLM 调用数、最终回答、耗时
`final_answer.md`	Markdown	合并所有轮次的问答（第X轮问/答格式）

示例图片

[在此处插入多轮对话 messages.json 截图（展示两轮 user → assistant 的消息序列）]
[在此处插入 trace.json 中 multi_user 字段截图]

5.3 进阶功能 2：断点续跑（Checkpoint/Resume）

功能说明

在长时间运行的 Agent 任务中，可能因中断（如程序崩溃、手动终止、GPU OOM）导致进度丢失。Checkpoint 机制在每轮 LLM 调用后自动保存完整的执行状态（messages、tool_rounds、llm_calls、turns 等），支持通过 --resume 参数从断点恢复继续执行，避免重复调用 LLM 和工具。

设计要点：

每轮保存：checkpoint_interval 控制保存频率（默认每轮 1 次）
成功自动清理：任务正常结束后自动删除 checkpoint.json，失败时保留
原子写入：通过 write_json() 确保 checkpoint 文件完整性
版本校验：checkpoint 文件含 version 字段，不兼容版本拒绝加载

实现路径

文件 / 函数 / 脚本	作用
`b1_checkpoint.py:save_checkpoint()`	保存完整运行时状态到 `checkpoint.json`（messages、轮次、turns、runtime）
`b1_checkpoint.py:load_checkpoint()`	读取并校验 checkpoint 文件，返回状态字典
`b1_checkpoint.py:checkpoint_exists()`	检查是否存在有效的 checkpoint 文件
`b1_agent_runtime.py:run_agent()` 第 352-358 行	检测 `--resume`，如果存在 checkpoint 则恢复状态
`b1_agent_runtime.py:run_agent()` 第 546-550 行	每轮结束后保存 checkpoint
`b1_agent_runtime.py:run_agent()` 第 584-588 行	成功后自动清理 checkpoint 文件

流程：

[正常运行] → [每轮 LLM 后 save_checkpoint() → checkpoint.json]
            → [任务成功] → [删除 checkpoint.json]

[中断后恢复] → --resume → checkpoint_exists() → load_checkpoint()
             → [恢复 messages/tool_rounds/llm_calls/turns] → [继续 Agent Loop]

输入格式与样例

字段 / 输入文件 / 配置项	类型 / 格式	是否必需	说明
`--resume`	CLI flag	否	启用断点恢复
`--checkpoint_interval`	`integer`	否（默认 1）	每隔多少轮保存一次 checkpoint

演示命令

cd agent/code

# 1. 正常运行（自动保存 checkpoint，成功后自动删除）
python b1_agent_runtime.py \
  --input ../data/runtime_input.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode mock \
  --outdir ../outputs/B1_runtime/checkpoint_demo

# 2. 从 checkpoint 恢复（如果上一步失败，checkpoint.json 保留，可直接恢复）
python b1_agent_runtime.py \
  --input ../data/runtime_input.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode mock \
  --outdir ../outputs/B1_runtime/checkpoint_demo \
  --resume

输出格式

输出文件 / 返回字段	格式	说明
`checkpoint.json`	JSON 对象	含 `version`, `messages`, `tool_rounds`, `llm_calls`, `turns`, `all_tool_messages`, `runtime`, `conversation_id`, `turn_index_at_checkpoint`, `timestamp`
`trace.json` → `checkpoint.resumed`	`boolean`	本次运行是否从 checkpoint 恢复
`trace.json` → `checkpoint.checkpoints_saved`	`integer`	本次运行保存了多少次 checkpoint

示例图片

[在此处插入 checkpoint.json 文件内容截图]
[在此处插入 --resume 运行日志截图（显示 checkpoint=resume turn_index=XXX）]

5.4 进阶功能 3：批量任务处理（Batch Runner）

功能说明

支持一次性提交多个 Agent 任务，每个任务独立运行并生成完整产物，最终产出汇总对比报告。适用于批量评估不同工具在不同输入下的表现。

run_agent() 函数自动检测输入中是否同时包含 tasks 和 common_config，是则自动走批量模式，内部循环调度每个子任务，无需额外的批量执行器。

实现路径

文件 / 函数 / 脚本	作用
`b1_agent_runtime.py:_run_batch()` 第 222-321 行	批量模式核心：检测 `tasks+common_config`，分发执行，生成汇总
`b1_batch_runner.py`	独立批量 CLI 入口
`b1_agent_runtime.py:run_agent()` 第 341-343 行	自动检测批量输入格式

流程：

[读取 batch input] → [检测 tasks + common_config]
→ [遍历 tasks]
  → [合并 common_config + task 特定字段 → runtime_input.json]
  → [run_agent() 执行单个任务] → [收集结果]
→ [汇总 batch_summary.json + batch_report.md]

演示命令

cd agent/code
python b1_batch_runner.py \
  --input ../data/batch_input_demo.json \
  --tools_config ../configs/tools.yaml \
  --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml \
  --llm_mode mock \
  --outdir ../outputs/batch_demo

批量输入格式（data/batch_input_demo.json）：

{
  "batch_id": "batch_demo_001",
  "common_config": {
    "conversation_id": "batch_conv",
    "system_prompt_path": "../prompts/local_tool_agent.txt",
    "toolset": "basic_tools",
    "max_turns": 3,
    "save_memory": "none"
  },
  "tasks": [
    { "task_id": "task_001", "user_input": "帮我阅读 docs/agent_intro.txt..." },
    { "task_id": "task_002", "user_input": "帮我计算 23 * 17 + 9 的结果" }
  ]
}

输出格式

输出文件 / 返回字段	格式	说明
`batch_summary.json`	JSON 对象	批量汇总：`batch_id`, `summary` (total_tasks/success_count/error_count/total_elapsed_ms), `results[]`
`batch_report.md`	Markdown	人类可读的批量报告（表格+每任务详情）
`{task_id}/messages.json`	JSON 数组	每个任务的完整消息序列
`{task_id}/trace.json`	JSON 对象	每个任务的执行轨迹
`{task_id}/final_answer.md`	Markdown	每个任务的最终回答

示例图片

[在此处插入 batch_summary.json 内容截图（展示 5/5 success）]
[在此处插入 batch_report.md 汇总表格截图]

5.5 进阶功能 4：历史消息压缩（History Compression）

功能说明

在多轮工具调用场景下，messages 列表会迅速增长（AIMessage + ToolMessage 交替追加），超过模型上下文窗口限制。压缩功能在 token 数超过阈值时自动触发，将中间的旧消息压缩为一个中文摘要段落，保留 system message + 摘要 + 最近 4 条消息，大幅减少 token 消耗。

压缩策略：

只压缩 ≥6 条消息的对话（太少不值得压）
保留 system message（角色定义不能丢）+ 最近 4-5 条（当前上下文）
中间部分调用 B4（mock 模式）生成 ≤200 字中文摘要
如果 B4 不可用，fallback 到基于规则的摘要生成

实现路径

文件 / 函数 / 脚本	作用
`b1_history_compressor.py:estimate_tokens()`	字符级 token 估算（`total_chars / 2.5`），适配中英文混合场景
`b1_history_compressor.py:should_compress()`	判断是否需要压缩（消息数 ≥6 且 token 超限）
`b1_history_compressor.py:compress_messages()`	执行压缩：提取中间消息 → 生成摘要 → 重建压缩后 messages
`b1_history_compressor.py:_generate_summary()`	调用 B4 mock 生成摘要
`b1_history_compressor.py:_fallback_summary()`	规则兜底摘要（不依赖 B4）
`b1_agent_runtime.py:run_agent()` 第 449-459 行	在每轮 Agent Loop 前检查并触发压缩

流程：

[每轮 Agent Loop 开始前]
→ [should_compress(messages, max_tokens)] → 消息数 ≥6 且 token 超限？
  → [是] → [compress_messages()]
           → [system message 保留]
           → [中间消息截取 → _generate_summary() → 摘要]
           → [末尾 4-5 条保留]
           → [返回压缩后的 messages + metadata]
  → [否] → [继续正常流程]
→ [compression_events 记录到 trace.json]

输入格式与样例

字段 / 输入文件 / 配置项	类型 / 格式	是否必需	说明
`compression_enabled`	`boolean`	否（默认 `true`）	是否启用压缩
`max_context_tokens`	`integer`	否（默认 `4096`）	token 上限阈值

演示命令

cd agent/code
python -c "
import json
r = json.load(open('../data/runtime_input.json', encoding='utf-8'))
r['compression_enabled'] = True
r['max_context_tokens'] = 4096
with open('/tmp/runtime_compress.json', 'w', encoding='utf-8') as f:
    json.dump(r, f, ensure_ascii=False)
"
python b1_agent_runtime.py --input /tmp/runtime_compress.json \
  --tools_config ../configs/tools.yaml --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml --llm_mode mock --outdir ../outputs/B1_runtime/compress_test

输出格式

输出文件 / 返回字段	格式	说明
`trace.json` → `compression.triggered`	`boolean`	是否触发了压缩
`trace.json` → `compression.events[].compressed`	`boolean`	该次压缩是否生效
`trace.json` → `compression.events[].savings_tokens`	`integer`	压缩节省的 token 数
`trace.json` → `compression.events[].summary_text`	`string`	生成的中文摘要文本

示例图片

[在此处插入 compression events 截图（展示压缩前后的 token 数和摘要文本）]

5.6 进阶功能 5：多 System Prompt 模板切换与对比

功能说明

基础版本固定使用一个 system prompt。进阶 5 支持两种模式：

auto（自动匹配）：根据用户输入关键词（如"计算"、"文件"）自动从 auto_prompt_map 中选择对应的专家 prompt
compare（并行对比）：同一任务使用多套 prompt 模板分别运行完整的 Agent Loop，对比各模板的回答质量

内置四套 prompt 模板：

local_tool_agent.txt — 通用工具代理
file_expert_agent.txt — 文件阅读专家（含文件内容分析指引）
calculation_agent.txt — 计算分析专家（含数学推导指引）
conversion_agent.txt — 格式转换专家（含格式规范指引）

实现路径

文件 / 函数 / 脚本	作用
`b1_agent_runtime.py:_select_prompt_template()` 第 132-146 行	auto 模式：关键词匹配选择 prompt 模板
`b1_agent_runtime.py:_collect_multi_prompt_answers()` 第 149-217 行	compare 模式：并行遍历多个模板，每个模板独立走完整 Agent Loop
`prompts/local_tool_agent.txt`	通用 prompt 模板
`prompts/file_expert_agent.txt`	文件阅读专家 prompt 模板
`prompts/calculation_agent.txt`	计算分析专家 prompt 模板
`prompts/conversion_agent.txt`	格式转换专家 prompt 模板

流程（compare 模式）：

[读取 prompt_templates 列表]
→ [遍历每个模板]
  → [加载模板文本 → 合并 memory_context → 构建 Prompt-Specific messages]
  → [独立的 Agent Loop（B4 + B3）] → [收集该模板的 final_answer]
→ [所有模板结果写入 trace.json → prompt_comparisons]

输入格式与样例

字段 / 输入文件 / 配置项	类型 / 格式	是否必需	说明
`prompt_mode`	`"single" \| "auto" \| "compare"`	否（默认 `"single"`）	Prompt 模板使用模式
`prompt_templates`	`string[]`	compare 模式必需	要对比的模板路径列表
`auto_prompt_map`	`object` (keyword → path)	auto 模式必需	关键词到模板路径的映射

样例（compare 模式）：

{
  "prompt_mode": "compare",
  "prompt_templates": [
    "../prompts/local_tool_agent.txt",
    "../prompts/file_expert_agent.txt",
    "../prompts/calculation_agent.txt"
  ]
}

样例（auto 模式）：

{
  "prompt_mode": "auto",
  "auto_prompt_map": {
    "计算": "../prompts/calculation_agent.txt",
    "文件": "../prompts/file_expert_agent.txt",
    "转换": "../prompts/conversion_agent.txt"
  }
}

演示命令

compare 模式：

cd agent/code
python -c "
import json
r = json.load(open('../data/runtime_input.json', encoding='utf-8'))
r['prompt_mode'] = 'compare'
r['prompt_templates'] = [
    '../prompts/local_tool_agent.txt',
    '../prompts/file_expert_agent.txt',
    '../prompts/calculation_agent.txt'
]
with open('/tmp/runtime_prompts.json', 'w', encoding='utf-8') as f:
    json.dump(r, f, ensure_ascii=False)
"
python b1_agent_runtime.py --input /tmp/runtime_prompts.json \
  --tools_config ../configs/tools.yaml --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml --llm_mode mock --outdir ../outputs/B1_runtime/prompt_compare

auto 模式：

cd agent/code
python -c "
import json
r = json.load(open('../data/runtime_input.json', encoding='utf-8'))
r['prompt_mode'] = 'auto'
r['auto_prompt_map'] = {
    '计算': '../prompts/calculation_agent.txt',
    '文件': '../prompts/file_expert_agent.txt'
}
with open('/tmp/runtime_auto.json', 'w', encoding='utf-8') as f:
    json.dump(r, f, ensure_ascii=False)
"
python b1_agent_runtime.py --input /tmp/runtime_auto.json \
  --tools_config ../configs/tools.yaml --memory_config ../configs/memory.yaml \
  --model_config ../configs/model.yaml --llm_mode mock --outdir ../outputs/B1_runtime/prompt_auto

输出格式

输出文件 / 返回字段	格式	说明
`trace.json` → `prompt_mode`	`string`	当前使用的 prompt 模式
`trace.json` → `prompt_template_used`	`string`	实际使用的 prompt 模板路径
`trace.json` → `prompt_comparisons[]`	数组	compare 模式下每个模板的结果（`prompt_template`, `ai_message`, `conversation_id`）

示例图片

[在此处插入 prompt_compare 的 trace.json 中 prompt_comparisons 字段截图]
[在此处插入三套模板各自生成的不同 final_answer 对比截图]

6. 与团队系统的集成说明

6.1 被调用方式

B1 通过唯一的公开函数 run_agent() 对外暴露，团队系统（统一前端 unified_demo_server.py）通过 Python 函数调用接入：

from b1_agent_runtime import run_agent

result = run_agent(
    input_path="data/runtime_input.json",
    tools_config="configs/tools.yaml",
    memory_config="configs/memory.yaml",
    model_config="configs/model.yaml",
    outdir="outputs/B1_runtime",
    llm_mode="prompt_json",
    resume=False,
    checkpoint_interval=1,
)
# result: {conversation_id, status, final_answer, messages_path, trace_path, ...}

6.2 统一前端调用位置

统一前端功能	调用位置	说明
开始新对话	`unified_demo_server.py:_handle_chat_start()` → `_run_agent_round()` → `run_agent()`	POST `/api/chat/start`
继续对话	`unified_demo_server.py:_handle_chat_continue()` → `_run_agent_round()` → `run_agent()`	POST `/api/chat/continue`
批量任务	`unified_demo_server.py:_handle_chat_batch()` → `run_agent()`	POST `/api/chat/batch`
断点恢复	`unified_demo_server.py` 传 `resume=True` → `run_agent()`	POST `/api/chat/resume`

6.3 B1 调用的外部模块接口

┌─────────────────────────────────────────────────────┐
│                    B1 Runtime                        │
│  run_agent(input_path, tools, memory, model, outdir) │
├─────────────────────────────────────────────────────┤
│                                                      │
│  B5 接口（记忆管理）                                   │
│  from b5_memory import load_memory                   │
│  selected_memory = load_memory(ids, use_global, ...)  │
│  from b5_memory import save_memory                   │
│  saved_memory = save_memory(conv_id, type, ...)       │
│                                                      │
│  B3 接口（工具层）                                     │
│  from b3_tool_layer import get_tools_schema           │
│  tools_schema = get_tools_schema(config, toolset)     │
│  from b3_tool_layer import execute_tool_calls         │
│  tool_messages = execute_tool_calls(calls, config, ..)│
│                                                      │
│  B4 接口（LLM 推理）                                  │
│  from b4_local_agent_llm import generate_ai_message   │
│  ai_message = generate_ai_message(model, msgs, ...)   │
│                                                      │
│  B2 接口（间接调用）                                   │
│  通过 B3 → execute_tool_calls → b2_run_skill          │
│                                                      │
└─────────────────────────────────────────────────────┘

6.4 进阶参数传递链路

前端配置弹窗 → unified_demo_server.py → runtime_input.json → run_agent() → B3/B4/B5：

retrieval_type → load_memory(retrieval_type=...)（B5 进阶）
save_memory → save_memory(save_type=...)（B5 进阶）
use_native_tools → generate_ai_message(use_native_tools=...)（B4 进阶④）
model_profile → generate_ai_message(model_profile=...)（B4 进阶③）

6.5 联调注意事项

B1 的 run_agent() 函数签名中所有 config 参数（tools_config, memory_config, model_config）在 integrated 模式下为必需，fixture 模式下可为 None
批量模式通过输入数据自动检测（tasks + common_config 同时存在），无需单独接口
B1 的产物文件统一写入 outdir 目录，调用方通过返回值中的 messages_path/trace_path/final_answer_path 获取路径
退出码约定：0 = 成功（业务错误已写入结构化产物），1 = 致命错误

7. 已知问题与后续改进

问题	当前原因	后续改进
Mock 模式下历史压缩调用 B4 mock 生成摘要，摘要质量一般	mock 模式的 B4 返回固定模板回答，不是真正的 LLM 摘要	prompt_json 模式下 B4 真实推理可生成更精准的摘要
Fixture 模式不支持批量	Fixture 使用固定预设数据，批量分发需修改预设逻辑	可为 fixture 模式添加批量变体，用于压力测试消息流正确性
Token 估算使用字符数/2.5 的简单启发式	避免引入 tiktoken 等重量级依赖，适配中英混合文本	可选项：集成 tokenizer 获取精确 token 数，但会增加模型加载开销
Prompt 对比模式在大型模板数下耗时线性增长	每个模板独立运行完整 Agent Loop（含多次 B4 调用）	可考虑共享工具执行结果，仅对比 LLM 推理环节
Checkpoint 文件包含完整 messages 列表，多轮对话下文件较大	当前直接将整个 messages 数组序列化写入	可考虑增量存储（仅存差异）或压缩存储

附录：实验对比脚本

B1 模块提供 6 个独立实验脚本，覆盖关键参数和配置的对比分析：

脚本	用途	演示命令
`experiments/compare_llm_modes.py`	mock vs prompt_json 模式对比	`python compare_llm_modes.py --input ../data/runtime_input.json --outdir ../outputs/exp_compare_modes`
`experiments/compare_max_turns.py`	不同 max_turns 值对比（1,2,3,5）	`python compare_max_turns.py --input ../data/runtime_input.json --turns 1 2 3 5 --outdir ../outputs/exp_max_turns`
`experiments/compare_prompts.py`	不同 system prompt 对比	`python compare_prompts.py --input ../data/runtime_input.json --outdir ../outputs/exp_compare_prompts`
`experiments/error_case_analysis.py`	错误案例和异常处理验证	`python error_case_analysis.py --outdir ../outputs/exp_error_cases`
`experiments/with_without_memory.py`	有无 memory 对比	`python with_without_memory.py --input ../data/runtime_input.json --outdir ../outputs/exp_memory_compare`
`experiments/batch_experiment.py`	端到端批量实验	`python batch_experiment.py --outdir ../outputs/exp_batch`

所有实验脚本从 agent/code/ 启动，需先 cd experiments，结果统一写入 outputs/exp_* 目录。

README.md Unescape Escape

个人模块 README

1. 模块概述

1.1 模块名称

1.2 模块说明

1.3 完成情况概览

2. 环境、模型与数据依赖

2.1 运行环境

2.2 模型依赖

2.3 数据集或样例数据依赖

2.4 安装步骤

3. 文件结构与接口边界

3.1 文件结构

3.2 接口边界

4. 基础要求实现与演示

4.1 基础功能说明

4.2 基础功能实现路径

4.3 基础功能输入格式与样例

4.4 基础功能演示命令

4.5 基础功能输出格式

4.6 基础功能结果截图

5. 进阶要求实现与演示

5.1 选择的进阶要求

5.2 进阶功能 1：多轮用户输入

功能说明

实现路径

输入格式与样例

演示命令

输出格式

示例图片

5.3 进阶功能 2：断点续跑（Checkpoint/Resume）

功能说明

实现路径

输入格式与样例

演示命令

输出格式

示例图片

5.4 进阶功能 3：批量任务处理（Batch Runner）

功能说明

实现路径

演示命令

输出格式

示例图片

5.5 进阶功能 4：历史消息压缩（History Compression）

功能说明

实现路径

输入格式与样例

演示命令

输出格式

示例图片

5.6 进阶功能 5：多 System Prompt 模板切换与对比

功能说明

实现路径

输入格式与样例

演示命令

输出格式

示例图片

6. 与团队系统的集成说明

6.1 被调用方式

6.2 统一前端调用位置

6.3 B1 调用的外部模块接口

6.4 进阶参数传递链路

6.5 联调注意事项

7. 已知问题与后续改进

附录：实验对比脚本

README.md