Reborn 的技术博客

核心增量：chat_loop()（~120 行）+ PersistenceManager.list_sessions() + SkillManager.reset()

一、问题：单次问答够用吗？

前两篇文章构建的 Agent 有 ReAct 循环（思考→行动→观察）和渐进式 Skill 加载，但每次问答都是一次性的：

用户: "北京天气怎么样？"
Agent: "北京今天晴，25°C"
                              ← 上下文丢失
用户: "那上海呢？"
Agent: "你说的是哪个上海？"     ← 不记得刚才在讨论天气

二、核心结构：chat_loop 三条线

chat_loop()
  ├── 交互线：input() 循环 + 输出
  ├── 命令线：/exit /history /switch /new /help
  └── 状态线：tracer + ctx + pm + skills

三条线的交点是 session_id——所有状态的"身份证"。

三、主循环

def chat_loop(client, skills, base_system_prompt, model, session_id=None, store_dir="./agent_sessions"):
    tracer = AgentTracer()
    ctx = ContextManager()
    pm = PersistenceManager(Store(store_dir))

    if session_id is None:
        session_id = pm.new_session_id()

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n👋 再见")
            break

        if not user_input:
            continue

        if user_input == "/exit":
            break
        elif user_input == "/history":
            _show_history(pm, session_id)
        elif user_input.startswith("/switch"):
            _switch_session(user_input, pm, session_id, tracer, ctx, skills)
        elif user_input == "/new":
            session_id, tracer, ctx = _new_session(pm, skills)
        elif user_input == "/help":
            _show_help()
        else:
            answer = run_agent_with_trace(
                user_input, tracer=tracer, client=client, ctx=ctx, pm=pm,
                skills=skills, base_system_prompt=base_system_prompt,
                session_id=session_id, model=model,
            )
            print(f"Agent: {answer}\n")

设计要点

| 要点 | 说明 | | --- | --- | | session_id 贯穿所有调用 | 对话历史自动累积 | | 命令短路径 | /exit、/help 直接 return，零 token 消耗 | | 异常安全 | EOFError/KeyboardInterrupt 捕获 | | 空输入跳过 | 防止空白行触发 LLM 调用 |

四、会话管理三件套

新建会话

elif user_input == "/new":
    session_id = pm.new_session_id()   # 新 ID
    tracer = AgentTracer()             # 清空 trace
    ctx = ContextManager()             # 清空上下文摘要
    skills.reset()                     # 卸载所有激活的 Skill

历史会话列表

elif user_input == "/history":
    sessions = pm.list_sessions()
    for s in sessions:
        marker = " ← 当前" if s["id"] == session_id else ""
        print(f"  {s['id']} | {s['message_count']}条 | {s['updated']} | {s['last_message']}{marker}")

会话切换

elif user_input.startswith("/switch"):
    new_id = user_input.split(" ", 1)[1].strip()
    state = pm.load_session(new_id)
    if not state["messages"]:
        print(f"❌ 会话 {new_id} 不存在或为空")
        continue

    session_id = new_id
    tracer = AgentTracer()
    ctx = ContextManager()
    ctx.restore(state["summary"])
    skills.reset()

五、设计精要

命令是"带外信道"

带内信道：自然语言 → LLM → ReAct 循环 → 工具调用
带外信道：/ 前缀命令 → chat_loop 直接处理

好处：命令不会被 LLM 误解析；执行零延迟；可扩展 /export、/settings、/retry。

session_id 是胶水

首次调用创建会话 → 后续调用加载历史 + 追加消息 → 对话越来越长（ctx.maybe_compact 自动压缩）。