这是一个基于大型语言模型 (LLM) 的 AI Agent 框架，旨在通过自然语言指令自动执行编码和系统管理任务。Agent 可以解析任务，与 LLM 交互生成操作计划（文件操作、程序执行），执行这些计划，并根据反馈进行调整。
该框架的核心设计思想是将配置、数据结构、API 交互、文件执行和程序执行等功能模块化，使其更易于维护和扩展。
- 配置驱动: 通过
config.json文件集中管理 API 密钥、模型参数、工作路径、执行限制等，方便调整。 - 任务管理: 支持通过任务 ID (
--task_id) 启动和管理不同的任务，每个任务拥有独立的工作目录和聊天记录。 - 模块化设计: 清晰分离了核心逻辑 (
agent.py)、API 交互 (chat_api.py)、数据结构 (data_struct.py)、文件操作 (file_exec.py) 和程序执行 (program_exec.py)。 - 文件操作: 支持创建、删除、读取、替换文件内容以及创建/删除目录，包含路径安全检查和可配置的文件读取限制。
- 程序执行: 支持执行外部命令/程序，可并行执行多个命令，包含超时控制、输出/错误捕获和可选的周期性监控反馈机制。
- LLM 交互: 封装了与 OpenAI 兼容 API 的交互逻辑，包括流式响应处理、上下文窗口管理和基本的错误重试。
- 状态持久化: 将每个任务的聊天记录保存到 JSON 文件中，方便调试和恢复。
config.json: 配置文件，包含所有可调参数。agent.py: Agent 的主控制流。加载配置，初始化组件，管理任务循环，解析 LLM 响应，调用执行器，处理反馈。chat_api.py: 负责与 LLM API 通信。构建请求，发送请求，处理流式响应，管理对话历史上下文。data_struct.py: 定义 Agent 内部及与 LLM 交互所使用的所有数据类 (Dataclasses) 和枚举 (Enums)，如操作请求、反馈结果等。file_exec.py:FileExecutor类，负责执行所有文件系统相关的操作。根据config.json中的设置进行操作。program_exec.py:CommandExecutor和ParallelExecutor类，负责执行外部命令。CommandExecutor处理单个命令，ParallelExecutor管理多个命令的并行执行和监控。ProgramCheck类用于实现周期性反馈。
配置文件 config.json 分为几个主要部分：
api: LLM API 相关设置 (密钥api_key,base_url,model,timeout)。paths: 工作目录 (work_dir)、任务目录 (task_dir)、Prompt 文件 (prompt_file) 的相对路径。agent: Agent 行为设置 (聊天记录保存数量save_chat_count, 初始上下文消息数initial_context_messages)。file_execution: 文件操作相关设置 (最大读取长度read_max_length, 应用长度限制的文件类型read_limit_extensions, 允许读取的文件类型allowed_read_extensions, 默认编码default_encoding)。program_execution: 程序执行相关设置 (默认超时default_timeout, 监控线程检查间隔monitor_interval, 反馈给 AI 的检查间隔feedback_check_interval, 输出缓存大小限制output_buffer_limit)。
注意: 首次使用前，请务必在 config.json 中填入有效的 api_key。
- 配置: 确保
config.json文件存在于agent.py同级目录下，并已填入正确的 API 密钥和所需配置。 - 安装依赖: 确保已安装所需的 Python 库 (主要是
openai):pip install openai
- 运行 Agent: 在
ai/目录下执行以下命令启动 Agent 并指定任务 ID：
# -*- coding: utf-8 -*-
import json
import re
import os
import argparse # Import argparse for potential command-line overrides
import logging # Added logging
from typing import Any, Dict, List, Optional # Added Optional
from agent_coder import AutoCoder
logger = logging.getLogger(__name__) # Module-level logger
def setup_logging(config: Dict[str, Any]):
"""Configures logging based on the provided configuration."""
log_config = config.get('logging', {})
log_level_str = log_config.get('log_level', 'INFO').upper()
log_level = getattr(logging, log_level_str, logging.INFO)
log_format = log_config.get('log_format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
log_date_format = log_config.get('log_date_format', '%Y-%m-%d %H:%M:%S')
log_file = log_config.get('log_file') # Log file path is optional
formatter = logging.Formatter(log_format, datefmt=log_date_format)
# Clear existing handlers to avoid duplicate logs if re-configured
if logger.hasHandlers():
logger.handlers.clear()
# Configure console handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
logger.addHandler(ch)
# Configure file handler if log_file is specified
if log_file:
try:
# Ensure log directory exists if log_file path includes directories
log_dir = os.path.dirname(log_file)
if log_dir and not os.path.exists(log_dir):
os.makedirs(log_dir)
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.info(f"Logging to file: {log_file}")
except Exception as e:
logger.error(f"Failed to configure file logging to {log_file}: {e}")
logger.setLevel(log_level)
logger.info(f"Logging initialized. Level: {log_level_str}")
DEFAULT_CONFIG_PATH = 'config.json'
def load_config(config_path: str = DEFAULT_CONFIG_PATH) -> Dict[str, Any]:
"""Loads configuration from a JSON file."""
if not os.path.exists(config_path):
raise FileNotFoundError(f"Configuration file not found: {config_path}")
try:
with open(config_path, 'r') as f:
config = json.load(f)
# Basic validation (can be expanded)
if 'api' not in config or 'paths' not in config or 'agent' not in config:
raise ValueError("Config file missing required top-level keys: 'api', 'paths', 'agent'")
return config
except json.JSONDecodeError as e:
raise ValueError(f"Error decoding JSON from config file {config_path}: {e}")
except Exception as e:
raise RuntimeError(f"Error loading config file {config_path}: {e}")
# --- JSON Extraction (Improved) ---
def extract_json_content(ai_response: str) -> Dict[str, Any]:
"""
提取AI返回的结构化JSON内容。
参数:
ai_response (str): AI返回的内容，包含JSON格式的结构化文本。
返回:
Dict[str, Any]: 提取的结构化字段字典。
"""
# 使用正则表达式提取JSON内容
match = re.search(r'#####--\s*(.*?)\s*--#####', ai_response, re.DOTALL)
if match:
json_content = match.group(1) # 提取匹配的内容
try:
# 尝试将字符串解析为JSON对象
structured_data = json.loads(json_content)
return structured_data
except json.JSONDecodeError as e:
raise ValueError(f"无法正确解析提取的JSON内容，确保其为有效的JSON格式（####--json内容--####）。{e}")
else:
# Keep original error message for consistency if needed, or improve it
raise ValueError("未找到有效的JSON结构化内容（单个），确保其为有效的JSON格式（#####--json内容--#####）。")
def main():
"""Main function to load config, setup logging, and run the AutoCoder."""
# parser = argparse.ArgumentParser(description="AutoCoder Agent")
# parser.add_argument('--config', type=str, default=DEFAULT_CONFIG_PATH, help=f"Path to configuration file (default: {DEFAULT_CONFIG_PATH})")
# parser.add_argument('--task_id', type=str, required=True, help="Unique ID for the current task")
# parser.add_argument('--request', type=str, default="", help="Initial user request/prompt for the task")
# # Add other command-line arguments if.txt needed (e.g., overriding work_dir)
#
# args = parser.parse_args()
config_path = fr'C:\Users\config.json'
task_id = '001'
try:
config = load_config(config_path)
# Setup logging right after loading config
setup_logging(config)
logger.info("Configuration loaded successfully.")
except (FileNotFoundError, ValueError, RuntimeError) as e:
# Logger might not be set up, so print is safer here
print(f"CRITICAL: Error loading configuration: {e}")
# Attempt to log as well, might fail if setup_logging failed
try:
logger.critical(f"Error loading configuration: {e}", exc_info=True)
except Exception:
pass
return # Exit if config fails
logger.debug(
f"Loaded config: {json.dumps(config, indent=2, ensure_ascii=False)}") # Log loaded config at DEBUG level
# You might want to allow overriding work_dir via command line too
# config['paths']['work_dir'] = args.work_dir if.txt args.work_dir else config['paths']['work_dir']
# Use initial request from command line if.txt provided
initial_content = f"请你在{config['paths']['work_dir']}下实现贪吃蛇游戏" # Default if no request given
logger.info(f"Starting AutoCoder for task_id: {task_id}")
logger.debug(f"Initial request content (first 100 chars): {initial_content[:100]}")
try:
# Pass the config file path to the AutoCoder constructor
auto_coder = AutoCoder(config=config,
config_path=config_path, # Pass the actual config path
task_id=task_id,
initial_chat_content=initial_content)
auto_coder.run()
logger.info(f"AutoCoder run finished for task_id: {task_id}")
except (FileNotFoundError, ValueError, RuntimeError) as e:
# These are likely setup errors before the main loop
logger.critical(f"Setup error during AutoCoder initialization: {e}", exc_info=True)
except Exception as e:
# Catch-all for unexpected errors during the run
logger.critical(f"An unexpected error occurred during AutoCoder execution: {e}", exc_info=True)
if __name__ == "__main__":
main()
* 加载 `config.json`。
* 在 `paths.task_dir` 下创建或使用 `<your_task_id>` 目录。
* 加载该任务的聊天记录（如果存在）。
* 加载 `paths.prompt_file` 中的系统 Prompt。
* 开始与 LLM 交互以执行 `task.txt` (如果存在于任务目录中) 或等待用户输入初始任务。
- Python 3.7+
openailibrary