Unix 哲学在 AI 时代的回响：Agent 设计哲学

# 反 Unix 的设计：一个 Agent 做所有事
class SuperAgent:
    def execute(self, task):
        # 规划
        # 记忆管理
        # 工具调用
        # 代码生成
        # 自我反思
        # ... 所有逻辑耦合在一起

# 每个 Agent 用自己的数据格式
agent_a_output = {
    "result": {...},
    "metadata": {...},
    "_internal_state": {...}  # 只有 Agent A 能解析
}
# Agent B 无法理解，需要转换器

# 层层嵌套，难以追踪
result = agent_a.run(
    agent_b.process(
        agent_c.analyze(
            agent_d.fetch(query)
        )
    )
)

# Unix 风格的设计：每个 Agent 专注单一职责

class PlannerAgent:
    """只做一件事：将任务分解为步骤"""
    
    def execute(self, task: str) -> Plan:
        """返回执行计划，不执行任何操作"""
        return self.llm.generate(
            prompt=f"将以下任务分解为具体步骤：{task}",
            output_schema=PlanSchema
        )

class ExecutorAgent:
    """只做一件事：执行具体步骤"""
    
    def execute(self, step: Step, tools: ToolRegistry) -> Result:
        """执行单个步骤，不处理规划"""
        tool = tools.get(step.tool_name)
        return tool.run(step.parameters)

class MemoryAgent:
    """只做一件事：管理记忆存储和检索"""
    
    def store(self, key: str, value: Any, ttl: Optional[int] = None):
        """存储记忆"""
        pass
    
    def retrieve(self, query: str, limit: int = 5) -> List[Memory]:
        """检索相关记忆"""
        pass

class ReflectionAgent:
    """只做一件事：分析执行过程并提取经验"""
    
    def execute(self, execution_log: ExecutionLog) -> Lessons:
        """从执行日志中提取经验教训"""
        pass

# 定义 Agent 之间的通用消息格式（类似 Unix 的文本流）

@dataclass
class AgentMessage:
    """Agent 之间的标准消息格式"""
    role: Literal["user", "assistant", "system", "tool"]
    content: Union[str, Dict]  # 文本或结构化数据
    metadata: MessageMetadata  # 时间戳、来源、追踪ID等
    context: Optional[Context] = None  # 执行上下文

@dataclass
class AgentOutput:
    """Agent 的标准输出格式"""
    success: bool
    result: Any
    logs: List[LogEntry]  # 详细的执行日志（透明性）
    next_actions: Optional[List[Action]] = None

# Unix 风格：| 管道操作符
result = task | planner | executor | validator

# 实际实现
output_a = planner_agent.execute(task)
output_b = executor_agent.execute(output_a.result)
output_c = validator_agent.execute(output_b.result)

class TransparentAgent:
    """透明的 Agent：每一步都可见"""
    
    def execute(self, task: str, verbose: bool = True) -> Result:
        trace = ExecutionTrace()
        
        if verbose:
            trace.log(f"🎯 接收任务: {task}")
        
        # 规划阶段
        plan = self.planner.create_plan(task)
        if verbose:
            trace.log(f"📋 生成计划: {len(plan.steps)} 个步骤")
            for i, step in enumerate(plan.steps, 1):
                trace.log(f"   {i}. {step.description}")
        
        # 执行阶段
        results = []
        for step in plan.steps:
            if verbose:
                trace.log(f"⚡ 执行: {step.description}")
            
            result = self.executor.run(step)
            results.append(result)
            
            if verbose:
                trace.log(f"   结果: {result.summary}")
                if result.tool_calls:
                    for call in result.tool_calls:
                        trace.log(f"   调用: {call.tool}({call.params})")
        
        # 验证阶段
        validation = self.validator.check(results)
        if verbose:
            trace.log(f"✅ 验证结果: {validation.status}")
            if validation.issues:
                for issue in validation.issues:
                    trace.log(f"   ⚠️  {issue}")
        
        return Result(
            success=validation.success,
            output=self.synthesize(results),
            trace=trace  # 完整的执行轨迹
        )

class QuietAgent:
    """Unix 风格的安静 Agent"""
    
    def execute(self, task: str, verbose: bool = False) -> Result:
        try:
            result = self._do_work(task)
            if verbose:
                return Result(success=True, output=result, message="完成")
            return Result(success=True, output=result)  # 静默成功
        except Exception as e:
            # 失败时必须提供清晰的错误信息
            return Result(
                success=False, 
                error=f"{self.__class__.__name__}: {str(e)}",
                suggestion=self._get_suggestion(e)
            )

