Agent Harness：AI工程师必须掌握的设计方法论

from pydantic import BaseModel, Field
from typing import Callable, Any
 
class ToolSchema(BaseModel):
    """工具的元数据描述"""
    name: str = Field(..., description="工具名称")
    description: str = Field(..., description="工具功能描述")
    parameters: dict[str, Any] = Field(..., description="JSON Schema格式参数")
    examples: list[dict[str, Any]] = Field(default_factory=list) # Few-shot示例
    category: str = Field("default", description="工具分类，用于层次路由")
 
class Tool(BaseModel):
    schema: ToolSchema
    handler: Callable
 
    model_config = {"arbitrary_types_allowed": True}

from enum import Enum
from pydantic import BaseModel
from typing import Any
 
class ToolErrorType(str, Enum):
    PARAMETER_ERROR = "parameter_error"
    EXECUTION_ERROR = "execution_error"
    NOT_FOUND = "not_found"
 
class ToolResult(BaseModel):
    """强制结构化的工具返回，防止幻觉"""
    success: bool
    data: Any = None
    error: ToolErrorType | None = None
    error_message: str | None = None

from pydantic import ValidationError
 
class ToolRouter:
    def route(self, tool_name: str, **kwargs) -> ToolResult:
        tool = self.get_tool(tool_name)
        if not tool:
            return ToolResult(
                success=False,
                error=ToolErrorType.NOT_FOUND,
                error_message=f"Tool {tool_name} not found"
            )
 
        # 第一层：Schema验证
        try:
            validated_args = tool.schema.parameters_model(**kwargs)
        except ValidationError as e:
            return ToolResult(
                success=False,
                error=ToolErrorType.PARAMETER_ERROR,
                error_message=str(e)
            )
 
        # 第二层：执行工具
        try:
            result = tool.handler(**validated_args.model_dump())
            return ToolResult(success=True, data=result)
        except Exception as e:
            return ToolResult(
                success=False,
                error=ToolErrorType.EXECUTION_ERROR,
                error_message=str(e)
            )

from typing import Generic, TypeVar, Any
from pydantic import BaseModel, Field
from datetime import datetime
import copy
 
T = TypeVar('T')
 
class Checkpoint(BaseModel):
    version: int
    timestamp: datetime = Field(default_factory=datetime.now)
    state: dict[str, Any]
    metadata: dict[str, Any] = Field(default_factory=dict)
 
class StateManager(Generic[T]):
    def __init__(self, initial_state: T):
        self.current_state = initial_state
        self.checkpoints: list[Checkpoint] = []
        self.version = 0
 
    def checkpoint(self, metadata: dict | None = None) -> int:
        """创建检查点"""
        self.version += 1
 
        state_data = (self.current_state.model_dump()
                      if isinstance(self.current_state, BaseModel)
                      else copy.deepcopy(self.current_state))
 
        checkpoint = Checkpoint(
            version=self.version,
            timestamp=datetime.now(),
            state=state_data,
            metadata=metadata or {}
        )
        self.checkpoints.append(checkpoint)
        return self.version
 
    def rollback(self, version: int) -> T:
        """时间旅行：回滚到指定版本"""
        for cp in reversed(self.checkpoints):
            if cp.version == version:
                if hasattr(self.current_state, "__class__") and issubclass(self.current_state.__class__, BaseModel):
                     self.current_state = self.current_state.__class__.model_validate(cp.state)
                else:
                     self.current_state = copy.deepcopy(cp.state)
                return self.current_state
        raise ValueError(f"Version {version} not found")

from abc import ABC, abstractmethod
from pydantic import BaseModel, Field
from typing import Any
 
class EvalScore(BaseModel):
    value: float = Field(..., ge=0, le=1)  # 强制分数在0-1之间
    reason: str
    metadata: dict[str, Any] = Field(default_factory=dict)
 
class Evaluator(ABC):
    @abstractmethod
    def evaluate(self, output: Any, context: dict) -> EvalScore:
        pass
 
class EvalPipeline:
    """评估管道：依次执行多层验证"""
    def __init__(self):
        self.validators: list[Evaluator] = []
 
    def evaluate(self, output: Any, context: dict) -> dict[str, EvalScore]:
        """执行所有评估器"""
        results = {}
        for i, validator in enumerate(self.validators):
            layer_name = validator.__class__.__name__
            results[layer_name] = validator.evaluate(output, context)
 
            # 快速失败：语法层失败则不继续
            if i == 0 and results[layer_name].value < 0.5:
                break
 
        return results
 
    def get_overall_score(self, results: dict[str, EvalScore]) -> float:
        """加权平均计算综合分数"""
        weights = {
            "SyntaxValidator": 0.4,
            "SemanticValidator": 0.3,
            "TaskValidator": 0.3
        }
        total = sum(
            results[k].value * weights.get(k, 0.1)
            for k in results
        )
        return max(0, min(1, total))

from pydantic import BaseModel, ConfigDict
from typing import Callable, Any
 
class Transition(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
 
    from_state: str
    to_state: str | Callable
    condition: Callable | None = None
 
class StateMachine:
    def __init__(self):
        self.states: dict[str, Callable] = {}
        self.transitions: list[Transition] = []
        self.current_state: str | None = None
 
    def add_state(self, name: str, handler: Callable) -> None:
        self.states[name] = handler
 
    def add_transition(self, transition: Transition) -> None:
        self.transitions.append(transition)
 
    def start(self, initial_state: str, context: dict) -> Any:
        self.current_state = initial_state
        result = None
        max_steps = 100  # 预算控制：防止无限循环
        step = 0
 
        while self.current_state and step < max_steps:
            step += 1
            handler = self.states.get(self.current_state)
            if not handler:
                raise ValueError(f"State {self.current_state} not found")
 
            result = handler(context)
            self.current_state = self._find_next_state(result, context)
 
        return result
 
    def _find_next_state(self, result: Any, context: dict) -> str | None:
        for trans in self.transitions:
            if trans.from_state != self.current_state:
                continue
            if trans.condition and not trans.condition(result, context):
                continue
            if callable(trans.to_state):
                return trans.to_state(result, context)
            return trans.to_state
        return None