AWorld = 「智能体自我改进（Self-Improvement）运行时」

1. 总体定位与核心命题

AWorld = 「智能体自我改进（Self-Improvement）运行时」。它不是单纯的调用 LLM 的 SDK，而是一个围绕「智能体在可组合的运行环境中持续演化」的框架：
1) 多智能体拓扑（Swarm）＝结构化编排。
2) 上下文/记忆/工具/事件/追踪统一协议＝可控与可观测。
3) Forward 采集 → Backward 训练（外部 RL / 微调）闭环。
4) 云原生 + 分布式执行通道（Local / Ray / Spark / K8s）＝高并发演化。
5) MCP、工具体系、Agent-as-Tool ＝ 增强行动空间与反馈质量。

目标：让智能体不止「回答」，而是「组织步骤—调用工具—协调其他 agent—吸收经验—生成高质量训练/演化数据」，持续提升行为策略与知识资源。

2. 设计哲学（从代码与实现抽象出的原则）

3. 核心对象总览（概念图心智模型）

输入(Task) → 调度(Runner) → 上下文(Context) → Agent/Swarm 决策(policy) → ActionModel → 工具/子Agent 执行 → Observation → 记忆写入 / 事件流输出 → 循环直到终止 → 汇总 Outputs → 轨迹沉淀供训练（Backward）。

4. 运行主链路（请求一次执行的生命周期）

1. 构造 Task（任务 id / 输入 / 工具集合 / 绑定 agent 或 swarm / 配置 TaskConfig）
2. Runner（如 Runners.sync_run / streamed_run_task）装配 Context（含 session_id / task_id / outputs / policy 状态）
3. 若是 Swarm：Swarm.reset() 构建 AgentGraph（拓扑分析 + 有向无环/团队模式判断 + star 识别）
4. Agent.run / async_run：

• 发送 StepOutput.start 事件（事件总线或直接输出）
• pre_run 钩子
• 调用 async_policy：
  a. async_desc_transform 生成工具/代理/MCP function schema
  b. async_messages_transform：系统提示 → 记忆检索/截取 → 人类输入/工具结果注入 → 消息压缩(ProcessMessages)
  c. LLM 调用（支持流式）→ ModelResponse（content + tool_calls）
  d. 输出解析器 LlmOutputParser → AgentResult(actions)
• post_run：封装成 ToolMessage / AgentMessage / GroupMessage

1. 动作执行：

• 若工具：exec_tool → ToolResult → MemoryToolMessage 写入
• 若 agent-as-tool：递归进入子 Agent 流程

1. 直到 agent.finished / max_steps / TaskConfig 条件 → 汇总 answer / trajectory
2. Memory 触发器（短期→摘要→可能长记忆提炼：User Profile / Agent Experience）异步或按阈值运行
3. 轨迹（ActionModel 序列 + 中间 Observation + Tool 调用结果）用于后续训练/评估。

5. Agent 内部结构深解

关键字段与流程（来自 llm_agent.py + base.py）：

• tool_names / handoffs：可直接调用的工具 & 可「委派」的其他 Agent。
• mcp_servers / sandbox：通过 MCP 标准协议向外部能力扩展。
• system_prompt_template + agent_prompt：二级提示系统（角色定义 + 上下文注射）。
• use_tools_in_prompt VS function call schema：二种工具暴露模式。
• memory：MemoryFactory.instance() 单例访问，统一写入/提取。
• async_messages_transform：
  1) 系统提示写入记忆（首次）
  2) 根据 Observation 区分工具结果 / 用户输入
  3) 取历史 N 条（受 memory_config.history_number 与压缩策略影响）
  4) 插入 tool_calls 协议字段（兼容 OpenAI 风格）
  5) PromptProcessor（裁剪 / 优化 / token 压缩）
• _call_llm_model：支持流式（事件总线异步增量输出）和非流式。
• model_output_parser：将 function call / JSON 语义解析成统一 ActionModel 列表：
• 如果含 tool_calls：拆分为 (tool_name, action_name, params)
• 否则作为「自然语言策略输出」回退形式
• execution_tools（当 wait_tool_result=True）：同步执行工具并聚合多结果，再作为新的 Observation（典型 ReAct 循环）。
• 记忆写入分类：MemorySystemMessage / MemoryHumanMessage / MemoryAIMessage / MemoryToolMessage。
• Hook 机制：PRE_LLM_CALL / POST_LLM_CALL，便于插拔监控或上下文再加工。

