循环、图与骨架工程

死法一：打转（oscillation）
  search[天气] → 结果不满意 → search[天气] → ……   ← 第 8 章的重复检测

死法二：发散（drift）
  任务：写周报 → 中途开始"优化模板" → 又开始"调研工具" → 忘了初衷
  每步都"合理"，整体离题万里——目标漂移

死法三：滚雪球（cascade）
  轮次↑ → 上下文↑ → 注意力稀释 → 判断质量↓ → 更多无效调用 → 轮次↑
  正反馈循环，越跑越糟

# 软闸门：先警告，给模型自纠的机会
if self.tokens_used > self.budget * 0.7:
    if not self._warned:
        self.messages.append({"role": "user",
            "content": "提示：token 预算已使用 70%。请尽快总结当前成果并收尾。"})
        self._warned = True

# 硬闸门：无情切断
if self.tokens_used > self.budget:
    return self._partial_result()   # 带着部分成果退出，而不是空手

def _partial_result(self) -> str:
    resp = client.chat.completions.create(
        model=self.model,
        messages=self.messages + [{
            "role": "user",
            "content": "预算已用尽。请基于已收集到的信息，"
                       "给出当前最佳的不完整答案，并注明局限。",
        }],
    )
    return resp.choices[0].message.content

def _goal_check(self, original_task: str) -> bool:
    """每 N 轮做一次目标对齐检查"""
    resp = client.chat.completions.create(
        model=self.model, temperature=0,
        messages=[
            {"role": "system",
             "content": "判断当前进度是否仍在服务原始任务。"
                        "只输出 ALIGNED 或 DRIFTED。"},
            {"role": "user",
             "content": f"原始任务：{original_task}\n\n"
                        f"最近的行动：{self.recent_actions_summary()}"},
        ],
    )
    return resp.choices[0].message.content.strip() == "ALIGNED"

# ① 结果型纠偏：工具返回了错误信息（第 7 章的既有模式）
#    模型看到 "参数校验失败：..." 通常下一轮自己修正

# ② 提示型纠偏：框架检测到模式问题，注入 user 消息
def _inject_correction(self, issue: str, guidance: str):
    self.messages.append({
        "role": "user",
        "content": f"[系统提示] 检测到问题：{issue}。{guidance}",
    })
# 用例：重复调用后 → "你已连续3次做相同调用，请更换策略"
#      目标漂移后 → "当前行动偏离了原始任务，请回到：{task}"

# ③ 结构型纠偏：重置部分状态（最重）
def _structural_reset(self):
    """保留任务和关键结论，丢弃失败过程"""
    self.messages = [
        self.messages[0],                      # system
        {"role": "user", "content": self.original_task},
        {"role": "assistant",
         "content": f"已完成部分：{self.achieved_summary}。"
                    f"失败尝试：{self.failed_summary}（避免重复）。"},
    ]

PHASES = ["analyze", "execute", "verify"]

def run_phased(self, task: str):
    context = {"task": task, "findings": []}
    for phase in PHASES:
        phase_prompt = self.phase_prompts[phase].format(**context)
        self.messages.append({"role": "user", "content": phase_prompt})
        # 阶段内跑第 8 章的标准循环（限更小轮次）
        result = self._bounded_loop(max_rounds=3)
        context["findings"].append(result)
        # 阶段间质量控制：verify 不过可以折返
        if phase == "verify" and self._failed(result):
            return self.run_phased(task)   # 或定向折返 execute
    return context["findings"][-1]

result = analyze(task)
result = execute(result)
if not verified(result):
    result = execute(result)      # 折返重试
return report(result)

        ┌──────────┐
        │ analyze  │
        └────┬─────┘
             ▼
        ┌──────────┐
   ┌───▶│ execute  │
   │    └────┬─────┘
   │         ▼
   │  ┌──────────┐   fail
   └──│ verify   │────────┐
      └────┬─────┘        ▼
           │ pass    ┌────────┐
           ▼         │ retry  │──┐
      ┌────────┐    └────────┘  │(计数)
      │ report │       ……
      └────────┘

from dataclasses import dataclass, field

@dataclass
class Node:
    name: str
    execute: callable                    # 执行函数：state → state
    kind: str = "agent"                  # agent / tool / human / logic

from typing import TypedDict

class WorkflowState(TypedDict):
    task: str                    # 原始任务
    findings: list[str]          # 各节点产出的累积
    attempt_count: int           # 重试计数
    verified: bool               # 验证结论
    current_node: str            # 执行位置

@dataclass
class Edge:
    source: str
    target: str
    condition: callable = None    # 条件边：state → bool

from typing import Callable

