# Fastants **Repository Path**: nkdit/fastants ## Basic Information - **Project Name**: Fastants - **Description**: CS模式的多智能体协同框架 1. 单Agent记忆 2. 多Agent协同 - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 1 - **Created**: 2026-01-23 - **Last Updated**: 2026-01-23 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # FastAnts - 带有记忆的多Agent框架 FastAnts 是一个强大的多Agent协作框架，具备RAG增强的长时记忆、反思机制和灵活的工作流编排能力。 ## 🚀 核心特性 - **🧠 RAG增强记忆**: 基于向量数据库的语义搜索和长期记忆存储 - **🔄 反思机制**: Agent通过结构化反思从经验中学习 - **🎛️ 灵活编排**: 支持顺序、并行和自定义工作流模式 - **🔌 插件化架构**: 基于模板的Agent创建和配置 - **📝 智能日志**: 结构化日志输出，支持彩色控制台和文件持久化 - **🛡️ 权限控制**: 细粒度的文件访问控制 - **⚡ 性能优化**: 记忆压缩和高效的向量操作 ## 📁 项目结构 ``` fastants/ ├── src/agentforge/ # 🟢 框架源码 │ ├── core/ # 核心基础设施 │ │ ├── base_agent.py # Agent基类，实现生命周期管理 │ │ ├── memory.py # 记忆内核，RAG增强 │ │ └── workflow.py # 工作流编排引擎 │ ├── agents/ # Agent工厂和模板 │ │ └── agent_factory.py # 动态Agent创建 │ └── utils/ # 工具模块 │ ├── file_io.py # 文件操作抽象 │ └── logger.py # 结构化日志系统 ├── examples/ # 🔵 示例实现 │ └── rpt_to_json/ # RPT转换示例 │ ├── main.py # 主入口 │ ├── impl/ # 自定义Agent实现 │ └── shared/ # 共享工作空间 ├── legacy/ # 原有代码（参考） ├── pyproject.toml # 项目配置 ├── requirements.txt # 依赖列表 └── README.md # 项目文档 ``` ## 🏃‍♂️ 快速开始 ### 安装依赖 ```bash # 安装框架核心依赖 pip install -r requirements.txt # 开发环境安装 pip install -e . ``` ### 基础使用示例 ```python from agentforge import WorkflowEngine, AgentFactory, SimpleStopConditions from pathlib import Path # 1. 创建工作流引擎 engine = WorkflowEngine(max_rounds=10) # 2. 使用工厂创建Agent factory = AgentFactory() developer = factory.create_agent( agent_type="code_generator", name="Developer", work_dir="./workspace/dev" ) tester = factory.create_agent( agent_type="tester", name="Tester", work_dir="./workspace/test" ) # 3. 注册Agent到引擎 engine.register_agent(developer) engine.register_agent(tester) # 4. 配置停止条件 engine.add_stop_condition( SimpleStopConditions.file_contains("feedback.md", "[PASS]") ) # 5. 添加上下文钩子 def read_feedback_hook(round_idx): if Path("feedback.md").exists(): content = Path("feedback.md").read_text() return {"feedback": content[-500:]} # 最近500字符 return {} engine.add_round_hook(read_feedback_hook) # 6. 运行工作流 success = engine.run( task_instruction="实现新功能并确保所有测试通过", orchestration_pattern="sequential" ) ``` ## 🧠 记忆系统 FastAnts实现三层记忆架构，模拟人类认知系统： ### 1. 短时记忆（Short-term） - Agent执行期间的上下文窗口 - 任务结束后自动清理 ### 2. 工作记忆（Working Memory） - `private_memory.md` 文件存储当前策略和反思 - 自压缩算法，当内容超过阈值时自动压缩 - 保留核心策略和最新反思 ### 3. 长期情景记忆（Long-term Episodic） - 基于ChromaDB的向量存储 - 使用sentence-transformers进行语义嵌入 - 支持RAG（检索增强生成）的智能召回 ### 记忆生命周期 ```python # Agent执行流程 class Agent: def run(self, task, context): # 1. 召回（Recall） relevant_memories = self.memory_kernel.recall(context) # 2. 行动（Act） result = self.execute_with_aider(task, relevant_memories) # 3. 巩固（Consolidate） self.consolidate_to_long_term_memory() ``` ## 🎛️ 工作流编排 ### 编排模式 ```python # 顺序执行 - Agent依次执行 engine.run(task="...", orchestration_pattern="sequential") # 并行执行 - Agent同时执行（实验性） engine.run(task="...", orchestration_pattern="parallel") ``` ### 终止条件 ```python # 文件包含特定标记 engine.add_stop_condition( SimpleStopConditions.file_contains("result.md", "✅ SUCCESS") ) # 文件存在 engine.add_stop_condition( SimpleStopConditions.file_exists("output.json") ) # 自定义条件 engine.add_stop_condition(lambda: check_custom_logic()) ``` ### 轮次钩子 ```python # 读取文件内容 engine.add_round_hook( CommonHooks.read_file_content("feedback.md", max_chars=1000) ) # 提取错误信息 engine.add_round_hook( CommonHooks.extract_errors( "logs.txt", error_patterns=[r"ERROR: (.*)", r"Exception: (.*)"] ) ) ``` ## 🔧 Agent配置 ### 使用预定义模板 ```python # 代码生成器 factory.create_agent("code_generator", name="Dev", work_dir="./dev") # 代码审查员 factory.create_agent("code_reviewer", name="Reviewer", work_dir="./review") # 测试员 factory.create_agent("tester", name="Tester", work_dir="./test") ``` ### 自定义Agent ```python from agentforge.core.base_agent import BaseAgent class CustomAgent(BaseAgent): def __init__(self, name, work_dir, custom_param=None): super().__init__(name, work_dir) self.custom_param = custom_param def get_system_prompt(self): return """你是一个专门的AI助手，负责... 你的职责： 1. ... 2. ... 3. ... """ def get_context_files(self): return { "edit": [ f"{self.work_dir}/main.py", f"{self.work_dir}/config.py" ], "read_only": [ f"{self.work_dir}/requirements.md", f"{self.work_dir}/api_docs.md" ] } ``` ### YAML模板配置 ```yaml # templates/my_agent.yaml system_prompt: | 你是一个数据分析专家，专门处理大型数据集。 你的能力： - 数据清洗和预处理 - 统计分析和可视化 - 机器学习模型训练 edit_patterns: - "*.py" - "*.ipynb" read_only_patterns: - "data/*.csv" - "docs/*.md" model_config: model: "openai/GPT-4" temperature: "0.2" ``` ## 📊 示例：RPT转JSON转换器 查看 `examples/rpt_to_json_converter/` 目录，了解如何使用FastAnts构建具体应用： ```bash # 运行示例 python examples/rpt_to_json_converter/main.py ``` 该示例展示了： - 自定义Agent实现 - 顺序工作流执行 - 基于文件的通信 - 错误处理和反馈机制 ## 🔧 高级特性 ### 记忆检索优化 ```python # 基于来源过滤记忆 memories = agent.memory_kernel.recall( query="如何处理JSON解析错误", filter_dict={"source": "error_solution"} ) # 限制返回数量 memories = agent.memory_kernel.recall( query="性能优化技巧", n_results=5 ) ``` ### 自定义记忆元数据 ```python # 存储时添加元数据 agent.memory_kernel.memorize( text="重要的解决方案", source="success_case", metadata={ "category": "optimization", "effectiveness": "high", "context": "json_processing" } ) ``` ### 记忆统计和管理 ```python # 获取记忆统计 stats = agent.get_memory_stats() print(f"总记忆数: {stats['total_memories']}") print(f"私有记忆大小: {stats['private_memory_size']} bytes") # 清理所有记忆 agent.clear_all_memories() ``` ## 🔍 日志和监控 ### 结构化日志 ```python from agentforge.utils import get_logger # 创建日志器 logger = get_logger("MyAgent", log_dir="./logs") # 记录不同级别的日志 logger.info("开始执行任务", task="data_processing", step=1) logger.warning("发现潜在问题", issue="missing_data", count=5) logger.error("执行失败", error=str(e), context="stage_3") # 专用日志方法 logger.log_agent_start("DataProcessor", "清洗用户数据") logger.log_agent_complete("DataProcessor", success=True) logger.log_workflow_start([agent1, agent2], rounds=5) ``` ### 工作流持久化 ```python # 启用工作流状态持久化 engine = WorkflowEngine( max_rounds=10, persistence_dir="./workflow_state" ) # 工作流状态会自动保存到JSON文件 # 支持中断后恢复 ``` ## 🚀 部署和扩展 ### 环境变量 ```bash # OpenAI API配置 export OPENAI_API_KEY=your_api_key export OPENAI_API_BASE=https://api.openai.com/v1 # Aider配置 export AIDER_MODEL=openai/GPT-4 export AIDER_MAX_TOKENS=100000 # 框架配置 export AGENTFORGE_LOG_LEVEL=INFO export AGENTFORGE_LOG_DIR=./logs ``` ### Docker部署 ```dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY src/ ./src/ COPY examples/ ./examples/ CMD ["python", "examples/rpt_to_json_converter/main.py"] ``` ### 扩展框架 ```python # 创建自定义工作流引擎 class CustomWorkflowEngine(WorkflowEngine): def _execute_custom_pattern(self, task, context): # 实现自定义编排逻辑 pass # 创建自定义记忆后端 class CustomMemoryKernel(MemoryKernel): def __init__(self, agent_name, custom_config): # 集成其他向量数据库 pass ``` ## 🧪 开发和测试 ### 运行测试 ```bash # 安装开发依赖 pip install -e .[dev] # 运行所有测试 pytest # 运行特定测试 pytest tests/test_workflow.py # 生成覆盖率报告 pytest --cov=agentforge ``` ### 代码质量 ```bash # 代码格式化 black src/ isort src/ # 类型检查 mypy src/ # 代码检查 flake8 src/ ``` ## 🤝 贡献指南 1. Fork 项目 2. 创建特性分支 (`git checkout -b feature/amazing-feature`) 3. 提交更改 (`git commit -m 'Add amazing feature'`) 4. 推送到分支 (`git push origin feature/amazing-feature`) 5. 创建 Pull Request ## 📄 许可证 本项目采用 MIT 许可证 - 查看 [LICENSE](LICENSE) 文件了解详情。 ## 🙏 致谢 - 基于 [Aider](https://github.com/paul-gauthier/aider) 进行AI驱动的代码编辑 - 使用 [ChromaDB](https://www.trychroma.com/) 进行向量存储 - 集成 [Sentence Transformers](https://www.sbert.net/) 进行文本嵌入 - 设计灵感来源于认知科学中的记忆和反思机制 ## 📚 更多资源 - [API文档](docs/api.md) - [示例集合](examples/) - [最佳实践指南](docs/best_practices.md) - [常见问题](docs/faq.md) --- **FastAnts** - 让AI Agent协作更智能、更高效！