第22章 从零构建 Mini-Harness —— Agent Loop 的最小实现

# Mini-Harness 初始化模块

def initialize(prompt, system_prompt=None):
    """
    初始化 Mini-Harness
    
    参数:
        prompt: 用户任务描述
        system_prompt: 系统提示词（可选）
    
    返回:
        state: 包含 messages 数组的状态
    """
    # 1. 默认 System Prompt
    if system_prompt is None:
        system_prompt = """
你是代码助手 Agent。你可以使用以下工具：

工具列表：
- Read: 读取文件
- Write: 写入文件
- Bash: 执行命令

约束：
- 每次只调用一个工具
- 工具调用后等待结果
- 完成任务后返回 end_turn
"""
    
    # 2. 构建初始 messages
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt}
    ]
    
    # 3. 初始化状态
    state = {
        "messages": messages,
        "loops": 0,
        "tool_calls": 0,
        "current_tool": None,
        "last_result": None
    }
    
    return state


# 测试初始化
def test_initialize():
    prompt = "读取 app.ts 文件，分析代码结构"
    state = initialize(prompt)
    
    print("初始化完成:")
    print(f"- messages 数量: {len(state['messages'])}")
    print(f"- loops: {state['loops']}")
    print(f"- tool_calls: {state['tool_calls']}")
    
    return state


# 运行测试
test_initialize()

# Mini-Harness Agent Loop

def agent_loop(state, model_api, tools):
    """
    Agent Loop 核心循环
    
    参数:
        state: 包含 messages 的状态
        model_api: Model API 接口
        tools: 工具执行函数映射
    
    返回:
        state: 更新后的状态
        response: Model 的最终响应
    """
    while True:
        # 1. observe: 接收 messages
        messages = state["messages"]
        
        # 2. think: Model 推理
        response = model_api(messages)
        
        # 3. 提取 stop_reason
        stop_reason = response.get("stop_reason", "end_turn")
        
        # 4. decide: 决策下一步
        if stop_reason == "end_turn":
            # end_turn: 结束对话
            return state, response
        
        elif stop_reason == "tool_use":
            # tool_use: 执行工具
            tool_use = extract_tool_use(response)
            
            # 5. act: 执行工具
            tool_name = tool_use["name"]
            tool_params = tool_use["parameters"]
            
            if tool_name in tools:
                tool_result = tools[tool_name](tool_params)
            else:
                tool_result = f"错误：工具 {tool_name} 不存在"
            
            # 6. feedback: 追加结果
            state["messages"].append({
                "role": "assistant",
                "content": response["content"],
                "tool_use": tool_use
            })
            
            state["messages"].append({
                "role": "user",
                "content": tool_result
            })
            
            # 7. 更新状态
            state["loops"] += 1
            state["tool_calls"] += 1
            state["current_tool"] = tool_name
            state["last_result"] = tool_result
            
            # 8. 循环限制（防止无限循环）
            if state["loops"] > 20:
                print("警告：循环超过 20 次，强制结束")
                return state, response


def extract_tool_use(response):
    """
    从 Model 响应中提取 tool_use
    
    参数:
        response: Model 响应
    
    返回:
        tool_use: 工具调用信息
    """
    content = response.get("content", "")
    
    # 简化版：假设 Model 返回格式为 JSON
    # 实际情况需要解析 Model 的具体输出格式
    
    import json
    import re
    
    # 尻找 JSON 格式的 tool_use
    json_match = re.search(r'\{[^}]+\}', content)
    if json_match:
        tool_info = json.loads(json_match.group())
        return {
            "name": tool_info.get("tool", "unknown"),
            "parameters": tool_info.get("params", {})
        }
    
    return {"name": "unknown", "parameters": {}}

# Mini-Harness 基本工具系统

import os
import subprocess

def tool_read(params):
    """
    Read 工具：读取文件
    
    参数:
        params: {"file_path": "/path/to/file"}
    
    返回:
        文件内容
    """
    file_path = params.get("file_path")
    
    if not file_path:
        return "错误：缺少 file_path 参数"
    
    if not os.path.exists(file_path):
        return f"错误：文件 {file_path} 不存在"
    
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        return f"文件内容:\n{content}"
    
    except Exception as e:
        return f"错误：读取文件失败 - {str(e)}"