简化策略循环伪流程（逻辑顺序）：
Init → 描述获取 → 消息构建 → LLM 推理 → 输出解析 → (可选工具执行) → Memory 写入 → 判断完成。

6. Swarm 多智能体拓扑体系

swarm.py 实现对多 Agent 结构抽象：
支持三大范式：

1. WORKFLOW（编排式 / DAG 串并行）
2. HANDOFF（接力式—Agent 彼此作为工具，强调控制权转移）
3. TEAM（协调式—领导者+执行者星型 / 层级扩展）

核心构件：

• AgentGraph：存储有向图（predecessor / successor / ordered_agents）。
• 拓扑构建器：WorkflowBuilder / HandoffBuilder / TeamBuilder
• Workflow 支持嵌套 (BaseAgent | tuple => 串行组合 | list => 并行组合) 自动封装 SerialableAgent / ParallelizableAgent。
• Handoff 验证成对约束，强调「右节点作为左节点的延续工具」，常用于对话/接力思考链。
• Team 识别领导者（root），其余节点被挂载为 handoffs；星型结构可自动降级/切换构建类型。
• 动态特性：
• 自动检测星型拓扑 _is_star() 进而智能选择 Team 构建。
• DAG Cycle 检测（Workflow 禁止环；循环需求用 LoopableAgent 包裹）。
• Loop 机制：loop_agent 用 LoopableAgent 替换特定节点，加上 max_run_times、stop_func 实现局部回路。

价值：开发者可用简单 Python 结构（嵌套 tuple/list/实例）表达复杂执行图，不直接操心图算法细节。

7. 动作—观察协议与消息分类

动作统一：ActionModel

• tool_name / action_name / params / policy_info（原始 LLM 文本辅助）
  消息分类（事件投递/执行调度时使用）：
• ToolMessage：调用外部工具
• AgentMessage：交给另一个 Agent
• GroupMessage：复杂复合调用（多工具/多子 agent）
• StepOutput / MessageOutput：统一输出结构（流式渲染 / 终端 UI / Web UI）。

Observation（环境可见状态）反馈给 Agent：再进入下一轮。

8. Context + Prompt 压缩优化

PromptProcessor + ContextRuleConfig 作用：

• 预算管理：截断历史消息数 / token 控制
• 可选 LLM 压缩（LlmCompressionConfig.enabled）→ 历史对话摘要 + 保留关键 tool_calls
• 处理指标追踪（span attributes：origin_len / compressed_len / ratio）
  结果：消息量大时保持语义相关性与成本平衡。

9. 记忆系统（短期窗口 → 结构化长记忆）

文件：memory.py
分层：

• 短期：对话轮次序列（message / tool_result / system）→ LLM 直接上下文
• 长期：在触发条件下提炼出
• 用户画像（UserProfile）
• 事实/知识（Fact）
• 智能体经验（AgentExperience：技能分类 + 动作链条 + 上下文 + outcome）
  触发配置（TriggerConfig）：消息计数 / 时间 / 重要性关键字（error/success/失败/完成）。
  抽取配置（ExtractionConfig）：
• prompt 模板包含结构化 JSON 输出约束（高可解析性）
• 最大抽取条数 + 置信阈值（未来可挂检测器）
  存储策略（StorageConfig）：去重 + 老化清理（保鲜策略）。
  ProcessingConfig：后台处理 / 实时处理模式选择（吞吐与时效权衡）。

关键抽取 Prompt 设计特点：

• 显式分类键（可索引）
• Few-Shot 示例保证输出结构稳定
• 语言保持一致性（避免跨语言丢失文化语境）

好处：

• 画像有助于后续上下文个性化
• AgentExperience 可反哺微调 / 检索式提示（技能模板化）
• 减少重复 token 投入（用经验摘要替代大量原始交互）

10. 工具 / MCP / Sandbox 体系

• Tool 描述经 tool_desc_transform → OpenAI function schema → LLM 可调用。
• Agent 也被包装成「工具」供其他 Agent 决策调用（Agent-as-Tool pattern）。
• MCP（Model Context Protocol）集成：mcp_servers + sandbox，可同时挂载本地脚本 / 远程服务 / 文件系统 / 浏览器自动化。
• Sandbox 用于执行隔离 + 资源访问控制 + 工具注册（高级场景：安全审计 / 副作用管控）。

11. 事件（eventbus）与追踪（trace）

