1. 构建最小Agent Harness骨架

任务目标

构建一个最小可运行的 Agent Harness，包含三个核心能力：解析用户输入、注册并调用工具、维护执行循环。最终通过命令行输入触发一次真实的工具调用并返回结果。

环境依赖

仅需 Python 3.9+，无第三方库依赖。代码使用标准库 json 与 datetime。

核心实现

import json
import datetime
from typing import Dict, Callable, Any

class ToolRegistry:
    """工具注册表：管理工具名到函数的映射"""
    def __init__(self):
        self._tools: Dict[str, Callable] = {}

    def register(self, name: str, func: Callable) -> None:
        self._tools[name] = func

    def get(self, name: str) -> Callable:
        if name not in self._tools:
            raise KeyError(f"未知工具: {name}")
        return self._tools[name]

    def list_tools(self) -> list:
        return list(self._tools.keys())

class AgentHarness:
    """最小 Agent 执行骨架"""
    def __init__(self):
        self.registry = ToolRegistry()
        self._register_default_tools()

    def _register_default_tools(self):
        # 注册一个示例工具：获取当前时间
        self.registry.register("get_time", self._get_time)

    def _get_time(self) -> str:
        return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def parse_input(self, raw_input: str) -> Dict[str, Any]:
        """解析输入为 {tool: 工具名, args: 参数字典}"""
        parts = raw_input.strip().split()
        if not parts:
            raise ValueError("输入不能为空")
        tool_name = parts[0]
        args = {}
        if len(parts) > 1:
            try:
                args = json.loads(" ".join(parts[1:]))
            except json.JSONDecodeError:
                raise ValueError("参数必须是合法JSON对象，例如: {\"format\": \"short\"}")
        return {"tool": tool_name, "args": args}

    def execute_tool(self, tool_name: str, args: Dict[str, Any]) -> str:
        """执行工具并返回字符串结果"""
        func = self.registry.get(tool_name)
        result = func(**args)
        return str(result)

    def run(self, user_input: str) -> str:
        """执行循环：解析 -> 执行 -> 返回"""
        try:
            parsed = self.parse_input(user_input)
            output = self.execute_tool(parsed["tool"], parsed["args"])
            return f"工具 [{parsed['tool']}] 执行结果: {output}"
        except (KeyError, ValueError, TypeError) as e:
            return f"执行失败: {str(e)}"

def main():
    harness = AgentHarness()
    print("最小Agent Harness已启动。可用工具:", harness.registry.list_tools())
    print("输入格式: 工具名 参数JSON(可选)，例如: get_time")
    print("输入 'exit' 退出")
    while True:
        try:
            user_input = input("> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if user_input.lower() in ("exit", "quit"):
            break
        if not user_input:
            continue
        print(harness.run(user_input))

if __name__ == "__main__":
    main()

运行方式

python agent_harness.py

验证方法

启动程序后，依次执行以下操作验证闭环：

1. 输入 get_time，观察输出是否包含当前系统时间（格式为 YYYY-MM-DD HH:MM:SS）。
2. 输入 get_time {}，应得到与上一步相同的结果，证明空参数JSON可被正确解析。
3. 输入 unknown_tool，应返回 执行失败: 未知工具: unknown_tool，证明错误处理路径有效。
4. 输入 get_time {"invalid": json}，应返回参数解析错误提示，证明非法JSON被拦截。
5. 输入 exit，程序正常退出。

以上步骤全部通过，即证明该 Harness 具备输入解析、工具注册、执行循环的最小可用能力。

扩展说明

如需添加新工具，在 AgentHarness 中新增方法并通过 self.registry.register("工具名", 方法名) 注册即可，无需修改执行循环逻辑。当前实现刻意保持最小，未包含异步、流式输出或多轮记忆，这些属于后续案例的扩展范围。