def tool_write(params):
    """
    Write 工具：写入文件
    
    参数:
        params: {"file_path": "/path/to/file", "content": "file content"}
    
    返回:
        写入结果
    """
    file_path = params.get("file_path")
    content = params.get("content")
    
    if not file_path or not content:
        return "错误：缺少 file_path 或 content 参数"
    
    try:
        # 创建目录（如果不存在）
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        
        # 写入文件
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        
        return f"成功：文件 {file_path} 已写入"
    
    except Exception as e:
        return f"错误：写入文件失败 - {str(e)}"


def tool_bash(params):
    """
    Bash 工具：执行命令
    
    参数:
        params: {"command": "ls -la"}
    
    返回:
        命令输出
    """
    command = params.get("command")
    
    if not command:
        return "错误：缺少 command 参数"
    
    try:
        # 执行命令
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        
        # 返回输出
        if result.returncode == 0:
            return f"命令输出:\n{result.stdout}"
        else:
            return f"错误：命令执行失败\n{result.stderr}"
    
    except subprocess.TimeoutExpired:
        return "错误：命令执行超时（30秒）"
    
    except Exception as e:
        return f"错误：执行命令失败 - {str(e)}"


# 创建工具映射
tools = {
    "Read": tool_read,
    "Write": tool_write,
    "Bash": tool_bash
}


# 测试工具
def test_tools():
    # 测试 Read
    result = tool_read({"file_path": "/tmp/test.txt"})
    print(f"Read 结果:\n{result}\n")
    
    # 测试 Write
    result = tool_write({"file_path": "/tmp/test.txt", "content": "Hello Mini-Harness"})
    print(f"Write 结果:\n{result}\n")
    
    # 测试 Bash
    result = tool_bash({"command": "ls /tmp"})
    print(f"Bash 结果:\n{result}\n")


# 运行测试
test_tools()

# Mini-Harness Model API 接口（简化版）

def model_api(messages):
    """
    Model API 接口（简化版）
    
    参数:
        messages: messages 数组
    
    返回:
        response: Model 响应
    """
    # 简化版：模拟 Model 响应
    # 实际情况需要调用真实的 Model API
    
    last_message = messages[-1]["content"]
    
    # 简化版决策逻辑
    if "读取" in last_message or "查看" in last_message:
        # 决定调用 Read 工具
        return {
            "content": "决定读取文件。{\"tool\": \"Read\", \"params\": {\"file_path\": \"app.ts\"}}",
            "stop_reason": "tool_use"
        }
    
    elif "写入" in last_message or "创建" in last_message:
        # 决定调用 Write 工具
        return {
            "content": "决定写入文件。{\"tool\": \"Write\", \"params\": {\"file_path\": \"output.txt\", \"content\": \"result\"}}",
            "stop_reason": "tool_use"
        }
    
    elif "执行" in last_message or "运行" in last_message:
        # 决定调用 Bash 工具
        return {
            "content": "决定执行命令。{\"tool\": \"Bash\", \"params\": {\"command\": \"npm test\"}}",
            "stop_reason": "tool_use"
        }
    
    else:
        # end_turn: 结束对话
        return {
            "content": "任务已完成。",
            "stop_reason": "end_turn"
        }


# 实际 Model API 调用示例（伪代码）
def real_model_api(messages):
    """
    实际 Model API 调用（伪代码）
    
    实际情况需要：
    1. 调用 Anthropic API 或其他 Model API
    2. 处理 API 认证
    3. 处理 API 响应格式
    """
    import anthropic
    
    client = anthropic.Anthropic(api_key="YOUR_API_KEY")
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        messages=messages
    )
    
    return {
        "content": response.content[0].text,
        "stop_reason": response.stop_reason
    }

# Mini-Harness 完整实现