• eventbus：流式输出、实时 UI 更新、Hook 插件可监听。
• trace（OpenTelemetry 风格）：
• span 前缀 agent_llm_context_process 等
• 记录 token 压缩比、消息长度、时延
• 便于后期：性能剖析 / 失败重放 / 指标仪表板接入
  整体 = 「可观测性内建」，契合生产级系统需求。

12. RuntimeEngine：执行后端抽象

实现：LocalRuntime / SparkRuntime / RayRuntime / K8sRuntime（占位继承 Local）
统一接口：

• build_engine() 延迟初始化
• execute(funcs) 并行函数执行（无状态任务，如环境多进程 rollout）
  LocalRuntime：
• reuse_process 快速路径（单任务）
• 否则 ProcessPoolExecutor + 超时控制
  RayRuntime：
• remote 包装 + gather
  SparkRuntime：
• parallelize + collect（driver 端）
  作用场景：
• 多环境并发采样（生成策略数据 / 合成工具调用数据）
• 大规模验证 / 评估（Pass@K 统计）

13. Task 与 Runner

Task（dataclass）聚合：

• 输入 / 工具集合 / agent 或 swarm / hooks / context / outputs / endless_threshold
  Runner：
• 模板方法：pre_run → do_run（核心逻辑）→ post_run
• 支持守护线程（daemon）用于持续监控（如心跳 / 中断检测）
  Runners API（未展示全文件但可推断）：封装 sync / async / stream 执行入口。

14. 自我改进闭环（Forward → Backward）

Forward（在线运行时）输出：

• 完整上下文（messages + compressed variant）
• 工具调用决策链（ActionModel 序列）
• 工具结果 + 是否成功 + 时间/耗材指标
• 经验与画像抽取（结构化 JSON）

Backward：

• README 展示与 ms-swift (RL 框架) 打补丁整合的流程：AWorld 做 「环境」+多 agent 策略 → 采样 → RL / 监督微调 / 自蒸馏。
• 多样拓扑 + 工具使用 = 比单一对话更丰富的「训练 signal」，提高策略泛化。
• 未来可叠加：奖励模型（Reward shaping）、失败回放（Trajectory pruning）、技能库动态演化。

15. 可扩展点清单

16. 典型拓扑与适配场景

17. 性能与资源治理

• 消息压缩（比率指标追踪）
• Stream 输出避免大响应阻塞
• 工具执行隔离（Sandbox + 进程池）
• 分布式后端抽象（可按资源成本切换）
• LLM 调用错误 → 回退提示输出 & 记忆写入（便于后续失败分析）
• 超时控制（executor + tool result）

18. 与常见 Agent 框架差异（抽象总结）

19. 建议学习路线（循序渐进）

1. Quickstart 单 Agent（理解 Observation → ActionModel）
2. Swarm 三模式（Workflow vs Team vs Handoff）
3. 工具 & MCP 注入（文件系统例子）
4. 记忆：短期历史 + 提炼（调低阈值快速观察抽取 JSON）
5. Hook / trace 观测（插入日志检查消息剪裁）
6. 构建并行/串行混合拓扑（Serialable + Parallelizable 嵌套）
7. 分布式执行（切换 Local → Ray）
8. 经验数据导出与再训练（接入 RL / 微调框架）

20. 潜在改进方向（基于代码阅读的可行建议）

21. 总结（认知闭包）

AWorld 已建立一套「从执行到演化」的结构化抽象：

• 以 Swarm 图结构组织智能体协作
• 以 Memory + 提炼 Prompt 把交互沉淀为可重复引用的知识与技能模式
• 以 ActionModel + Tool/Agent 调度扩展行动空间
• 以 RuntimeEngine/Hook/trace 支撑规模化与可观测
• 以 Forward→Backward 形态对接外部训练系统，完成自我改进闭环

它的独特价值在于：明确把「可解析经验」作为一等结果产物，而不仅是对话输出；同时减少多智能体编排的样板与图管理负担，让研究者专注于策略设计与演化逻辑。

需求：输出一个「最小 多智能体 + 工具 + 记忆 + 经验抽取」演示脚本设计草案。下面给出结构化方案：包含目标、组件选型、目录、核心代码骨架、运行步骤、可观察点与扩展建议。内容力求精简却可直接落地。

✅ 目标

在最少代码下演示：