class Graph:
    def __init__(self):
        self.nodes: dict[str, Callable] = {}
        self.edges: list[Edge] = []

    def add_node(self, name: str, fn: Callable):
        self.nodes[name] = fn

    def add_edge(self, source: str, target: str,
                 condition: Callable | None = None):
        self.edges.append(Edge(source, target, condition))

    def run(self, initial_state: dict, start: str,
            max_steps: int = 20) -> dict:
        state = initial_state
        current = start
        for step in range(max_steps):              # 第一部分的保险丝
            state["current_node"] = current
            state = self.nodes[current](state)     # 执行节点

            # 找下一个节点：优先匹配条件边
            next_node = None
            for e in self.edges:
                if e.source != current:
                    continue
                if e.condition is None or e.condition(state):
                    next_node = e.target
                    break
            if next_node is None:                  # 没有出边 = 自然终点
                return state
            current = next_node
        raise RuntimeError("图执行超过步数上限（疑似环死循环）")

g = Graph()

def analyze(state):
    state["findings"].append(f"分析结果: {state['task'][:20]}...")
    return state

def execute(state):
    state["findings"].append("执行结果: ...")
    return state

def verify(state):
    state["verified"] = len(state["findings"]) >= 2    # 演示用判断
    return state

def report(state):
    state["final"] = "\n".join(state["findings"])
    return state

g.add_node("analyze", analyze)
g.add_node("execute", execute)
g.add_node("verify", verify)
g.add_node("report", report)

g.add_edge("analyze", "execute")
g.add_edge("execute", "verify")
g.add_edge("verify", "report", condition=lambda s: s["verified"])
g.add_edge("verify", "execute",
           condition=lambda s: not s["verified"] and s["attempt_count"] < 3)

result = g.run({"task": "写周报", "findings": [],
                "attempt_count": 0}, start="analyze")
print(result["final"])

        ┌→ research_A ─┐
analyze ┤              ├→ synthesize
        └→ research_B ─┘

draft → [human_review] → publish
              │ reject
              ▼
          revise → draft

纯代码流程 ── 固定图 ── 混合图（部分边动态）── 路由循环 ── 纯自由循环
（n8n 工作流）  （审批流）      （客服分流+处理）      （第8章ReAct）

┌─────────────────────────────────────────┐
│              Agent Harness                │
│                                           │
│  ┌─────────┐  模型接口层                  │
│  │ 模型调用 │  超时、重试、降级、备用模型    │
│  └─────────┘                              │
│  ┌─────────┐  工具执行层                   │
│  │ 工具系统 │  权限、沙箱、超时、结果截断    │  ← 第 7/11/15 章
│  └─────────┘                              │
│  ┌─────────┐  循环控制层                   │
│  │ 主循环   │  轮次、预算、卡死检测         │  ← 第 8 章
│  └─────────┘                              │
│  ┌─────────┐  上下文层                     │
│  │ 上下文   │  组装、裁剪、预算分配           │  ← 第 13 章
│  └─────────┘                              │
│  ┌─────────┐  反馈层                       │
│  │ 校验重试 │  输出校验、失败重试、自纠错     │  ← 第 6 章
│  └─────────┘                              │
│  ┌─────────┐  观测层                       │
│  │ 事件审计 │  轨迹、日志、熔断              │  ← 第 8/15 章
│  └─────────┘                              │
└─────────────────────────────────────────┘

def run_with_recovery(self, task: str, max_retries: int = 2) -> str:
    for attempt in range(max_retries + 1):
        try:
            result = self.run(task)
            if not self._looks_failed(result):        # 启发式检查结果质量
                return result
            failure_reason = "结果质量可疑"
        except Exception as e:
            failure_reason = str(e)

        if attempt < max_retries:
            # 带着教训重试：失败原因进入下一轮的系统提示词
            self.messages.append({
                "role": "user",
                "content": f"上次尝试失败（{failure_reason}）。"
                           f"请换一种策略重新完成任务。",
            })
            self._cleanup_partial_work()              # 回滚半成品副作用
    return self._fallback_answer(task)                 # 优雅降级

def checkpoint(self, task_id: str):
    """把当前状态序列化——messages + 关键中间产物"""
    state = {"messages": self.messages,
             "round": self.current_round}
    Path(f".checkpoints/{task_id}.json").write_text(
        json.dumps(state, ensure_ascii=False), encoding="utf-8")

def resume(self, task_id: str) -> bool:
    """从最近的存档点恢复"""
    p = Path(f".checkpoints/{task_id}.json")
    if not p.exists():
        return False
    state = json.loads(p.read_text(encoding="utf-8"))
    self.messages = state["messages"]
    self.current_round = state["round"]
    return True

┌────────────────────────────────────────────────┐
│                  Agent 系统                      │
│                                                  │
│   Prompt Eng.    怎么把一件事说清楚（13 章）     │
│        ↓ 每条指令进入                             │
│   Context Eng.   窗口里装什么（13 章）           │
│        ↓ 每轮组装成                               │
│   Harness Eng.   系统骨架怎么搭（本章）           │
│        ↓ 骨架的运行时是                           │
│   Loop Eng.      循环怎么跑怎么停（本章）          │
│        ↓ 复杂流程结构化为                         │
│   Graph Eng.     流程怎么编排（本章）              │
└────────────────────────────────────────────────┘