harnessloop 0.2.1

Lightweight zero-dependency Agent orchestration framework: Tool Registry, Pipeline/ReAct Agent, Multi-Agent Orchestrator, Pluggable Verification, and Loop Engineering

agent-harness

轻量级零依赖 Agent 编排框架
Model + Harness = Agent

---

概述

agent-harness 是一个零外部依赖的 Agent 编排框架，提供从 Tool 注册、Agent 执行循环、多 Agent 编排到验证规则引擎的完整控制流。

核心思想：Agent 循环的本质是 决策→执行→观察→验证→收敛 这套骨架，Pipeline、ReAct、多 Agent 协作都只是在骨架上的不同变体。

架构

agent-harness/
├── harness/
│   ├── tool_registry.py       # Tool / ToolResult / ToolRegistry
│   ├── agent.py               # Agent（Pipeline）+ ReActAgent（LLM 动态推理）
│   ├── orchestrator.py        # Orchestrator（约束→执行→验证→纠错→收敛 + HITL）
│   ├── verification.py        # RuleEngine（可插拔验证 + 3 条内置规则）
│   └── loop/                  # Loop Engineering 生产级加固
│       ├── termination.py     #   Terminator：6 种可组合终止条件
│       ├── smart_retry.py     #   SmartRetry：错误分类 + 指数退避 + 死信队列
│       ├── loop_state.py      #   LoopState：结构化循环状态追踪
│       ├── hitl.py            #   HITLManager：异步人机协作
│       └── plan_agent.py      #   PlanAgent：先规划再执行
└── setup.py

---

安装

pip install -e /path/to/agent-harness
# 或
pip install agent-harness

---

快速开始

1. 定义 Tool

from harness import Tool, ToolResult

class GreetingTool(Tool):
    name = "greet"
    description = "Say hello to someone"
    input_schema = {"name": "str"}

    def execute(self, name: str = "World", **kwargs) -> ToolResult:
        return ToolResult(success=True, data={"greeting": f"Hello, {name}!"})

class WeatherTool(Tool):
    name = "weather"
    description = "Get weather for a city"
    input_schema = {"city": "str"}

    def execute(self, city: str = "Beijing", **kwargs) -> ToolResult:
        return ToolResult(success=True, data={"city": city, "temp": "22°C"})

2. Pipeline 模式（固定管线）

from harness import Agent, Orchestrator

agent = Agent(
    name="greeter",
    tools=[GreetingTool(), WeatherTool()],
    system_prompt="You are a helpful assistant.",
)

orchestrator = Orchestrator(agents=[agent], max_loops=10)
result = orchestrator.run(
    task="Greet Alice and check Beijing weather",
    context={"name": "Alice", "city": "Beijing"},
)
print(result.to_dict())

3. ReAct 模式（LLM 动态推理）

from harness import ReActAgent

def my_llm(prompt: str) -> str:
    """替换为你自己的 LLM 调用"""
    import openai
    client = openai.OpenAI(api_key="your-key")
    response = client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": prompt}],
        temperature=0.1, max_tokens=2000,
    )
    return response.choices[0].message.content

agent = ReActAgent(
    name="data_analyst",
    tools=[GreetingTool(), WeatherTool()],
    system_prompt="You are a helpful assistant. Use tools to answer questions.",
    llm_call=my_llm,
    max_steps=5,
)

orchestrator = Orchestrator(agents=[agent])
result = orchestrator.run(task="What's the weather in Tokyo?")

4. Loop Engineering 加固（可选启用）

from harness import LoopConfig, LoopState, Terminator

config = LoopConfig(
    max_steps=10,
    consecutive_failures_limit=3,
    enable_terminator=True,
    enable_smart_retry=True,
    enable_loop_state=True,
)

orchestrator = Orchestrator(
    agents=[agent],
    loop_config=config,
)

5. 多 Agent 协作

agent_a = Agent(name="parser", tools=[...])
agent_b = Agent(name="auditor", tools=[...])

orchestrator = Orchestrator(agents=[agent_a, agent_b])
# agent_a 的 final_data 自动注入 agent_b 的 context["previous_agent_output"]
result = orchestrator.run(task="Parse and audit the document")

6. 验证规则

from harness import RuleEngine, NotNullRule, ColumnConsistencyRule

verifier = RuleEngine()
verifier.register(NotNullRule())
verifier.register(ColumnConsistencyRule())

orchestrator = Orchestrator(agents=[agent], verifier=verifier)

---

模块说明

模块	作用	关键类
tool_registry	Tool 抽象与注册中心	Tool, ToolResult, ToolRegistry
agent	Agent 基类 + ReActAgent	Agent（Pipeline）, ReActAgent（LLM驱动推理）
orchestrator	多 Agent 编排器	Orchestrator, OrchestrationResult
verification	可插拔验证	Rule, RuleEngine, NotNullRule, ColumnConsistencyRule, TableCountRule
loop/termination	多条件终止引擎	Terminator, TermReason（6 种）
loop/smart_retry	智能重试	SmartRetry, ErrorCategory, DeadLetter
loop/loop_state	结构化状态追踪	LoopState, AttemptRecord, Strategy
loop/hitl	人机协作	HITLManager, HITLRequest（同步+异步双模式）
loop/plan_agent	先规划再执行	PlanAgent, PlanStep

---

Loop Engineering（v0.2.0 新增）

Agent 循环的生产级加固，全部通过 LoopConfig 灰度开关逐模块启用：

config = LoopConfig(
    # Terminator
    max_steps=10,
    consecutive_failures_limit=3,
    loop_detection_threshold=5,

    # SmartRetry
    base_delay=1.0,
    max_delay=30.0,
    jitter_factor=0.1,

    # HITL
    hitl_timeout_seconds=300,

    # 灰度开关
    enable_terminator=True,
    enable_smart_retry=True,
    enable_loop_state=True,
    enable_hitl=True,
    enable_plan_agent=False,  # 默认关闭
)

Terminator 终止条件优先级

P0: EXPLICIT_ABORT / TASK_COMPLETE
P1: MAX_STEPS / CONSECUTIVE_FAILURES
P2: LOOP_DETECTED / ALL_TOOLS_EXHAUSTED

SmartRetry 错误分类

• TRANSIENT（瞬时故障）→ 指数退避重试
• PERMANENT（永久故障）→ 立即返回失败，进死信队列
• UNKNOWN（未知）→ 保守重试

支持中英文错误关键词分类。

HITL 双模式

• 异步模式：挂起任务 → 返回 AWAITING_HUMAN_DECISION → 等待外部 API 决策
• 同步模式：调用回调函数阻塞等待（兼容旧接口）

---

设计原则

1. 零依赖：只使用 Python 标准库（dataclasses, enum, concurrent.futures, hashlib, json, re, time）
2. LLM 由外部注入：Agent 框架本身不绑定任何 LLM SDK，通过 llm_call 回调注入
3. 分层明确：Tool → Agent → Orchestrator 三层独立，每层可单独测试
4. 灰度可控：Loop Engineering 所有模块通过 LoopConfig 按需启用

---

实战验证

已在 BankDataViz 项目中实战验证：

• 12 家银行 2020-2024 年报全流程处理
• Pipeline：OCR → LLM分析 → 表格重建 → 审计
• ReAct：自然语言驱动的多 Tool 协作数据分析
• Loop Engineering：连续失败保护、循环检测、死信队列

---

License

MIT