1. 两个协作 Agent（Researcher → Summarizer）组成 Workflow/Team。
2. 挂载一个自定义简单 Tool（关键字计数 / 情感粗评）。
3. 激活短期记忆并触发一次「Agent 经验」与「用户画像」抽取。
4. 演示一次任务调用：输入一段文本，Researcher 调用工具，Summarizer 汇总。
5. 输出：最终回答、工具调用轨迹、记忆内容（短期 + 抽取后的结构化 JSON）。

✅ 设计要点

• 使用已有 Agent 类（不自定义子类，减少样板）。
• 给 AgentConfig.memory_config 打开 long_term + 降低触发门槛（如 message_count_threshold=2）。
• Tool 采用同步 Tool（BaseTool 继承）形式，只实现 reset / do_step。
• 通过手动「指导」 LLM（为简化，不实际调用真实 LLM 时可 mock 返回 tool_calls JSON）。
• 若想真实调用，需要设置环境变量（LLM_API_KEY、LLM_MODEL_NAME）。Mock 版本用伪 LLM 输出替换 _call_llm_model。

✅ 最小目录建议（demo 下）

demo_minimal/
  run_demo.py
  tools/
    __init__.py
    text_analyzer_tool.py
  config/
    agent_researcher.yaml (可选，或直接在代码里写)
    agent_summarizer.yaml (可选)
  README.md

demo_minimal/
  run_demo.py
  tools/
    __init__.py
    text_analyzer_tool.py
  config/
    agent_researcher.yaml (可选，或直接在代码里写)
    agent_summarizer.yaml (可选)
  README.md

✅ 自定义 Tool 设计

名称：text_analyzer
意图：输入 ActionModel（policy_info = 原文），输出 Observation：

• content: 结构化文本（JSON 字符串化）
  {
  “keyword_freq”: {“ai”:3,”agent”:2},
  “sentiment”: “neutral”
  }
• action_result: 与每个 action 对应的 ActionResult（content 同上）

伪实现策略：

• 简单分词（按空格+标点过滤）
• 关键字表：[“ai”,”agent”,”model”,”tool”]
• 情感：出现 “good”/”great” 算 positive；出现 “bad”/”error”/”fail” 算 negative；否则 neutral。

✅ Memory / 经验抽取策略

• 设置 AgentConfig.memory_config.enable_long_term = True
• memory_config.long_term_config = LongTermConfig.create_simple_config(
  message_threshold=2,
  enable_user_profiles=True,
  enable_agent_experiences=True
  )
• 由于真实抽取调用 LLM，最小可选：
  1) 提供真实 key 允许执行；
  2) 或在 demo 中模拟 long_term 抽取：直接打印一个伪 JSON（若用户当前暂不需要真实调用）。

（如想真实：需确保 MemoryFactory.init(…) 中 embedding 与 vector_store 至少配置一个简化 backend；否则仅短期内存存放也可。）

✅ 运行主流程（run_demo.py 逻辑概述）

1. 初始化 MemoryFactory（可选：默认实例）。
2. 构建 Tool 并注册到 ToolFactory（在 import text_analyzer_tool 时自动注册）。
3. 创建两个 Agent：Researcher（调用 text_analyzer + handoff Summarizer）、Summarizer（仅总结）。
4. 构造 Swarm：WorkflowSwarm(Researcher, Summarizer) 或 TeamSwarm(Researcher, Summarizer)。
5. 创建 Task(input=长文本)；注入 tool_names=[“text_analyzer”]；挂 swarm。
6. 使用 Runners.sync_run / streamed_run_task（若需要流式）。
7. 打印：

• 最终 answer
• 工具调用结果
• memory.get_last_n / long_term （如果已触发）输出

1. 强制调用一次（可选）经验抽取函数（如果没有自然触发）。

✅ 关键伪代码骨架

(1) 工具 text_analyzer_tool.py

from aworld.core.tool.base import Tool
from aworld.core.common import Observation, ActionResult
from aworld.core.common import ActionModel
from aworld.core.tool.base import ToolFactory
import re, json

