daironpf/socialseed-tasker
GitHub: daironpf/socialseed-tasker
基于Neo4j图数据库的AI原生任务管理框架，为自主软件工程智能体提供任务追踪、依赖治理和协作看板能力。
Stars: 0 | Forks: 0
# SocialSeed Tasker
## 🔭 图原生工程与自主智能体治理
一个专门利用 **Neo4j** 的框架，为 AI 智能体提供无限的架构上下文和严格治理。
",
"labels": ["bug", "security"]
}'
```
**优先级值**：`LOW`、`MEDIUM`、`HIGH`、`CRITICAL`
#### 列出问题
```
# 所有问题
curl http://localhost:8000/api/v1/issues
# 按状态筛选
curl "http://localhost:8000/api/v1/issues?status=OPEN"
# 按组件筛选
curl "http://localhost:8000/api/v1/issues?component="
```
#### 更新问题
```
# 更新状态
curl -X PATCH http://localhost:8000/api/v1/issues/ \
-H "Content-Type: application/json" \
-d '{"status": "IN_PROGRESS"}'
# 标记 AI 代理正在处理此问题
curl -X PATCH http://localhost:8000/api/v1/issues/ \
-H "Content-Type: application/json" \
-d '{"agent_working": true}'
# 更新优先级
curl -X PATCH http://localhost:8000/api/v1/issues/ \
-H "Content-Type: application/json" \
-d '{"priority": "CRITICAL"}'
# 更新描述
curl -X PATCH http://localhost:8000/api/v1/issues/ \
-H "Content-Type: application/json" \
-d '{"description": "Updated description"}'
# 关闭问题
curl -X POST http://localhost:8000/api/v1/issues//close
```
**状态值**：`OPEN`、`IN_PROGRESS`、`BLOCKED`、`CLOSED`
#### 删除问题
```
curl -X DELETE http://localhost:8000/api/v1/issues/
```
### 依赖关系
依赖关系定义哪些问题阻塞其他问题。AI 智能体利用这些信息来确定可以处理的工作。
#### 添加依赖关系
```
# 问题 A 依赖于问题 B（必须先完成 B）
curl -X POST http://localhost:8000/api/v1/issues//dependencies \
-H "Content-Type: application/json" \
-d '{"depends_on_id": ""}'
```
#### 列出依赖关系
```
# 此问题依赖于什么？
curl http://localhost:8000/api/v1/issues//dependencies
```
#### 移除依赖关系
```
curl -X DELETE http://localhost:8000/api/v1/issues//dependencies/
```
### 智能体工作指示器
AI 智能体可以在问题上设置 `agent_working: true` 来表示他们正在积极处理该问题。这会在看板上显示一个青色机器人图标。
```
import requests
# 告知系统您正在处理此问题
requests.patch(
"http://localhost:8000/api/v1/issues/",
json={"agent_working": True}
)
# 完成后清除标记
requests.patch(
"http://localhost:8000/api/v1/issues/",
json={"agent_working": False}
)
```
## 🤖 AI 智能体工作流
### AI 智能体推荐工作流
```
import requests
from datetime import datetime
API_BASE = "http://localhost:8000/api/v1"
def start_working_on_issue(issue_id, todo_items):
"""AI agent starts working on an issue - updates status and sets todo."""
# 1. Create a detailed TODO list in the description
todo_text = "## TODO:\n" + "\n".join([f"- [ ] {item}" for item in todo_items])
todo_text += f"\n\n## Progress (started {datetime.now().strftime('%Y-%m-%d %H:%M')}):\n"
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": todo_text,
"agent_working": True,
"status": "IN_PROGRESS"
})
def update_progress(issue_id, completed_item, next_step):
"""Update progress on the issue."""
# Get current description
issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
desc = issue.get("description", "")
# Mark completed item
desc = desc.replace(f"- [ ] {completed_item}", f"- [x] {completed_item}")
# Add progress note
desc += f"\n- **In progress**: {next_step}"
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": desc
})
def finish_issue(issue_id, solution_summary):
"""Mark issue as completed with solution summary."""
# Get current description
issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
desc = issue.get("description", "")
# Add solution summary
desc += f"\n\n## Solution:\n{solution_summary}"
# Close the issue
requests.post(f"{API_BASE}/issues/{issue_id}/close")
# Clear agent working flag
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": desc,
"agent_working": False
})
```
def create_issue_and_work(title, component_id, description=""):
"""创建一个问题并开始处理。"""
```
# 1. 创建问题
response = requests.post(f"{API_BASE}/issues", json={
"title": title,
"description": description,
"component_id": component_id,
"priority": "MEDIUM"
})
issue = response.json()["data"]
issue_id = issue["id"]
# 2. 标记为代理正在处理（在看板上显示）
requests.patch(f"{API_BASE}/{issue_id}", json={"agent_working": True})
# 3. 移至进行中
requests.patch(f"{API_BASE}/{issue_id}", json={"status": "IN_PROGRESS"})
# 4. 执行工作...
# 5. 标记为完成
requests.post(f"{API_BASE}/{issue_id}/close")
# 6. 清除代理工作标记
requests.patch(f"{API_BASE}/{issue_id}", json={"agent_working": False})
return issue_id
```
```
### 完整示例：AI 代理解决问题
```python
import requests
API_BASE = "http://localhost:8000/api/v1"
def solve_issue(issue_id, problem_description):
"""
AI agent solves an issue, keeping the board updated with progress.
"""
# Step 1: Analyze and plan
todo_items = [
"Analyze the problem and identify root cause",
"Write test to reproduce the issue",
"Implement the fix",
"Run tests to verify the solution",
"Update documentation if needed"
]
# Step 2: Start working - update status and add TODO to description
initial_desc = f"## Problem\n{problem_description}\n\n"
initial_desc += "## TODO:\n" + "\n".join([f"- [ ] {item}" for item in todo_items])
initial_desc += f"\n\n## Started at: {datetime.now().isoformat()}"
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": initial_desc,
"status": "IN_PROGRESS",
"agent_working": True
})
# Step 3: First TODO complete - Analyze
update_issue_description(issue_id, todo_items[0], "Analyzing root cause...")
# Step 4: Found the issue - update with findings
update_issue_description(issue_id, todo_items[0],
"Root cause: Missing null check in auth handler")
# Step 5: Write test
update_issue_description(issue_id, todo_items[1],
"Test written: test_login_with_special_chars")
# Step 6: Implement fix
update_issue_description(issue_id, todo_items[2],
"Fix implemented: Added null validation")
# Step 7: Tests pass
update_issue_description(issue_id, todo_items[3],
"All 45 tests pass")
# Step 8: Close issue with summary
solution_summary = """
## 解决方案已应用
- Added null validation for password field
- Added test case with special characters: !@#$%^&*()
- All existing tests continue to pass
## 文件已更改
- src/auth/handlers.py (line 45-47)
- tests/test_auth.py (added test_login_with_special_chars)
"""
requests.post(f"{API_BASE}/issues/{issue_id}/close")
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": initial_desc + solution_summary,
"agent_working": False
})
def update_issue_description(issue_id, completed_item, next_action):
"""Helper to mark a TODO complete and note next action."""
# Get current description
issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
desc = issue.get("description", "")
# Mark completed item
desc = desc.replace(f"- [ ] {completed_item}", f"- [x] {completed_item}")
# Add progress note with timestamp
timestamp = datetime.now().strftime("%H:%M")
desc += f"\n[{timestamp}] **Next**: {next_action}"
requests.patch(f"{API_BASE}/issues/{issue_id}", json={
"description": desc
})
```
### 为什么保持看板更新？
| 原因 | 描述 |
|--------|-------------|
| **透明度** | 人类可以看到 AI 正在做什么 |
| **可追溯性** | 解决方案的完整历史记录在问题中 |
| **协作** | 其他智能体或人类可以看到进度 |
| **调试** | 如果出现问题，轨迹是清晰的 |
| **状态同步** | 看板始终反映实际情况 |
def get_workable_issues():
"""获取可以处理的问题（未被阻塞的）。"""
all_issues = requests.get(f"{API_BASE}/issues").json()["data"]["items"]
```
workable = []
for issue in all_issues:
if issue["status"] == "CLOSED":
continue
# Check dependencies - if all dependencies are closed, it's workable
deps = requests.get(f"{API_BASE}/issues/{issue['id']}/dependencies").json()["data"]
if not deps or all(d["status"] == "CLOSED" for d in deps):
workable.append(issue)
return workable
```
```
---
## 📋 完整示例：完整工作流程
```bash
# 1. 创建一个组件
COMPONENT=$(curl -s -X POST http://localhost:8000/api/v1/components \
-H "Content-Type: application/json" \
-d '{"name": "api-service", "project": "myapp"}' | jq -r '.data.id')
echo "Created component: $COMPONENT"
# 2. 创建多个问题
ISSUE1=$(curl -s -X POST http://localhost:8000/api/v1/issues \
-H "Content-Type: application/json" \
-d "{\"title\": \"Setup database schema\", \"component_id\": \"$COMPONENT\", \"priority\": \"HIGH\"}" | jq -r '.data.id')
ISSUE2=$(curl -s -X POST http://localhost:8000/api/v1/issues \
-H "Content-Type: application/json" \
-d "{\"title\": \"Create API endpoints\", \"component_id\": \"$COMPONENT\", \"priority\": \"HIGH\"}" | jq -r '.data.id')
# 3. 添加依赖关系：API 端点依赖于数据库架构
curl -X POST "http://localhost:8000/api/v1/issues/$ISSUE2/dependencies" \
-H "Content-Type: application/json" \
-d "{\"depends_on_id\": \"$ISSUE1\"}"
# 4. 标记 AI 正在处理数据库架构
curl -X PATCH "http://localhost:8000/api/v1/issues/$ISSUE1" \
-H "Content-Type: application/json" \
-d '{"agent_working": true, "status": "IN_PROGRESS"}'
# 5. 完成数据库后，关闭它并处理 API
curl -X POST "http://localhost:8000/api/v1/issues/$ISSUE1/close"
curl -X PATCH "http://localhost:8000/api/v1/issues/$ISSUE2" \
-H "Content-Type: application/json" \
-d '{"agent_working": true, "status": "IN_PROGRESS"}'
# 6. 检查所有问题
curl http://localhost:8000/api/v1/issues | jq '.data.items[] | {title, status, agent_working, dependencies}'
```
## 🔍 查找可处理的问题
AI 智能体可以查询以找到可以处理的问题：
```
# 获取所有未解决的问题
curl "http://localhost:8000/api/v1/issues?status=OPEN" | jq '.data.items[] | .title'
# 对于每个问题，检查依赖关系是否已解决
curl "http://localhost:8000/api/v1/issues//dependencies"
```
## 🧠 GraphRAG：AI 智能体的无限上下文
Tasker 图不仅仅用于跟踪问题——它是一个 **知识图谱**，在提出解决方案之前为 AI 智能体提供无限的上下文。
### 工作原理
1. **依赖关系跟踪**：每个问题关系都存储为 Neo4j 中的有向边
2. **影响分析**：在处理问题之前，智能体可以查询完整的依赖关系链以了解下游影响
3. **根因发现**：已关闭的问题与测试失败相关联，支持自动根因分析
### 示例：修复前的影响分析
```
# 在重构 Auth 模块之前，检查将受影响的内容
import requests
issue_id = ""
impact = requests.get(f"{API_BASE}/analyze/impact/{issue_id}").json()["data"]
print(f"Directly affected: {len(impact['directly_affected'])} issues")
print(f"Transitively affected: {len(impact['transitively_affected'])} issues")
print(f"Risk level: {impact['risk_level']}")
```
这使智能体能够根据完整的架构上下文（而不仅仅是孤立的问题）做出 **明智的决策**。
## 🔧 环境变量
| 变量 | 默认值 | 描述 |
|----------|---------|-------------|
| `TASKER_NEO4J_URI` | `bolt://localhost:7687` | Neo4j 连接 URI |
| `TASKER_NEO4J_USER` | `neo4j` | Neo4j 用户名 |
| `TASKER_NEO4J_PASSWORD` | （无） | Neo4j 密码（必填）|
| `API_PORT` | `8000` | API 服务器端口 |
## 🐳 Docker Compose
包含的 `docker-compose.yml` 启动：
- **Neo4j**（端口 7474/7687）- 图数据库
- **API**（端口 8000）- 供 AI 智能体使用的 REST API
- **前端**（端口 8080）- 人类看板
```
# 启动所有服务
docker compose up -d
# 查看日志
docker compose logs -f
# 停止所有服务
docker compose down
```
## 📊 架构
```
┌──────────────────────────┐
│ AI Agent / Human UI │
│ REST API (port 8000) │
└────────────┬─────────────┘
▼
┌──────────────────────────────┐
│ Application Core │
│ (Hexagonal Architecture) │
│ • Governance Engine │
│ • Dependency BFS Analysis │
│ • Root Cause Detection │
└────────────┬─────────────────┘
▼
┌──────────────────────────────┐
│ Neo4j Graph DB │
│ (The Source of Truth) │
│ • Relationship Tracking │
│ • Causal Traceability │
└──────────────────────────────┘
```
## 🔗 相关文档
- **[CLI 参考](#)** - 命令行界面
- **[API 端点](#)** - 详细 API 文档
- **[开发](#)** - 运行测试、贡献代码
标签：AI代理治理, AI原生, Apache 2.0, DNS解析, Docker, GraphRAG, Neo4j, Python, REST API, Web框架, 任务管理, 六边形架构, 后端开发, 图原生, 安全防御评估, 开源项目, 数据管道, 无后门, 网络测绘, 自主代理, 请求拦截, 软件工程