1. 引言

2025年，AI Agent 已从实验室概念走向生产级应用。然而，当开发者真正着手构建一个 Agent 系统时，最先面临的不是模型选型，而是一个更底层的问题：Agent 的“骨架”该怎么搭？

这个“骨架”就是 Agent Harness——它定义了 Agent 如何感知环境、如何调用工具、如何管理记忆、如何编排任务。不同的 Harness 设计哲学，决定了 Agent 的能力边界、扩展成本和故障模式。

本文将深度解构四种主流的 Agent Harness 设计哲学，从架构原理到代码实现，帮你找到最适合自己业务场景的“骨架”。

2. 什么是 Agent Harness？

Agent Harness 是 Agent 系统的运行时框架，它承担以下核心职责：

• 感知层：接收用户输入、环境状态、系统事件
• 推理层：调用大模型进行决策和规划
• 行动层：执行工具调用、API 请求、代码运行
• 记忆层：管理短期上下文和长期知识
• 编排层：控制任务流程、错误恢复、并发调度

如果把 Agent 比作一个机器人，Harness 就是它的骨架和神经系统——决定了它能做什么动作、动作的协调程度、以及出故障时能否自我修复。

3. 四种 Harness 设计哲学

3.1 循环式（Loop-based）Harness

核心思想：Agent 在一个 while 循环中反复执行“观察→思考→行动”三步，直到任务完成或达到终止条件。

架构特点：

• 简单直观，适合单轮对话式任务
• 状态管理在循环体内显式维护
• 工具调用结果直接喂回下一轮推理

代码示例：