class TextAnalyzerTool(Tool):
    def reset(self, *, seed=None, options=None):
        return Observation(content="ready", action_result=[]), {}

    def do_step(self, action_models, **kwargs):
        # action_models: List[ActionModel]
        contents = []
        for am in action_models:
            text = am.policy_info or ""
            stats = self._analyze(text)
            contents.append(stats)
        merged = contents[-1]
        obs = Observation(
            content=json.dumps(merged, ensure_ascii=False),
            action_result=[
                ActionResult(content=json.dumps(c, ensure_ascii=False), success=True, metadata={}) 
                for c in contents
            ],
            is_tool_result=True
        )
        reward=0.0; terminated=False; truncated=False; info={}
        return (obs, reward, terminated, truncated, info)

    def _analyze(self, text: str):
        tokens = [t.lower() for t in re.findall(r"[a-zA-Z]+", text)]
        keywords = ["ai","agent","model","tool"]
        freq = {}
        for k in keywords:
            freq[k] = tokens.count(k)
        sentiment = "neutral"
        if any(w in tokens for w in ["good","great","excellent"]): sentiment="positive"
        if any(w in tokens for w in ["bad","error","fail","failure"]): sentiment="negative"
        return {"keyword_freq": freq, "sentiment": sentiment}

    def close(self): pass

# 注册
ToolFactory.register("text_analyzer", "analyze text keywords and sentiment")

from aworld.core.tool.base import Tool
from aworld.core.common import Observation, ActionResult
from aworld.core.common import ActionModel
from aworld.core.tool.base import ToolFactory
import re, json

class TextAnalyzerTool(Tool):
    def reset(self, *, seed=None, options=None):
        return Observation(content="ready", action_result=[]), {}

    def do_step(self, action_models, **kwargs):
        # action_models: List[ActionModel]
        contents = []
        for am in action_models:
            text = am.policy_info or ""
            stats = self._analyze(text)
            contents.append(stats)
        merged = contents[-1]
        obs = Observation(
            content=json.dumps(merged, ensure_ascii=False),
            action_result=[
                ActionResult(content=json.dumps(c, ensure_ascii=False), success=True, metadata={}) 
                for c in contents
            ],
            is_tool_result=True
        )
        reward=0.0; terminated=False; truncated=False; info={}
        return (obs, reward, terminated, truncated, info)

    def _analyze(self, text: str):
        tokens = [t.lower() for t in re.findall(r"[a-zA-Z]+", text)]
        keywords = ["ai","agent","model","tool"]
        freq = {}
        for k in keywords:
            freq[k] = tokens.count(k)
        sentiment = "neutral"
        if any(w in tokens for w in ["good","great","excellent"]): sentiment="positive"
        if any(w in tokens for w in ["bad","error","fail","failure"]): sentiment="negative"
        return {"keyword_freq": freq, "sentiment": sentiment}

    def close(self): pass

# 注册
ToolFactory.register("text_analyzer", "analyze text keywords and sentiment")

(2) 构建 Agents 与 Swarm

from aworld.agents.llm_agent import Agent
from aworld.config.conf import AgentConfig
from aworld.core.agent.swarm import Swarm, GraphBuildType

research_conf = AgentConfig(
  llm_provider="openai",
  llm_model_name=os.getenv("LLM_MODEL_NAME","gpt-4o"),
  llm_api_key=os.getenv("LLM_API_KEY"),
  use_tools_in_prompt=False,
  memory_config=AgentConfig().memory_config
)
# 调低记忆阈值:
research_conf.memory_config.enable_long_term=True
from aworld.core.memory import LongTermConfig
research_conf.memory_config.long_term_config = LongTermConfig.create_simple_config(
  message_threshold=2,
  enable_user_profiles=True,
  enable_agent_experiences=True
)

researcher = Agent(
  name="researcher",
  conf=research_conf,
  system_prompt="You are a research agent. First decide whether to call the tool 'text_analyzer' with the raw user input. Then summarize key statistics in natural language. Use function calls when needed.",
  tool_names=["text_analyzer"],
  agent_names=["summarizer"] # handoff 目标
)

summ_conf = research_conf
summarizer = Agent(
  name="summarizer",
  conf=summ_conf,
  system_prompt="You are a summarizer. Receive prior analysis context and produce a concise final answer to the user."
)

swarm = Swarm(researcher, summarizer, build_type=GraphBuildType.WORKFLOW)

from aworld.agents.llm_agent import Agent
from aworld.config.conf import AgentConfig
from aworld.core.agent.swarm import Swarm, GraphBuildType

research_conf = AgentConfig(
  llm_provider="openai",
  llm_model_name=os.getenv("LLM_MODEL_NAME","gpt-4o"),
  llm_api_key=os.getenv("LLM_API_KEY"),
  use_tools_in_prompt=False,
  memory_config=AgentConfig().memory_config
)
# 调低记忆阈值:
research_conf.memory_config.enable_long_term=True
from aworld.core.memory import LongTermConfig
research_conf.memory_config.long_term_config = LongTermConfig.create_simple_config(
  message_threshold=2,
  enable_user_profiles=True,
  enable_agent_experiences=True
)