class MiniHarness:
    """
    Mini-Harness: Agent Loop 的最小实现
    """
    
    def __init__(self, system_prompt=None):
        """
        初始化 Mini-Harness
        
        参数:
            system_prompt: 系统提示词（可选）
        """
        self.system_prompt = system_prompt or self.default_system_prompt()
        self.tools = self.create_tools()
        self.state = None
    
    def default_system_prompt(self):
        """默认 System Prompt"""
        return """
你是代码助手 Agent。你可以使用以下工具：

工具列表：
- Read: 读取文件（参数：file_path）
- Write: 写入文件（参数：file_path, content）
- Bash: 执行命令（参数：command）

约束：
- 每次只调用一个工具
- 工具调用后等待结果
- 完成任务后返回 end_turn
"""
    
    def create_tools(self):
        """创建工具映射"""
        return {
            "Read": self.tool_read,
            "Write": self.tool_write,
            "Bash": self.tool_bash
        }
    
    def tool_read(self, params):
        """Read 工具"""
        file_path = params.get("file_path")
        
        if not file_path:
            return "错误：缺少 file_path 参数"
        
        if not os.path.exists(file_path):
            return f"错误：文件 {file_path} 不存在"
        
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            return f"文件内容:\n{content}"
        except Exception as e:
            return f"错误：读取文件失败 - {str(e)}"
    
    def tool_write(self, params):
        """Write 工具"""
        file_path = params.get("file_path")
        content = params.get("content")
        
        if not file_path or not content:
            return "错误：缺少 file_path 或 content 参数"
        
        try:
            os.makedirs(os.path.dirname(file_path), exist_ok=True)
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)
            return f"成功：文件 {file_path} 已写入"
        except Exception as e:
            return f"错误：写入文件失败 - {str(e)}"
    
    def tool_bash(self, params):
        """Bash 工具"""
        command = params.get("command")
        
        if not command:
            return "错误：缺少 command 参数"
        
        try:
            result = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=30
            )
            
            if result.returncode == 0:
                return f"命令输出:\n{result.stdout}"
            else:
                return f"错误：命令执行失败\n{result.stderr}"
        except subprocess.TimeoutExpired:
            return "错误：命令执行超时（30秒）"
        except Exception as e:
            return f"错误：执行命令失败 - {str(e)}"
    
    def initialize(self, prompt):
        """初始化"""
        self.state = {
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": prompt}
            ],
            "loops": 0,
            "tool_calls": 0
        }
        return self.state
    
    def agent_loop(self, model_api):
        """Agent Loop"""
        while True:
            messages = self.state["messages"]
            
            # Model 推理
            response = model_api(messages)
            
            stop_reason = response.get("stop_reason", "end_turn")
            
            if stop_reason == "end_turn":
                return response
            
            elif stop_reason == "tool_use":
                tool_use = self.extract_tool_use(response)
                
                tool_name = tool_use["name"]
                tool_params = tool_use["parameters"]
                
                if tool_name in self.tools:
                    tool_result = self.tools[tool_name](tool_params)
                else:
                    tool_result = f"错误：工具 {tool_name} 不存在"
                
                # 追加结果
                self.state["messages"].append({
                    "role": "assistant",
                    "content": response["content"]
                })
                
                self.state["messages"].append({
                    "role": "user",
                    "content": tool_result
                })
                
                # 更新状态
                self.state["loops"] += 1
                self.state["tool_calls"] += 1
                
                # 循环限制
                if self.state["loops"] > 20:
                    print("警告：循环超过 20 次，强制结束")
                    return response
    
    def extract_tool_use(self, response):
        """提取 tool_use"""
        content = response.get("content", "")
        
        import json
        import re
        
        json_match = re.search(r'\{[^}]+\}', content)
        if json_match:
            tool_info = json.loads(json_match.group())
            return {
                "name": tool_info.get("tool", "unknown"),
                "parameters": tool_info.get("params", {})
            }
        
        return {"name": "unknown", "parameters": {}}
    
    def run(self, prompt, model_api):
        """完整运行"""
        # 1. 初始化
        self.initialize(prompt)
        
        # 2. Agent Loop
        response = self.agent_loop(model_api)
        
        # 3. 返回结果
        return {
            "response": response,
            "loops": self.state["loops"],
            "tool_calls": self.state["tool_calls"],
            "messages": self.state["messages"]
        }


# 测试 Mini-Harness
def test_mini_harness():
    harness = MiniHarness()
    
    prompt = "读取 /tmp/test.txt 文件"
    
    result = harness.run(prompt, model_api)
    
    print("运行完成:")
    print(f"- 循环次数: {result['loops']}")
    print(f"- 工具调用次数: {result['tool_calls']}")
    print(f"- 最终响应: {result['response']['content']}")


# 运行测试
test_mini_harness()