Initial commit: PRD v0.2, PROGRESS, knowledge base, PoC

poc/.gitignore

@@ -0,0 +1,4 @@
+node_modules/
+dist/
+.env
+*.log

poc/README.md

@@ -0,0 +1,66 @@
+# FlowPilot Agent Orchestration PoC
+
+This proof-of-concept demonstrates the core agent orchestration ideas from the FlowPilot PRD:
+
+## Core Concepts Demonstrated
+
+1. **HR Manager Agent**: Decomposes high-level tasks into atomic tasks and creates executor agents
+2. **Experience Manager Agent**: Logs executions, provides context, and maintains a knowledge base
+3. **Atomic Task Decomposition**: Breaking down complex tasks into single-agent-completable units
+4. **Agent Lifecycle Management**: Dynamic creation of executor agents for specific task types
+
+## How to Run
+
+```bash
+python3 agent_poc.py
+```
+
+## Output
+
+The PoC processes three sample high-level tasks from the project management handbook:
+- Creating a project charter
+- Identifying software development risks
+- Creating a project schedule
+
+For each task, it shows:
+- How the HR Manager decomposes the task into atomic tasks
+- How executor agents are created for each atomic task type
+- How the Experience Manager logs executions and provides context
+- The final results and execution logs
+
+## Next Steps for Development
+
+1. **Replace simulation with actual LLM calls** - Integrate with OpenAI/Anthropic/etc APIs
+2. **Add real task execution** - Instead of mock results, have agents actually perform tasks
+3. **Implement persistence** - Store agent configurations, execution logs, and knowledge base in database
+4. **Add feedback loops** - HR Manager improves based on execution scores
+5. **Integrate with Feishu** - Build the actual frontend and Feishu bot interfaces
+
+## Files
+
+- `agent_poc.py`: Main PoC implementation
+- `poc_results.json`: Detailed execution results from the last run
+- `.gitignore`: Git ignore file
+
+## Design Notes
+
+This PoC focuses on demonstrating the orchestration logic rather than actual AI execution.
+In a real implementation:
+- The `_simulate_execution` method would call actual LLM APIs
+- Agents would have specific prompt templates for their task types
+- The Experience Manager would use vector embeddings for context retrieval
+- Task decomposition would be more sophisticated (possibly using LLMs themselves)
+
+## Relation to PRD
+
+This PoC validates the core architectural concepts from PRD v0.2:
+- Dual agent management (HR Manager + Experience Manager)
+- Atomic task decomposition
+- Agent lifecycle management
+- Context sharing and knowledge base
+
+It does not yet cover:
+- Feishu integration
+- Full PMBOK process coverage
+- Production-level error handling and scaling
+- Actual LLM provider integration

poc/agent_poc.py