class AgentPipeline:
    """Agent 管道：Unix 管道的现代版"""
    
    def __init__(self):
        self.agents: List[Agent] = []
    
    def pipe(self, agent: Agent) -> "AgentPipeline":
        """添加 Agent 到管道（类似 Unix 的 |）"""
        self.agents.append(agent)
        return self
    
    def execute(self, input_data: Any) -> Result:
        """顺序执行管道中的所有 Agent"""
        result = input_data
        for agent in self.agents:
            result = agent.execute(result)
            if not result.success:
                return result  # 短路失败
        return Result(success=True, output=result)

# 使用示例：类 Unix 语法
pipeline = (
    AgentPipeline()
    .pipe(PlannerAgent())      # 规划
    .pipe(BreakdownAgent())    # 细化
    .pipe(ExecutorAgent())     # 执行
    .pipe(ValidatorAgent())    # 验证
)

result = pipeline.execute("重构用户认证模块")

class ConditionalOrchestrator:
    """条件编排：根据结果选择路径"""
    
    def __init__(self):
        self.branches: Dict[str, Agent] = {}
    
    def when(self, condition: str, agent: Agent):
        self.branches[condition] = agent
        return self
    
    def execute(self, input_data: Any) -> Result:
        # 使用 Router Agent 决定路径
        route = self.router.decide(input_data)
        agent = self.branches.get(route, self.default_agent)
        return agent.execute(input_data)

# 使用
orchestrator = (
    ConditionalOrchestrator()
    .when("technical", TechnicalAgent())
    .when("business", BusinessAgent())
    .when("creative", CreativeAgent())
)

class ParallelOrchestrator:
    """并行编排：同时执行多个 Agent"""
    
    def execute(self, items: List[Any], agent: Agent, max_workers: int = 4) -> List[Result]:
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(agent.execute, item) for item in items]
            return [f.result() for f in futures]

# 使用：并行处理 100 个任务
results = ParallelOrchestrator().execute(tasks, MyAgent(), max_workers=10)

class ReduceOrchestrator:
    """Reduce 编排：分而治之，然后聚合"""
    
    def execute(
        self, 
        items: List[Any], 
        map_agent: Agent, 
        reduce_agent: Agent
    ) -> Result:
        # Map：并行处理
        map_results = ParallelOrchestrator().execute(items, map_agent)
        
        # Reduce：聚合结果
        return reduce_agent.execute(map_results)

# 使用：Map-Reduce 风格分析
result = ReduceOrchestrator().execute(
    items=documents,
    map_agent=SummarizeAgent(),      # 每个文档独立总结
    reduce_agent=MergeSummaryAgent()  # 合并所有总结
)

# 构建一个类 Unix 的代码审查管道

code_review_pipeline = (
    AgentPipeline()
    # Step 1: 提取变更（类似 git diff）
    .pipe(DiffExtractorAgent())
    
    # Step 2: 语法分析（类似 linter）
    .pipe(SyntaxAnalyzerAgent())
    
    # Step 3: 安全扫描（类似 grep 敏感模式）
    .pipe(SecurityScannerAgent())
    
    # Step 4: 风格检查（类似 style checker）
    .pipe(StyleCheckerAgent())
    
    # Step 5: 逻辑审查（复杂分析）
    .pipe(LogicReviewerAgent())
    
    # Step 6: 生成报告（类似格式化输出）
    .pipe(ReportGeneratorAgent())
)

# 执行
result = code_review_pipeline.execute({
    "pr_number": 1234,
    "repo": "myorg/myrepo"
})

# 输出类似：
# === Code Review Report ===
# Files changed: 5
# Issues found: 3
# - [SECURITY] Hardcoded API key in config.py:42
# - [STYLE] Function too long (150 lines) in utils.py:89
# - [LOGIC] Potential race condition in cache.py:23

# 定义 Intent 标准格式（类似 Unix 的文本流，但是结构化的）

@dataclass
class Intent:
    """标准化的 Intent 格式"""
    action: str                    # 动作类型
    target: str                    # 目标对象
    parameters: Dict[str, Any]     # 参数
    constraints: List[Constraint]  # 约束条件
    context: Optional[Context]     # 上下文

