🤝 高内聚低耦合：AI Agent 长任务设计的「分家」艺术

一句话理解

概念	通俗解释	好/坏
高耦合	两个东西绑得太紧，动一个，另一个也得跟着动	❌ 坏
低内聚	一个东西干太多不相关的事，乱七八糟	❌ 坏
低耦合	两个东西相对独立，互不影响	✅ 好
高内聚	一个东西只干一类事，专注专业	✅ 好

记住口诀：高内聚 = 专业分工，低耦合 = 互不干扰

🍳 用厨房比喻

❌ 高耦合 + 低内聚（混乱厨房）

你的厨房：

• 菜刀和砧板用胶水粘在一起（高耦合）
• 冰箱里有菜刀、砧板、衣服、螺丝刀（低内聚）

结果：

• 想换把刀？砧板也得换
• 找把刀得翻遍冰箱
• 做饭效率极低

✅ 低耦合 + 高内聚（专业厨房）

你的厨房：

• 菜刀放刀架，砧板放台面，各归各位（低耦合）
• 刀具区只放刀，调料区只放调料（高内聚）

结果：

• 换刀不影响砧板
• 找东西一目了然
• 大厨来了都能直接上手

🏢 用公司组织架构比喻

❌ 混乱公司（高耦合低内聚）

「全能员工」小张：
├─ 接待客户
├─ 写代码
├─ 设计海报
├─ 打扫卫生
├─ 管财务
└─ 还要给老板泡茶

问题：
- 小张请假，公司停摆
- 没人知道小张具体在干啥
- 招不到第二个「小张」

✅ 专业公司（低耦合高内聚）

专业分工：
├─ 销售部：只负责接待客户
├─ 技术部：只负责写代码
├─ 设计部：只负责做设计
├─ 行政部：只负责后勤
└─ 财务部：只管钱

好处：
- 销售请假，技术照样写代码
- 每个部门都有明确职责
- 缺人随时能招，培训成本低

🤖 应用到 AI Agent 长时间任务

场景：让 Agent 写一篇深度研究报告

❌ 错误做法（高耦合 + 低内聚）

# 一个巨型 Agent 干所有事
class BadAgent:
    def do_everything(self, topic):
        # 1. 搜索资料
        results = self.search_web(topic)
        
        # 2. 分析资料（混在一起）
        analysis = ""
        for r in results:
            analysis += self.summarize(r)  # 边搜边总结
            self.save_to_db(r)             # 边搜边存
            self.check_plagiarism(r)       # 边搜边查重
        
        # 3. 写报告（所有逻辑堆在一起）
        report = self.write(analysis)
        
        # 4. 生成图表（硬编码在里面）
        chart = self.generate_chart(report)
        
        # 5. 导出PDF（还是它干）
        pdf = self.export_pdf(report, chart)
        
        return pdf

问题：

• 🔴 搜、析、写、画、导全绑在一起（高耦合）
• 🔴 一个环节出错，整个任务崩溃
• 🔴 无法单独测试「搜索」或「图表生成」
• 🔴 代码又长又乱（低内聚）

✅ 正确做法（低耦合 + 高内聚）

┌─────────────────────────────────────────┐
│           任务调度器 (Orchestrator)        │
│         「项目经理，只负责协调」            │
└─────────────────────────────────────────┘
                    │
    ┌───────────────┼───────────────┐
    ▼               ▼               ▼
┌────────┐    ┌────────┐    ┌────────┐
│ 研究员  │    │ 撰稿人  │    │ 美编   │
│Researcher│   │ Writer │    │ Designer│
│「只搜资料」│   │「只写报告」│   │「只做图」│
└────────┘    └────────┘    └────────┘
    │               │               │
    ▼               ▼               ▼
┌────────┐    ┌────────┐    ┌────────┐
│ 资料库  │    │ 报告库  │    │ 图表库  │
│(中间存储)│   │(中间存储)│   │(中间存储)│
└────────┘    └────────┘    └────────┘

代码结构：

# 每个 Agent 只干一件事（高内聚）

class ResearcherAgent:
    """研究员：只负责搜索和整理资料"""
    def research(self, topic: str) -> ResearchData:
        results = self.search_web(topic)
        return ResearchData(
            sources=results,
            summary=self.summarize(results),
            keywords=self.extract_keywords(results)
        )