class LoopHarness:
    def __init__(self, llm, tools, max_steps=10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps
        self.messages = []

    async def run(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        for step in range(self.max_steps):
            response = await self.llm.generate(self.messages)
            action = self._parse_action(response)
            if action["type"] == "final_answer":
                return action["content"]
            result = await self.tools[action["name"]].call(**action["args"])
            self.messages.append({"role": "tool", "content": result})
        return "Max steps reached."

适用场景：简单问答、单工具调用、原型验证。

局限性：缺乏并行能力，错误恢复机制薄弱，不适合复杂多步骤任务。

3.2 图式（Graph-based）Harness

核心思想：将 Agent 的行为建模为有向图，节点是状态或操作，边是条件转移。Agent 的执行就是在这个图上行走。

架构特点：

• 显式定义状态机和转移条件
• 支持分支、循环、并行子图
• 每个节点可独立配置工具和模型

代码示例（基于 LangGraph 风格）：

from typing import TypedDict, Literal
import json

class AgentState(TypedDict):
    messages: list
    next_node: str

class GraphHarness:
    def __init__(self):
        self.nodes = {}
        self.edges = {}

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

    def add_edge(self, from_node: str, to_node: str, condition=None):
        self.edges.setdefault(from_node, []).append((to_node, condition))

    async def run(self, state: AgentState):
        current = "start"
        while current != "__end__":
            handler = self.nodes[current]
            state = await handler(state)
            for next_node, condition in self.edges.get(current, []):
                if condition is None or condition(state):
                    current = next_node
                    break
            else:
                current = "__end__"
        return state

适用场景：复杂业务流程、多步骤推理、需要显式状态管理的生产系统。

优势：可观测性强，错误恢复路径清晰，适合团队协作维护。

3.3 事件驱动式（Event-driven）Harness

核心思想：Agent 系统由事件总线驱动，各模块通过发布/订阅模式解耦。工具调用、模型响应、外部消息都作为事件流转。

架构特点：

• 模块间松耦合，易于扩展
• 天然支持异步和并发
• 事件溯源便于调试和回放

代码示例：

import asyncio
from collections import defaultdict

class EventBus:
    def __init__(self):
        self.handlers = defaultdict(list)

    def subscribe(self, event_type: str, handler):
        self.handlers[event_type].append(handler)

    async def emit(self, event_type: str, data: dict):
        for handler in self.handlers[event_type]:
            await handler(data)

class EventDrivenHarness:
    def __init__(self):
        self.bus = EventBus()
        self._setup_handlers()

    def _setup_handlers(self):
        self.bus.subscribe("user_input", self.on_user_input)
        self.bus.subscribe("llm_response", self.on_llm_response)
        self.bus.subscribe("tool_result", self.on_tool_result)

    async def on_user_input(self, data):
        # 触发 LLM 调用
        await self.bus.emit("llm_request", {"messages": data["messages"]})

    async def run(self, user_input: str):
        await self.bus.emit("user_input", {"messages": [{"role": "user", "content": user_input}]})
        await asyncio.Event().wait()  # 等待完成信号

适用场景：高并发场景、微服务架构、需要实时响应的系统。

优势：弹性好，组件可独立部署和升级。

3.4 管道式（Pipeline-based）Harness

核心思想：将 Agent 的处理流程拆解为一系列顺序或并行的处理阶段（Stage），每个阶段负责一个明确的子任务，数据在管道中流动。

架构特点：

• 每个 Stage 是纯函数或可调用对象
• 支持 Stage 的组合、复用和测试
• 数据流清晰，便于监控和调优

代码示例：

from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class Stage:
    name: str
    fn: Callable[[dict], dict]

class PipelineHarness:
    def __init__(self, stages: list[Stage]):
        self.stages = stages

    async def run(self, initial_input: dict) -> dict:
        context = initial_input
        for stage in self.stages:
            context = await stage.fn(context)
            if context.get("abort"):
                break
        return context

# 使用示例
pipeline = PipelineHarness([
    Stage("input_validation", validate_input),
    Stage("intent_classification", classify_intent),
    Stage("tool_selection", select_tools),
    Stage("execution", execute_tools),
    Stage("response_generation", generate_response),
])

适用场景：数据处理流水线、批处理任务、可预测的固定流程。

优势：可测试性强，每个 Stage 可独立优化和替换。

4. 四种哲学的对比分析

维度	循环式	图式	事件驱动	管道式
复杂度	低	中高	高	中
可扩展性	低	高	极高	中
可观测性	中	高	高	高
并发支持	无	有限	原生	有限
错误恢复	弱	强	中	中
学习曲线	低	中	高	低
典型框架	自实现	LangGraph	CrewAI	Haystack

5. 如何选择适合你的 Harness？

5.1 按场景选择

• 快速原型 / 简单问答 → 循环式
• 复杂业务流程 / 多步骤推理 → 图式
• 高并发 / 实时系统 → 事件驱动
• 固定流水线 / 数据处理 → 管道式

5.2 按团队能力选择

• 小团队快速迭代 → 循环式或管道式
• 中大型团队协作 → 图式（状态图便于沟通）
• 基础设施团队 → 事件驱动（适合平台化）

5.3 混合架构

实际生产系统中，往往不是非此即彼。一个典型的混合方案是：

• 外层用事件驱动架构处理请求路由和并发
• 内层用图式编排复杂业务逻辑
• 每个节点内部用管道式处理数据变换
• 简单工具调用用循环式快速完成

6. 实战：从零搭建一个混合 Harness

下面我们实现一个融合了图式编排和管道式处理的混合 Harness：

class HybridHarness:
    def __init__(self):
        self.graph = GraphHarness()
        self._build_graph()

    def _build_graph(self):
        self.graph.add_node("start", self._start_node)
        self.graph.add_node("analyze", self._analyze_node)
        self.graph.add_node("execute", self._execute_node)
        self.graph.add_node("respond", self._respond_node)

        self.graph.add_edge("start", "analyze")
        self.graph.add_edge("analyze", "execute", condition=lambda s: s.get("needs_tool"))
        self.graph.add_edge("analyze", "respond", condition=lambda s: not s.get("needs_tool"))
        self.graph.add_edge("execute", "analyze")  # 循环直到完成
        self.graph.add_edge("respond", "__end__")

    async def _analyze_node(self, state):
        # 内部使用管道式处理
        pipeline = PipelineHarness([
            Stage("extract_intent", extract_intent),
            Stage("check_tools", check_tool_availability),
        ])
        return await pipeline.run(state)

    async def run(self, user_input: str):
        return await self.graph.run({"messages": [{"role": "user", "content": user_input}]})

7. 总结与展望

Agent Harness 的设计哲学没有银弹。循环式简单直接，图式严谨可控，事件驱动灵活弹性，管道式清晰可测——每种哲学都在“灵活性”和“确定性”之间做出了不同的取舍。

展望未来，我们可能会看到：

1. 自适应 Harness：根据任务复杂度动态切换编排模式
2. 多 Agent Harness：支持 Agent 之间的协作和竞争
3. 可微分 Harness：将编排策略本身纳入优化目标

选择 Harness 时，记住一句话：不要为未来可能不需要的灵活性，牺牲当下必须的确定性。

你的下一个 Agent 系统，选好骨架了吗？