@dataclass
class IntentResult:
    """标准化的结果格式"""
    status: Literal["success", "failure", "partial"]
    data: Any
    logs: List[LogEntry]
    next_intents: Optional[List[Intent]] = None

# Intent 驱动的 Agent 管道

class IntentRouter:
    """Intent 路由器：根据 Intent 分发到不同 Agent"""
    
    def __init__(self):
        self.handlers: Dict[str, Agent] = {}
    
    def register(self, action: str, agent: Agent):
        self.handlers[action] = agent
    
    def route(self, intent: Intent) -> Result:
        agent = self.handlers.get(intent.action)
        if not agent:
            return Result(
                success=False, 
                error=f"No handler for action: {intent.action}"
            )
        return agent.execute(intent)

# 使用示例
router = IntentRouter()
router.register("analyze", AnalysisAgent())
router.register("transform", TransformAgent())
router.register("validate", ValidationAgent())

# 构建 Intent 链
intents = [
    Intent(action="analyze", target="data.csv", parameters={"type": "schema"}),
    Intent(action="transform", target="data.csv", parameters={"clean": True}),
    Intent(action="validate", target="data.csv", parameters={"rules": "strict"}),
]

# 顺序执行
for intent in intents:
    result = router.route(intent)
    if not result.success:
        break

class HybridAgent:
    """混合接口：支持自然语言和结构化 Intent"""
    
    def execute(self, input: Union[str, Intent]) -> Result:
        if isinstance(input, str):
            # 自然语言 → Intent
            intent = self.nlu.parse(input)
        else:
            intent = input
        
        # 执行 Intent
        return self._execute_intent(intent)
    
    def _execute_intent(self, intent: Intent) -> Result:
        # 具体实现
        pass

# 两种使用方式等价
agent.execute("分析用户数据并生成报告")  # 自然语言
agent.execute(Intent(                      # 结构化 Intent
    action="analyze",
    target="user_data",
    parameters={"output": "report"}
))

# 好的设计：显式声明 Agent 的能力
class CodeReviewAgent:
    CAPABILITIES = [
        "syntax_check",
        "style_check", 
        "security_scan"
    ]
    
    LIMITATIONS = [
        "cannot_execute_code",
        "no_access_to_external_apis"
    ]

# 坏的设计：隐式的"全能"假设
class MagicAgent:
    # 没有声明能力，用户不知道它能做什么
    pass

class FailingFastAgent:
    """快速失败的 Agent"""
    
    def execute(self, task: str) -> Result:
        # 前置条件检查
        if not self._can_handle(task):
            return Result(
                success=False,
                error=f"Cannot handle task: {task}. "
                      f"My capabilities: {self.CAPABILITIES}",
                suggestion="Try using XXXAgent instead"
            )
        
        # 执行中遇到错误立即退出
        for step in self.plan(task):
            result = step.execute()
            if not result.success:
                return Result(
                    success=False,
                    error=f"Failed at step: {step.name}",
                    context={"failed_step": step, "partial_result": result}
                )
        
        return Result(success=True, output=result)

# 标准 Agent 接口
class Agent(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        """Agent 名称（类似 Unix 命令名）"""
        pass
    
    @property
    @abstractmethod
    def capabilities(self) -> List[str]:
        """声明的能力清单"""
        pass
    
    @abstractmethod
    def can_handle(self, intent: Intent) -> bool:
        """是否能处理此 Intent"""
        pass
    
    @abstractmethod
    def execute(self, intent: Intent) -> Result:
        """执行并返回结果"""
        pass

# 每个 Agent 独立测试（类似 Unix 的单元测试）

class TestPlannerAgent:
    def test_simple_task(self):
        agent = PlannerAgent()
        result = agent.execute(Intent(action="plan", target="simple task"))
        assert result.success
        assert len(result.output.steps) > 0
    
    def test_unsupported_task(self):
        agent = PlannerAgent()
        result = agent.execute(Intent(action="unsupported_action"))
        assert not result.success  # 快速失败
        assert "cannot handle" in result.error.lower()

# 管道集成测试
class TestPipeline:
    def test_full_pipeline(self):
        pipeline = AgentPipeline().pipe(A()).pipe(B()).pipe(C())
        result = pipeline.execute(test_input)
        assert result.success