class WriterAgent:
    """撰稿人：只负责写报告"""
    def write(self, research_data: ResearchData) -> Draft:
        # 只关心怎么写好，不关心资料从哪来
        outline = self.create_outline(research_data)
        content = self.expand_sections(outline)
        return Draft(content=content, citations=research_data.sources)

class DesignerAgent:
    """美编：只负责做图表"""
    def create_visuals(self, draft: Draft) -> Charts:
        # 只关心可视化，不关心内容从哪来
        data_points = self.extract_data(draft)
        return Charts(
            infographics=self.make_infographics(data_points),
            diagrams=self.make_diagrams(draft.structure)
        )

class PublisherAgent:
    """出版：只负责导出"""
    def publish(self, draft: Draft, charts: Charts) -> Document:
        return self.export_pdf(draft, charts)


# 调度器负责协调（低耦合）
class TaskOrchestrator:
    """项目经理：协调各 Agent，自己不干活"""
    
    def run_long_task(self, topic: str) -> Document:
        # 步骤1：研究
        researcher = ResearcherAgent()
        research_data = researcher.research(topic)
        self.save_checkpoint("research", research_data)
        
        # 步骤2：写作（研究失败可以跳过/重试）
        writer = WriterAgent()
        draft = writer.write(research_data)
        self.save_checkpoint("draft", draft)
        
        # 步骤3：设计（写作失败可以人工介入）
        designer = DesignerAgent()
        charts = designer.create_visuals(draft)
        self.save_checkpoint("charts", charts)
        
        # 步骤4：出版
        publisher = PublisherAgent()
        return publisher.publish(draft, charts)

🎁 好处对比

方面	高耦合低内聚	低耦合高内聚
出错恢复	一步错，全崩	哪步错，重跑哪步
进度保存	不知道存啥	每步都有 checkpoint
并行执行	没法并行	多个 Researcher 同时搜
人工介入	只能从头来	中间结果可人工修改
测试	巨难测试	每个 Agent 单独测
复用	复制粘贴代码	直接 import 用

💡 关键设计原则

1. 中间存储解耦

# ❌ 不好：直接传对象（耦合度高）
writer.write(researcher.research(topic))

# ✅ 好：通过存储解耦
research_data = researcher.research(topic)
save_to_db("research", research_data)  # 存下来

# Writer 从存储读，不依赖 Researcher
research_data = load_from_db("research")
draft = writer.write(research_data)

2. 标准化接口

# 每个 Agent 都遵循相同接口
class Agent(Protocol):
    def run(self, input_data: Input) -> Output: ...
    def can_handle(self, task_type: str) -> bool: ...

3. 失败隔离

def run_with_retry(agent, input_data, max_retries=3):
    for i in range(max_retries):
        try:
            return agent.run(input_data)
        except Exception as e:
            if i == max_retries - 1:
                # 失败也保存中间结果，方便排查
                save_failed_checkpoint(agent.name, input_data, e)
                raise
            sleep(2 ** i)  # 指数退避

🦞 OpenClaw 中的实践

OpenClaw 的 Sessions + Cron + Gateway 架构本身就是低耦合高内聚的例子：

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Gateway   │────▶│   Session   │────▶│    Cron     │
│  (接电话)    │     │ (处理消息)   │     │ (定时任务)   │
└─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │
       └───────────────────┴───────────────────┘
                    通过消息/事件解耦

• Gateway 只负责收消息，不处理业务逻辑（高内聚）
• Session 只处理一次对话，不保存长期状态（高内聚）
• Cron 只负责定时触发，不管具体任务（高内聚）
• 三者通过消息传递，互不影响（低耦合）

🎯 总结

原则	怎么做	效果
高内聚	一个 Agent 只干一类事	代码清晰、易维护
低耦合	Agent 之间通过标准接口通信	灵活组合、易扩展

记住：设计 Agent 长任务时，想象你在开公司——专业的人干专业的事，大家通过「标准流程」协作，而不是一个人包办所有。

有问题？欢迎在 Telegram 群「小创-useclaw-」讨论！