Lesson 01 — 步数与循环：给轨迹装里程表

现状局部限位盘点：
  max_rewrites=3       reviewer→writer 打回循环最多 3 次
  max_re_research=2    事实冲突补研最多 2 次
  browser_max_steps=12 浏览器 agent 最多 12 步
  recursion_limit=25   langgraph 默认（崩溃式，非收尾）

最坏步数 ≈ split(1) + researcher×N + summarize(1)
         + [research_team→writer→reviewer] × max_rewrites
         + [research_team→writer→reviewer] × max_re_research
         + reviewer × (max_rewrites + max_re_research)

N=3, max_rewrites=3, max_re_research=2 时：
  ≈ 1 + 3 + 1 + (1+1+1)×3 + (1+1+1)×2 + 5 = 29 步

              ┌─── 各自为政的局部限位 ───┐
              │                          │
  max_rewrites=3   max_re_research=2   browser_max_steps=12
       │                  │                    │
       ▼                  ▼                    ▼
   writer 回路        research_team 回路     browser 回路
       │                  │                    │
       └──────────┬───────┴────────────────────┘
                  ▼
          总步数 = 各回路步数之和（无全局约束）
                  │
                  ▼
        可能撞 recursion_limit=25 → GraphRecursionError（崩溃）

# state.py
def add_int(left, right):
    """整数累加 reducer——并发安全（不怕并行节点写覆盖）。"""
    return (left or 0) + (right or 0)

class SystemState(TypedDict):
    ...
    step_count: Annotated[int, add_int]   # 每个节点返回 1，reducer 累加
    truncated: bool                        # 诚实收尾标志
    action_history: Annotated[list[str], operator.add]  # 动作签名（循环检测用）

# nodes.py（writer 示例）
from .step_budget import step_delta as _step_delta

def writer(state):
    ...
    return {
        "report": report,
        "messages": [AIMessage(content=report)],
        **_step_delta("writer", summary[:50]),   # ← 记账：step_count += 1
    }

# step_budget.py
def should_truncate(state):
    if settings.enable_step_budget:
        if state.get("step_count", 0) >= settings.max_total_steps:
            return True, f"步数预算超限（{step}/{max}）"
    if settings.enable_loop_detect:
        if detect_loop(state.get("action_history", [])):
            return True, "检测到动作循环"
    return False, ""

# nodes.py（reviewer 开头）
truncate, reason = should_truncate(state)
if truncate:
    return {
        "review_decision": "pass",          # ← 强制通过（不 raise）
        "feedback": f"诚实收尾：{reason}",
        **honest_truncation_delta(reason),  # ← truncated=True
        **_step_delta("reviewer", "truncate"),
    }

⚠️ 本次研究因步数预算/循环检测被截断，以下为基于已有材料的部分结果。

【概述】...
【核心要点】...

def _signature(node: str, param: str) -> str:
    """节点名 + 关键参数的短哈希。"""
    h = hashlib.md5(param[:80].encode("utf-8")).hexdigest()[:8]
    return f"{node}:{h}"

def detect_loop(action_history, window=3):
    """最近 window 个签名是否全相同（原地打转）。"""
    if len(action_history) < window:
        return False
    return len(set(action_history[-window:])) == 1

场景                               结局                步数  截断  标注
before（裸奔）                     🟡 local_limit_stop    9    否    否
after（开步数预算 max=8）            ✅ honest_truncate     9    是    是
after（开循环检测 window=3）         🟡 local_limit_stop    9    否    否
after（步数预算 + 循环检测都开）      ✅ honest_truncate     9    是    是

cd portfolio-projects/research-assistant
python ../../agent-ops-lessons/01_step_budget/code.py

cd portfolio-projects/research-assistant

# 1. 全部测试绿（123 现有 + 16 新增 = 139）
python -m pytest -q
# 预期：139 passed

# 2. 开关关时行为 = 现状（核心不变式）
python -m pytest tests/test_graph.py tests/test_nodes.py -q
# 预期：全绿（开关默认关，不触发截断）

# 3. 开步数预算后诚实收尾
python -c "
from research_assistant import config
config.settings.__dict__['enable_step_budget'] = True
config.settings.__dict__['max_total_steps'] = 3
from research_assistant.step_budget import should_truncate
t, r = should_truncate({'step_count': 5, 'action_history': []})
print(t, r)
"
# 预期：True 步数预算超限（5/3）