researcher = Agent(
  name="researcher",
  conf=research_conf,
  system_prompt="You are a research agent. First decide whether to call the tool 'text_analyzer' with the raw user input. Then summarize key statistics in natural language. Use function calls when needed.",
  tool_names=["text_analyzer"],
  agent_names=["summarizer"] # handoff 目标
)

summ_conf = research_conf
summarizer = Agent(
  name="summarizer",
  conf=summ_conf,
  system_prompt="You are a summarizer. Receive prior analysis context and produce a concise final answer to the user."
)

swarm = Swarm(researcher, summarizer, build_type=GraphBuildType.WORKFLOW)

(3) 任务执行

from aworld.core.task import Task
from aworld.runner import Runners
task = Task(
  input="Give an overview of how AI agent tool ecosystems evolve. Mention model, agent, tool, agent again. It is great but sometimes there are failure cases.",
  swarm=swarm
)
result = Runners.sync_run(input=task.input, swarm=swarm)
print("Final answer:", result.answer)

from aworld.core.task import Task
from aworld.runner import Runners
task = Task(
  input="Give an overview of how AI agent tool ecosystems evolve. Mention model, agent, tool, agent again. It is great but sometimes there are failure cases.",
  swarm=swarm
)
result = Runners.sync_run(input=task.input, swarm=swarm)
print("Final answer:", result.answer)

(4) 强制查看 memory

mem = researcher.memory
short_hist = mem.get_last_n(20, filters={"agent_id":researcher.id(), "session_id": task.session_id, "task_id": task.id})
print("Short-term messages:", [ (m.metadata.role if hasattr(m,'metadata') else '', getattr(m,'content','')[:60]) for m in short_hist])

# 如果已有 long term:
# （依据具体 Memory 实现，可能需要 mem.search 或专门 API；若未实现，这里展示设计）
profiles = await mem.retrival_user_profile(user_id=task.user_id or "user", user_input="agent tool usage")
print("User profiles:", profiles)

mem = researcher.memory
short_hist = mem.get_last_n(20, filters={"agent_id":researcher.id(), "session_id": task.session_id, "task_id": task.id})
print("Short-term messages:", [ (m.metadata.role if hasattr(m,'metadata') else '', getattr(m,'content','')[:60]) for m in short_hist])

# 如果已有 long term:
# （依据具体 Memory 实现，可能需要 mem.search 或专门 API；若未实现，这里展示设计）
profiles = await mem.retrival_user_profile(user_id=task.user_id or "user", user_input="agent tool usage")
print("User profiles:", profiles)

(5) Mock LLM（若无真实 key）

• 可以自定义一个简单类替换 Agent._call_llm_model 返回固定 JSON：

{
 "content": "",
 "tool_calls":[
   {"id":"tc1","function":{"name":"text_analyzer","arguments":"{\"text\":\"<原任务文本>\"}"}}
 ]
}

{
 "content": "",
 "tool_calls":[
   {"id":"tc1","function":{"name":"text_analyzer","arguments":"{\"text\":\"<原任务文本>\"}"}}
 ]
}

第二轮（工具结果作为 Observation）再返回自然语言总结（无 tool_calls）结束。

✅ 预期一次执行的轨迹（理想）

1) Researcher 第一次输出 → tool_calls 调用 text_analyzer。
2) Tool 返回关键字统计 + 情感。
3) Researcher 接收到工具结果（Observation.is_tool_result=True），第二次 LLM 决策生成中间解释（可能直接结束并把内容交给 Summarizer）。
4) Summarizer 输出最终答复。
5) Memory：

• SystemPrompt / 用户输入 / 工具调用 / 工具结果 / LLM 回复全部写入短期。
• 达到触发阈值（≥2条），进行 UserProfile / AgentExperience 提取（如：skill=information_processing, actions=[…]）。

✅ 可观察点（调试建议）

✅ 出错/边界

✅ 后续可升级

• 增加 ParallelizableAgent（研究+事实核对两支并行）。
• 引入第二工具（web_search mock）。
• 将 AgentExperience 导出为 jsonl 用于微调。
• 加入 Hook（PRE_LLM_CALL）注入检索式提示（RAG）。

✅ README.md 草案核心段落（可放 demo_minimal/README.md）

内容：运行条件、环境变量设置、如何 Mock、如何查看内存与经验抽取结果——此处略写即可后续补充。