@@ -0,0 +1,381 @@
+#!/usr/bin/env python3
+"""
+Agent Orchestration PoC for FlowPilot
+Demonstrates:
+- HR Manager creating agents for tasks
+- Experience Manager logging and context sharing
+- Task decomposition into atomic types
+"""
+
+import uuid
+from datetime import datetime
+from enum import Enum
+from typing import Dict, List, Any, Optional
+
+
+class TaskType(Enum):
+    DOCUMENT_GENERATION = "document_generation"
+    ANALYSIS = "analysis"
+    COORDINATION = "coordination"
+    DATA_OPERATION = "data_operation"
+
+
+class AgentRole(Enum):
+    HR_MANAGER = "hr_manager"
+    EXPERIENCE_MANAGER = "experience_manager"
+    EXECUTOR = "executor"
+
+
+class TaskStatus(Enum):
+    PENDING = "pending"
+    ASSIGNED = "assigned"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class AtomicTask:
+    """Represents a task that can be completed by a single agent in one session."""
+    
+    def __init__(self, 
+                 task_type: TaskType,
+                 description: str,
+                 input_data: Dict[str, Any],
+                 expected_output_schema: Dict[str, Any]):
+        self.id = str(uuid.uuid4())
+        self.task_type = task_type
+        self.description = description
+        self.input_data = input_data
+        self.expected_output_schema = expected_output_schema
+        self.status = TaskStatus.PENDING
+        self.agent_id: Optional[str] = None
+        self.result: Optional[Dict[str, Any]] = None
+        self.created_at = datetime.now()
+        self.completed_at: Optional[datetime] = None
+    
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "task_type": self.task_type.value,
+            "description": self.description,
+            "input_data": self.input_data,
+            "expected_output_schema": self.expected_output_schema,
+            "status": self.status.value,
+            "agent_id": self.agent_id,
+            "result": self.result,
+            "created_at": self.created_at.isoformat(),
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None
+        }
+
+
+class Agent:
+    """Base agent class."""
+    
+    def __init__(self, agent_id: str, role: AgentRole, model: str = "default"):
+        self.agent_id = agent_id
+        self.role = role
+        self.model = model
+        self.capabilities: List[TaskType] = []
+    
+    def execute(self, task: AtomicTask) -> Dict[str, Any]:
+        """Execute a task and return result."""
+        raise NotImplementedError
+    
+    def can_handle(self, task_type: TaskType) -> bool:
+        return task_type in self.capabilities
+
+
+class HRManagerAgent(Agent):
+    """HR Manager: creates and manages executor agents."""
+    
+    def __init__(self, agent_id: str):
+        super().__init__(agent_id, AgentRole.HR_MANAGER)
+        # HR manager can handle task decomposition and agent creation
+        self.capabilities = [TaskType.DOCUMENT_GENERATION, TaskType.ANALYSIS]
+        self.available_models = {
+            TaskType.DOCUMENT_GENERATION: "claude-3-sonnet",
+            TaskType.ANALYSIS: "gpt-4-turbo",
+            TaskType.COORDINATION: "claude-3-haiku",
+            TaskType.DATA_OPERATION: "gpt-3.5-turbo"
+        }
+    
+    def decompose_task(self, high_level_task: str) -> List[AtomicTask]:
+        """Decompose a high-level task into atomic tasks.
+        In PoC, we'll use simple heuristics.
+        """
+        # Simple decomposition logic for demonstration
+        if "章程" in high_level_task or "charter" in high_level_task.lower():
+            return [
+                AtomicTask(
+                    task_type=TaskType.DOCUMENT_GENERATION,
+                    description="Generate project charter based on user input",
+                    input_data={"template": "project_charter", "user_input": high_level_task},
+                    expected_output_schema={"sections": ["目标", "范围", "里程碑", "干系人"]}
+                ),
+                AtomicTask(
+                    task_type=TaskType.ANALYSIS,
+                    description="Analyze stakeholders and create stakeholder register",
+                    input_data={"charter_placeholder": "will be filled"},
+                    expected_output_schema={"stakeholders": [{"name": "", "role": "", "power": 0, "interest": 0}]}
+                )
+            ]
+        elif "风险" in high_level_task or "risk" in high_level_task.lower():
+            return [
+                AtomicTask(
+                    task_type=TaskType.ANALYSIS,
+                    description="Identify project risks",
+                    input_data={"scope": high_level_task},
+                    expected_output_schema={"risks": [{"description": "", "probability": 0, "impact": 0}]}
+                ),
+                AtomicTask(
+                    task_type=TaskType.DOCUMENT_GENERATION,
+                    description="Generate risk mitigation plan",
+                    input_data={"risks_placeholder": "will be filled"},
+                    expected_output_schema={"mitigations": []}
+                )
+            ]
+        else:
+            # Default: treat as documentation task
+            return [
+                AtomicTask(
+                    task_type=TaskType.DOCUMENT_GENERATION,
+                    description=f"Generate document for: {high_level_task}",
+                    input_data={"request": high_level_task},
+                    expected_output_schema={"content": ""}
+                )
+            ]
+    
+    def create_executor_agent(self, task_type: TaskType) -> Agent:
+        """Create an executor agent for a specific task type."""
+        agent_id = str(uuid.uuid4())
+        # In reality, this would configure the agent with specific prompts and model
+        executor = Agent(agent_id, AgentRole.EXECUTOR, 
+                        model=self.available_models.get(task_type, "default"))
+        executor.capabilities = [task_type]
+        return executor
+    
+    def execute(self, task: AtomicTask) -> Dict[str, Any]:
+        """HR Manager handles task decomposition and agent creation."""
+        # For PoC, we simulate the decomposition
+        atomic_tasks = self.decompose_task(task.description)
+        task.agent_id = self.agent_id
+        task.status = TaskStatus.COMPLETED
+        task.result = {
+            "atomic_tasks": [at.to_dict() for at in atomic_tasks]
+        }
+        return task.result
+
+
+class ExperienceManagerAgent(Agent):
+    """Experience Manager: logs executions, provides context, manages knowledge."""
+    
+    def __init__(self, agent_id: str):
+        super().__init__(agent_id, AgentRole.EXPERIENCE_MANAGER)
+        self.capabilities = [TaskType.DATA_OPERATION, TaskType.ANALYSIS]
+        self.execution_log: List[Dict] = []
+        self.knowledge_base: Dict[str, Any] = {}
+    
+    def record_execution(self, agent_id: str, task: AtomicTask, result: Dict[str, Any]):
+        """Record an agent's task execution."""
+        log_entry = {
+            "timestamp": datetime.now().isoformat(),
+            "agent_id": agent_id,
+            "task_id": task.id,
+            "task_type": task.task_type.value,
+            "input": task.input_data,
+            "output": result,
+            "status": task.status.value
+        }
+        self.execution_log.append(log_entry)
+        
+        # Update knowledge base with patterns
+        self._update_knowledge_base(task, result)
+    
+    def _update_knowledge_base(self, task: AtomicTask, result: Dict[str, Any]):
+        """Simple knowledge base update: store successful patterns."""
+        if task.status == TaskStatus.COMPLETED:
+            key = f"{task.task_type.value}_{hash(task.description) % 1000}"
+            self.knowledge_base[key] = {
+                "task_description": task.description,
+                "successful_input_pattern": task.input_data,
+                "successful_output_pattern": result,
+                "count": self.knowledge_base.get(key, {}).get("count", 0) + 1
+            }
+    
+    def get_context_for_task(self, task_type: TaskType, keywords: str = "") -> Dict[str, Any]:
+        """Retrieve relevant context for a task."""
+        # Simple context retrieval: return recent similar executions
+        relevant_logs = [
+            log for log in self.execution_log[-20:]  # last 20
+            if log["task_type"] == task_type.value
+        ]
+        
+        return {
+            "recent_executions": relevant_logs,
+            "knowledge_base_snapshot": dict(list(self.knowledge_base.items())[:5])  # first 5
+        }
+    
+    def execute(self, task: AtomicTask) -> Dict[str, Any]:
+        """Experience Manager handles data operations and context queries."""
+        # For PoC, we simulate context provision
+        context = self.get_context_for_task(task.task_type, 
+                                          str(task.input_data))
+        task.agent_id = self.agent_id
+        task.status = TaskStatus.COMPLETED
+        task.result = {
+            "context": context,
+            "message": "Context provided for task execution"
+        }
+        return task.result
+
+
+class TaskOrchestrator:
+    """Orchestrates the workflow between HR Manager, Experience Manager, and executors."""
+    
+    def __init__(self):
+        self.hr_manager = HRManagerAgent("hr-001")
+        self.experience_manager = ExperienceManagerAgent("exp-001")
+        self.executor_agents: Dict[str, Agent] = {}
+        self.tasks: Dict[str, AtomicTask] = {}
+    
+    def process_high_level_task(self, description: str) -> Dict[str, Any]:
+        """Process a high-level task through the orchestrator."""
+        print(f"\n=== Processing High-Level Task: {description} ===")
+        
+        # Step 1: HR Manager decomposes the task
+        hr_task = AtomicTask(
+            task_type=TaskType.ANALYSIS,  # HR uses analysis to decompose
+            description=description,
+            input_data={},
+            expected_output_schema={"atomic_tasks": []}
+        )
+        
+        decomposition_result = self.hr_manager.execute(hr_task)
+        self.experience_manager.record_execution(self.hr_manager.agent_id, hr_task, decomposition_result)
+        
+        atomic_tasks_data = decomposition_result.get("atomic_tasks", [])
+        atomic_tasks = []
+        
+        # Reconstruct AtomicTask objects from decomposition result
+        for task_data in atomic_tasks_data:
+            task = AtomicTask(
+                task_type=TaskType(task_data["task_type"]),
+                description=task_data["description"],
+                input_data=task_data["input_data"],
+                expected_output_schema=task_data["expected_output_schema"]
+            )
+            task.id = task_data["id"]
+            atomic_tasks.append(task)
+            self.tasks[task.id] = task
+        
+        print(f"HR Manager decomposed into {len(atomic_tasks)} atomic tasks")
+        
+        # Step 2: For each atomic task, create/execute agent
+        results = []
+        for task in atomic_tasks:
+            print(f"\n--- Executing Task: {task.description} ---")
+            
+            # Get context from Experience Manager
+            context_task = AtomicTask(
+                task_type=TaskType.DATA_OPERATION,
+                description="Get context for task execution",
+                input_data={"task_type": task.task_type.value, "keywords": task.description},
+                expected_output_schema={"context": {}}
+            )
+            context_result = self.experience_manager.execute(context_task)
+            self.experience_manager.record_execution(self.experience_manager.agent_id, context_task, context_result)
+            
+            # Create executor agent for this task type
+            executor_id = str(uuid.uuid4())
+            executor = self.hr_manager.create_executor_agent(task.task_type)
+            executor.agent_id = executor_id  # Override with generated ID
+            self.executor_agents[executor_id] = executor
+            
+            # Simulate task execution (in PoC, we just generate a mock result)
+            # In reality, this would call the actual agent/LLM
+            mock_result = self._simulate_execution(executor, task, context_result["context"])
+            
+            # Record execution
+            task.agent_id = executor_id
+            task.status = TaskStatus.COMPLETED
+            task.result = mock_result
+            task.completed_at = datetime.now()
+            
+            self.experience_manager.record_execution(executor_id, task, mock_result)
+            results.append({
+                "task": task.to_dict(),
+                "executor_id": executor_id,
+                "result": mock_result
+            })
+            
+            print(f"Task completed by executor {executor_id[:8]}")
+        
+        return {
+            "high_level_task": description,
+            "hr_manager_task": hr_task.to_dict(),
+            "atomic_tasks": [t.to_dict() for t in atomic_tasks],
+            "execution_results": results,
+            "experience_manager_log": self.experience_manager.execution_log[-len(atomic_tasks)*2:]  # recent logs
+        }
+    
+    def _simulate_execution(self, executor: Agent, task: AtomicTask, context: Dict) -> Dict[str, Any]:
+        """Simulate agent execution. In real system, this would call LLM."""
+        # Simple mock based on task type
+        if task.task_type == TaskType.DOCUMENT_GENERATION:
+            return {
+                "generated_document": f"# Sample Document\n\nThis is a generated document for: {task.description}\n\nSections: {list(task.expected_output_schema.keys())}",
+                "model_used": executor.model,
+                "tokens_used": 1500
+            }
+        elif task.task_type == TaskType.ANALYSIS:
+            return {
+                "analysis_result": f"Analysis of: {task.description}\n\nKey findings: \n- Item 1: ...\n- Item 2: ...\n- Recommendation: ...",
+                "model_used": executor.model,
+                "tokens_used": 800
+            }
+        else:
+            return {
+                "output": f"Executed {task.task_type.value} for: {task.description}",
+                "model_used": executor.model,
+                "tokens_used": 500
+            }
+
+
+def main():
+    """Run the PoC demonstration."""
+    print("FlowPilot Agent Orchestration PoC")
+    print("=" * 50)
+    
+    orchestrator = TaskOrchestrator()
+    
+    # Example high-level tasks from the project management handbook
+    tasks_to_process = [
+        "创建项目章程，目标是开发一个新的客户管理系统，范围包括需求、开发、测试和上线",
+        "识别软件开发项目的主要风险",
+        "制定项目进度计划"
+    ]
+    
+    all_results = []
+    for task_desc in tasks_to_process:
+        result = orchestrator.process_high_level_task(task_desc)
+        all_results.append(result)
+        print("\n" + "="*50)
+    
+    # Summary
+    print("\n=== PoC Execution Summary ===")
+    print(f"High-level tasks processed: len {len(all_results)}")
+    total_atomic = sum(len(r["atomic_tasks"]) for r in all_results)
+    print(f"Total atomic tasks executed: {total_atomic}")
+    print(f"HR Manager agent ID: {orchestrator.hr_manager.agent_id}")
+    print(f"Experience Manager agent ID: {orchestrator.experience_manager.agent_id}")
+    print(f"Executor agents created: {len(orchestrator.executor_agents)}")
+    
+    # Save results to file for inspection
+    import json
+    with open("poc_results.json", "w") as f:
+        json.dump(all_results, f, indent=2, ensure_ascii=False)
+    print("\nDetailed results saved to poc_results.json")
+
+
+if __name__ == "__main__":
+    main()