针对“使用纯Python完整实现OpenClaw”的要求，首先需明确OpenClaw是一个开源AI代理框架，其核心是一个多组件系统，包括网关（Gateway）、代理运行器（Agent Runner）、数据库和模型接口等[1]。直接使用纯Python从零开始复刻其全部功能是极其庞大的工程。因此，**最务实的纯Python实现路径，并非构建整个分布式系统，而是通过其提供的Python SDK或OpenAPI来构建和运行特定的代理（Agent），从而实现其核心的自动化能力**[3]。以下将基于此思路，提供一个完整的、不依赖其他复杂工具链的纯Python解决方案。 ### 1. 核心架构与实现路径 OpenClaw的设计允许通过API与其服务交互。我们将构建一个简化的本地版，其核心是一个能处理用户查询并执行工具调用（Tool Call）的Python脚本。这本质上是一个实现了OpenAI Function Calling或类似Agentic工作流的Agent程序[1]。 下表对比了官方OpenClaw系统与我们将实现的纯Python代理的核心区别： | 特性/组件 | 官方OpenClaw系统 (全量部署) | 纯Python实现 (简化版Agent) | | :--- | :--- | :--- | | **架构** | 微服务（Gateway, Runner, DB） | 单脚本，面向对象设计 | | **部署** | 需Node.js/Docker/Python环境[2][5] | 仅Python环境（建议3.8+） | | **模型接入** | 支持多种（Ollama[1]、LM Studio[2]、ModelArts[3]等） | 直接调用本地或远程兼容OpenAI API的模型服务 | | **核心流程** | 多Agent协作、技能(Skill)路由、持久化 | 单轮次“思考-行动-观察”循环 | | **目标** | 企业级、高并发、可扩展的智能体平台 | 轻量化、快速验证Agent概念的Python脚本 | ### 2. 完整实现代码与详解 此实现包含三个核心部分：一个可执行工具的基础类、一个本地/远程大模型调用器、一个主控Agent类。该Agent遵循ReAct（Reasoning and Acting）范式，进行循环推理和工具调用。 ```python """ 纯Python实现的简化版OpenClaw Agent核心程序。 功能：接收自然语言指令，调用定义好的工具函数完成任务。 """ import json import requests from typing import Dict, Any, Callable, List from dataclasses import dataclass from enum import Enum class ToolType(Enum): """定义工具类型，模仿OpenAI Function Calling。""" FUNCTION = "function" @dataclass class Tool: """工具定义类，用于描述一个可供Agent调用的函数。""" name: str description: str function: Callable parameters: Dict[str, Any] # 符合JSON Schema格式的参数定义 class OpenAIClient: """简化的大模型客户端，用于与兼容OpenAI API的模型服务交互。""" def __init__(self, base_url: str = "http://localhost:11434/v1", api_key: str = "ollama", model: str = "qwen2.5:7b"): """ 初始化模型客户端。 Args: base_url: OpenAI兼容API的地址。例如： - 本地Ollama: "http://localhost:11434/v1" [1] - LM Studio: "http://localhost:1234/v1" [2] api_key: API密钥（对于本地模型通常可为任意值）。 model: 要使用的模型名称。 """ self.base_url = base_url.rstrip('/') self.api_key = api_key self.model = model self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } def create_chat_completion(self, messages: List[Dict], tools: List[Tool] = None) -> Dict[str, Any]: """ 发送聊天补全请求，支持工具调用。 Args: messages: 对话历史消息列表。 tools: 可选，可供模型调用的工具列表。 Returns: 模型的完整响应JSON。 """ url = f"{self.base_url}/chat/completions" data = { "model": self.model, "messages": messages, "stream": False } if tools: # 将Tool对象转换为OpenAI API要求的格式 data["tools"] = [self._tool_to_openai_spec(tool) for tool in tools] data["tool_choice"] = "auto" # 让模型自行决定是否调用工具 try: response = requests.post(url, headers=self.headers, json=data, timeout=60) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"调用模型API失败: {e}") def _tool_to_openai_spec(self, tool: Tool) -> Dict[str, Any]: """将内部Tool对象转换为OpenAI API的工具定义格式。""" return { "type": ToolType.FUNCTION.value, "function": { "name": tool.name, "description": tool.description, "parameters": { "type": "object", "properties": tool.parameters.get("properties", {}), "required": tool.parameters.get("required", []) } } } class PythonOpenClawAgent: """纯Python实现的OpenClaw风格智能体。""" def __init__(self, llm_client: OpenAIClient): """ 初始化Agent。 Args: llm_client: 配置好的大模型客户端实例。 """ self.llm = llm_client self.tools: Dict[str, Tool] = {} # 工具名称到Tool对象的映射 self.conversation_history = [] # 维护对话历史 def register_tool(self, tool: Tool) -> None: """注册一个工具到Agent。""" self.tools[tool.name] = tool def run(self, user_query: str, max_iterations: int = 5) -> str: """ 执行Agent的主循环。 Args: user_query: 用户的自然语言查询。 max_iterations: 最大循环次数，防止无限循环。 Returns: Agent的最终回答。 """ self.conversation_history.append({"role": "user", "content": user_query}) print(f"[用户] {user_query}") for i in range(max_iterations): print(f"\n=== 第 {i+1} 轮思考 ===") # 1. 调用LLM进行推理，并决定是否使用工具 llm_response = self.llm.create_chat_completion( messages=self.conversation_history, tools=list(self.tools.values()) if self.tools else None ) message = llm_response['choices'][0]['message'] self.conversation_history.append(message) print(f"[助理] {message.get('content', '(无文本内容)')}") # 2. 检查是否有工具调用 if message.get('tool_calls'): for tool_call in message['tool_calls']: function_name = tool_call['function']['name'] function_args = json.loads(tool_call['function']['arguments']) print(f"[行动] 调用工具 `{function_name}`， 参数: {function_args}") # 3. 执行工具调用 tool_result = self._execute_tool(function_name, function_args) result_message = { "role": "tool", "tool_call_id": tool_call['id'], "name": function_name, "content": str(tool_result) } self.conversation_history.append(result_message) print(f"[观察] 工具返回: {tool_result}") else: # 没有工具调用，说明推理完成，返回最终答案 final_answer = message.get('content', '').strip() print(f"\n===