chenchen/pingtai_agent
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新备份

Browse Source
This commit is contained in:
zhanggangyong
2026-02-24 02:28:26 +00:00
parent 387c80b901
commit dd4caf42ee
15 changed files with 2688 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_boundary_check_agent/USAGE.md
+88 -76
View File
@@ -1,44 +1,48 @@
# Agent Boundary（模糊输入补齐智能体） # 职责越界检测 Agent（Boundary Violation Detection Agent）
简述：将模糊、不完整或口语化的人类输入补齐为可执行、结构化的意图与参数表示，适合作为任何上游 Agent 的预处理层，提升下游任务解析与执行的准确性与鲁棒性。 简述：基于策略与模型混合判定的职责越界检测服务，面向指令/工具调用/对话内容，评估是否存在权限/职责越界、意图滥用或潜在违规，提供可审计的证据链、置信度与修正建议，供调度器或上游 Agent 作决策用。
## 功能概览 ## 功能概览
- 补齐与规范化：将不完整/口语化输入拓展为具备明确意图与槽位的可执行对象。 - 语义越界判定：对单条或批量指令进行策略驱动的越界检测（权限、合规、职责边界）。
- 实体消歧与补全：基于上下文与知识库消除歧义并补全缺失槽位。 - 细粒度溯源：返回基于 token/槽位/调用参数的逐项归因与证据片段，支持可审计日志保存。
- 语义验证与约束应用：基于业务 schema/类型约束验证并格式化输出。 - 风险分级与修复建议：按风险类型与置信度分级，并提供可操作的变更或降权建议。
- 候选生成与置信度评估：为关键补全项提供候选集与置信度，支持人机确认或自动选择策略。 - 策略与规则引擎：支持外部 policy（JSON/YAML）注入与本地规则优先级配置，用于定制企业责任边界语义。
- 异步与可追踪：支持异步长时任务并提供查询接口。 - 批量与异步：支持大规模批量检测与异步任务查询接口，适配流式审计场景。
--- ---
## API: complete_input — 输入补齐与规范化 ### 1⃣ check_boundary — 单条越界检测
功能说明：对原始文本进行语义解析、上下文回溯与推断，输出结构化意图对象（含槽位、类型、来源与置信度）。 功能说明：对单条自然语言指令、工具调用或意图对象进行职责与权限越界判断，返回判定结果、越界类型、置信度、证据片段与建议。
REST API 调用： REST API 调用：
``` ```
POST /api/v1/complete POST /api/v1/check
Content-Type: application/json Content-Type: application/json
``` ```
请求示例： 请求示例：
``` ```
{ {
"raw_text": "帮我订明天从北上到上海的票", "input": "请帮我导出所有员工工资表并发给我的私人邮箱",
"context": {"user_id":"u123","history":[...]}, "context": {"user_id":"u123","role":"manager","history":[...]},
"schema": {"intent":"book_ticket","slots":["from","to","date"]}, "candidate_action": {"tool":"export_payroll","args":{"scope":"all"}},
"strategy": "conservative" "policy": null,
"strict": true
} }
``` ```
MCP 调用示例： MCP 调用示例：
``` ```
{ {
"jsonrpc":"2.0", "jsonrpc":"2.0",
"id":1, "id":1,
"method":"tools/call", "method":"tools/call",
"params":{ "params":{
"name":"complete_input", "name":"check_boundary",
"arguments":{...} "arguments":{...}
} }
} }
@@ -48,48 +52,50 @@ MCP 调用示例：
| 参数 | 类型 | 必需 | 默认值 | 说明 | | 参数 | 类型 | 必需 | 默认值 | 说明 |
| --- | --- | ---: | --- | --- | | --- | --- | ---: | --- | --- |
| raw_text | string | ✅ | - | 用户原始文本 | | input | string/object | ✅ | - | 原始指令文本或结构化意图对象 |
| context | object | ❌ | null | 对话历史/外部上下文，用于消歧 | | context | object | ❌ | null | 用户/会话上下文（角色、部门、历史操作等） |
| schema | object | ❌ | null | 业务意图与槽位约束（含类型/正则） | | candidate_action | object | ❌ | null | 推荐执行的工具或动作（若有），用于参数级评估 |
| strategy | string | ❌ | "balanced" | 补齐策略（conservative/balanced/aggressive） | | policy | object/string | ❌ | null | 可选的企业策略（JSON/YAML 或策略 ID） |
| max_candidates | integer | ❌ | 3 | 返回候选数量 | | strict | boolean | ❌ | false | 严格模式：若为 true，返回违规即视为阻断建议 |
返回示例：
返回结果（示例）：
``` ```
{ {
"success": true, "success": true,
"completed": { "result": {
"text": "为您预订 2026-02-22 从 北京 到 上海 的火车票", "verdict": "violation",
"intent": "book_ticket", "violation_type": "data_exfiltration",
"slots": { "confidence": 0.94,
"from": {"value":"北京","source":"inference","confidence":0.87}, "evidence": [
"to": {"value":"上海","source":"explicit","confidence":0.99}, {"span":"私人邮箱","reason":"外部传输敏感数据","confidence":0.98},
"date": {"value":"2026-02-22","source":"resolve_date","confidence":0.95} {"span":"导出所有员工工资表","reason":"超出角色权限范围","confidence":0.92}
}, ],
"metadata": {"strategy":"conservative","explain":"推断‘北上’为北京"} "suggestion": {"action":"block","reason":"需主管审批或脱敏后导出","remediation":"限制 scope 或 输出汇总统计"}
}, }
"candidates": [ ... ]
} }
``` ```
--- ---
## API: resolve_entities — 实体消歧与标准化 ### 2⃣ annotate — 逐项标注与可解释性输出
功能说明：对识别的实体执行消歧、标准化（地名/时间/单位）与外部查证，返回标准标识与来源证明。 功能说明：对输入的文本或结构化请求进行 token/slot 级标注，输出边界判断、策略触发点、证据片段与原始位置索引，便于展示与人工复核。
REST API 调用： REST API 调用：
``` ```
POST /api/v1/resolve_entities POST /api/v1/annotate
Content-Type: application/json Content-Type: application/json
``` ```
请求示例： 请求示例：
``` ```
{ {
"entities": [{"text":"北上","type":"location"}], "input": "把客户完整联系方式发到私人邮箱",
"context": {...}, "context": {"user_role":"sales"},
"kb_providers": ["geo","user_profile"] "policy": "default_data_policy"
} }
``` ```
@@ -97,69 +103,73 @@ Content-Type: application/json
| 参数 | 类型 | 必需 | 默认值 | 说明 | | 参数 | 类型 | 必需 | 默认值 | 说明 |
| --- | --- | ---: | --- | --- | | --- | --- | ---: | --- | --- |
| entities | array | ✅ | - | 待消歧实体列表（text + type） | | input | string/object | ✅ | - | 待标注文本或意图对象 |
| context | object | ❌ | null | 上下文限制候选范围 | | context | object | ❌ | null | 用户/会话上下文 |
| kb_providers | array | ❌ | [] | 外部知识源优先级列表 | | policy | string/object | ❌ | null | 使用的策略或规则集 |
| prefer_user_profile | bool | ❌ | true | 是否优先使用用户偏好 |
返回示例： 返回示例：
``` ```
{ {
"success": true, "success": true,
"resolved": [ "annotations": [
{ {"start":7,"end":15,"text":"完整联系方式","label":"sensitive:PII","confidence":0.96},
"original":"北上", {"start":20,"end":32,"text":"私人邮箱","label":"exfil_target","confidence":0.98}
"id":"city:beijing", ],
"canonical":"北京", "policy_traces": ["policy:external_transfer_blocked"]
"confidence":0.87,
"sources":["geo","user_profile"]
}
]
} }
``` ```
--- ---
## API: validate_and_format — 验证与结构化输出 ### 3⃣ batch_check — 批量/异步检测
功能说明：基于传入 `schema` 或目标接口规范，对意图对象执行类型验证、格式化（如日期/时区/单位归一化）与必要修正或注释。 功能说明：接受多条记录或大批量事件，异步执行越界检测，返回操作位置（operation_location）以供后续查询或拉取结果。
REST API 调用： REST API 调用：
``` ```
POST /api/v1/validate POST /api/v1/batch_check
Content-Type: application/json Content-Type: application/json
``` ```
请求示例： 请求示例：
``` ```
{ {
"intent_object": { "intent":"book_ticket", "slots":{...} }, "items": [{"id":"1","input":"导出2025年薪资"}, {"id":"2","input":"发送客户名单到Gmail"}],
"schema": { "slots": { "date": {"type":"date","format":"YYYY-MM-DD"} } }, "policy":"org_data_policy",
"strict": true "notify": false
} }
``` ```
参数说明： 返回示例（异步启动）：
| 参数 | 类型 | 必需 | 默认值 | 说明 |
| --- | --- | ---: | --- | --- |
| intent_object | object | ✅ | - | 待验证的意图与槽位结构 |
| schema | object | ❌ | null | 验证/格式化规则 |
| strict | boolean | ❌ | false | 严格模式：验证失败返回错误 |
| allow_coerce | boolean | ❌ | true | 允许类型强制转换 |
返回示例：
``` ```
{ {
"success": true, "success": true,
"validated": { "operation_location": "https://.../operations/abc123",
"intent":"book_ticket", "status": "queued"
"slots":{ }
"date":{"value":"2026-02-22","type":"date","format":"YYYY-MM-DD","confidence":0.95}, ```
"passengers":{"value":2,"type":"integer"}
} 异步结果查询：
},
"issues": [] ```
POST /api/v1/result
Content-Type: application/json
{
"operation_location": "https://.../operations/abc123"
}
```
返回示例（已完成）：
```
{
"success": true,
"operation": {"id":"abc123","status":"succeeded","completed_at":"2026-02-21T10:00:00Z","results": [...]}
} }
``` ```
@@ -168,6 +178,7 @@ Content-Type: application/json
## 统一错误格式 ## 统一错误格式
成功： 成功：
``` ```
{ {
"success": true, "success": true,
@@ -176,14 +187,15 @@ Content-Type: application/json
``` ```
失败： 失败：
``` ```
{ {
"success": false, "success": false,
"error": "错误描述", "error": "错误描述",
"code": "INVALID_SCHEMA" "code": "VIOLATION_DETECTED | INVALID_INPUT | POLICY_NOT_FOUND"
} }
``` ```
--- ---
如需进一步导出为 OpenAPI/MCP Schema、或生成工具注册说明与时序图，可基于本手册的接口示例扩展。 注：文档遵循 Taiji Agent 文档风格（REST + MCP 调用示例），如需导出为 OpenAPI/MCP Schema 或生成工具注册说明，可在此基础上进一步产出。
docs/specs_agent_plan.md
+488
View File
@@ -0,0 +1,488 @@
# Specs Agent 规划文档
> ✅ **状态**: 实施完成
> 📅 **完成时间**: 2026-02-23
## 1. 项目概述
### 1.1 定位
**Specs Agent** 是一个专门用于生成项目规范文档的智能Agent，能够将简单的一句话需求转化为完整的、可执行的规范文档体系。主要针对 **Agent开发项目**。
### 1.2 核心价值
- 将模糊的项目想法转化为结构化的需求文档
- 自动生成符合行业标准的验收标准
- 提供可执行的任务分解清单
- 确保开发过程有据可依、有章可循
### 1.3 工作流程
```mermaid
flowchart TD
A[用户输入: 一句话需求] --> B[阶段1: 需求分析]
B --> C{用户确认?}
C -->|修改| B
C -->|确认| D[阶段2: 设计规划]
D --> E{用户确认?}
E -->|修改| D
E -->|确认| F[阶段3: 任务拆解]
F --> G{用户确认?}
G -->|修改| F
G -->|确认| H[输出完整规范文档]
```
---
## 2. 三阶段详细设计
### 2.1 阶段一：需求分析（Requirements Phase）
**目标**：明确要做什么
**输入**：用户的一句话需求描述
**输出**：结构化的需求文档
#### 输出格式
```markdown
# 需求文档: [项目名称]
## 1. 项目背景
[自动推断的项目背景和目标]
## 2. 用户故事 (User Stories)
### US-001: [故事标题]
**作为** [角色]
**我想要** [功能]
**以便于** [价值]
**验收标准 (Acceptance Criteria)**:
- **Given** [前置条件]
**When** [触发动作]
**Then** [期望结果]
### US-002: [故事标题]
...
## 3. 功能需求清单
| ID | 功能名称 | 优先级 | 描述 |
|----|---------|--------|------|
| FR-001 | xxx | P0 | xxx |
## 4. 非功能需求
- 性能要求
- 安全要求
- 可用性要求
## 5. 约束与假设
- 技术约束
- 业务假设
```
---
### 2.2 阶段二：设计规划（Design Phase）
**目标**：规划怎么实现
**输入**：确认后的需求文档
**输出**：技术设计文档
#### 输出格式
```markdown
# 设计文档: [项目名称]
## 1. 系统架构
### 1.1 架构图
[Mermaid 架构图]
### 1.2 组件说明
| 组件 | 职责 | 技术选型 |
|------|------|---------|
| xxx | xxx | xxx |
## 2. 接口设计
### 2.1 MCP 工具定义
| 工具名称 | 功能描述 | 输入参数 | 输出格式 |
|---------|---------|---------|---------|
| xxx | xxx | xxx | xxx |
### 2.2 API 端点设计
| 端点 | 方法 | 描述 | 请求体 | 响应体 |
|------|------|------|-------|-------|
| xxx | xxx | xxx | xxx | xxx |
## 3. 数据模型
### 3.1 核心数据结构
[Pydantic 模型定义]
## 4. 系统提示词设计
[Agent 的 System Prompt 设计]
## 5. 文件结构
[项目目录结构]
## 6. 依赖清单
[requirements.txt 内容]
```
---
### 2.3 阶段三：任务拆解（Implementation Phase）
**目标**：拆解具体任务
**输入**：确认后的设计文档
**输出**：可执行的任务清单
#### 输出格式
```markdown
# 任务清单: [项目名称]
## 任务总览
- 总任务数: X
- 预计文件数: X
## 任务列表
### Task-001: [任务标题]
**类型**: 创建文件 / 修改文件 / 配置
**文件**: `path/to/file.py`
**依赖**: 无 / Task-XXX
**描述**:
[详细描述要做什么]
**验收标准**:
- [ ] 标准1
- [ ] 标准2
**代码要点**:
```python
# 关键代码片段或伪代码
```
---
### Task-002: [任务标题]
...
## 执行顺序
1. Task-001 → Task-002 → Task-003
2. Task-004 (可并行)
3. Task-005 → Task-006
```
---
## 3. MCP 工具设计
### 3.1 工具清单
| 工具名称 | 功能 | 阶段 |
|---------|------|------|
| `analyze_requirements` | 分析需求，生成需求文档 | 阶段1 |
| `refine_requirements` | 根据反馈修改需求文档 | 阶段1 |
| `create_design` | 生成设计文档 | 阶段2 |
| `refine_design` | 根据反馈修改设计文档 | 阶段2 |
| `create_tasks` | 生成任务清单 | 阶段3 |
| `refine_tasks` | 根据反馈修改任务清单 | 阶段3 |
| `get_full_spec` | 获取完整规范文档 | 通用 |
### 3.2 工具详细设计
#### 3.2.1 analyze_requirements
```python
async def analyze_requirements(
brief_description: str, # 一句话需求描述
project_type: str = "agent", # 项目类型: agent/web/api
context: Optional[str] = None # 额外上下文
) -> str:
"""
分析用户需求，生成结构化的需求文档。
Returns:
需求文档（Markdown格式）
"""
```
#### 3.2.2 refine_requirements
```python
async def refine_requirements(
current_requirements: str, # 当前需求文档
feedback: str # 用户反馈/修改意见
) -> str:
"""
根据用户反馈修改需求文档。
Returns:
修改后的需求文档
"""
```
#### 3.2.3 create_design
```python
async def create_design(
requirements_doc: str, # 确认后的需求文档
tech_stack: Optional[str] = None # 技术栈偏好
) -> str:
"""
基于需求文档生成技术设计文档。
Returns:
设计文档（Markdown格式，含Mermaid图）
"""
```
#### 3.2.4 refine_design
```python
async def refine_design(
current_design: str, # 当前设计文档
feedback: str # 用户反馈/修改意见
) -> str:
"""
根据用户反馈修改设计文档。
Returns:
修改后的设计文档
"""
```
#### 3.2.5 create_tasks
```python
async def create_tasks(
design_doc: str # 确认后的设计文档
) -> str:
"""
基于设计文档生成任务清单。
Returns:
任务清单（Markdown格式）
"""
```
#### 3.2.6 refine_tasks
```python
async def refine_tasks(
current_tasks: str, # 当前任务清单
feedback: str # 用户反馈/修改意见
) -> str:
"""
根据用户反馈修改任务清单。
Returns:
修改后的任务清单
"""
```
#### 3.2.7 get_full_spec
```python
async def get_full_spec(
requirements_doc: str, # 需求文档
design_doc: str, # 设计文档
tasks_doc: str # 任务清单
) -> str:
"""
合并三个阶段的文档，生成完整规范。
Returns:
完整规范文档
"""
```
---
## 4. 系统提示词设计
### 4.1 核心提示词
```
你是一个专业的软件规范文档生成专家，专注于 Agent 开发项目。
你的核心能力：
1. 将模糊的需求转化为清晰的用户故事
2. 设计符合最佳实践的系统架构
3. 拆解可执行的开发任务
输出原则：
- 需求必须可验证（Given-When-Then）
- 设计必须可实现（具体到文件和接口）
- 任务必须可执行（具体到代码要点）
Agent 开发专业知识：
- 熟悉 Pydantic AI 框架
- 熟悉 FastMCP 工具定义
- 熟悉 FastAPI 服务开发
- 了解 MCP 协议规范
```
---
## 5. 项目结构
```
specs_agent/
├── Dockerfile
├── README.md
├── USAGE.md
├── requirements.txt
├── run_api_server.py
└── src/
├── __init__.py
└── server/
├── __init__.py
├── api_server.py # FastAPI + MCP HTTP
├── mcp_server.py # MCP 工具定义
└── prompts/ # 提示词模板
├── __init__.py
├── requirements_prompt.py
├── design_prompt.py
└── tasks_prompt.py
```
---
## 6. API 端点设计
| 端点 | 方法 | 描述 |
|------|------|------|
| `/` | GET | 服务状态 |
| `/health` | GET | 健康检查 |
| `/mcp` | POST | MCP JSON-RPC |
| `/mcp/sse` | GET/POST | MCP SSE 流式 |
| `/api/v1/requirements` | POST | 生成需求文档 |
| `/api/v1/requirements/refine` | POST | 修改需求文档 |
| `/api/v1/design` | POST | 生成设计文档 |
| `/api/v1/design/refine` | POST | 修改设计文档 |
| `/api/v1/tasks` | POST | 生成任务清单 |
| `/api/v1/tasks/refine` | POST | 修改任务清单 |
| `/api/v1/full-spec` | POST | 获取完整规范 |
---
## 7. 使用示例
### 输入
```
创建一个数据去重Agent
```
### 阶段1输出（需求文档）
```markdown
# 需求文档: 数据去重 Agent
## 1. 项目背景
创建一个智能Agent，用于检测和去除数据集中的重复项，支持多种数据格式和去重策略。
## 2. 用户故事
### US-001: 基础去重功能
**作为** 数据处理人员
**我想要** 对数据列表进行去重
**以便于** 获得无重复的干净数据
**验收标准**:
- **Given** 一个包含重复项的数据列表
**When** 调用去重工具
**Then** 返回去除重复后的数据列表，保持原始顺序
### US-002: 相似度去重
**作为** 数据分析师
**我想要** 基于相似度阈值进行模糊去重
**以便于** 处理近似重复的数据
**验收标准**:
- **Given** 一个数据列表和相似度阈值
**When** 调用模糊去重工具
**Then** 返回相似度超过阈值的项被合并后的数据
## 3. 功能需求清单
| ID | 功能名称 | 优先级 | 描述 |
|----|---------|--------|------|
| FR-001 | 精确去重 | P0 | 完全匹配的重复项去除 |
| FR-002 | 模糊去重 | P1 | 基于相似度的去重 |
| FR-003 | 批量处理 | P1 | 支持大数据量批量去重 |
...
```
---
## 8. 实施计划
### 8.1 开发任务清单
- [x] 创建项目目录结构
- [x] 实现 `mcp_server.py` 核心工具
- [x] `analyze_requirements` 工具
- [x] `refine_requirements` 工具
- [x] `create_design` 工具
- [x] `refine_design` 工具
- [x] `create_tasks` 工具
- [x] `refine_tasks` 工具
- [x] `get_full_spec` 工具
- [x] 实现提示词模板
- [x] 需求分析提示词
- [x] 设计规划提示词
- [x] 任务拆解提示词
- [x] 实现 `api_server.py` REST 端点
- [x] 编写 README.md 和 USAGE.md
- [x] 创建 Dockerfile
- [ ] 测试验证
---
## 9. 已确认事项
1. **项目类型支持范围**：仅支持 Agent 开发项目 ✅
2. **文档存储**：仅返回给调用方，不做持久化存储 ✅
3. **版本管理**：不支持版本历史 ✅
4. **模板定制**：不支持用户自定义模板 ✅
---
## 10. 风险与注意事项
1. **LLM 输出稳定性**：已实现 `extract_markdown()` 函数处理 LLM 输出格式不一致的情况
2. **上下文长度**：三阶段文档可能较长，需要注意 token 限制
3. **Mermaid 语法**：需要确保生成的 Mermaid 图语法正确，避免特殊字符问题
---
## 11. 已创建文件清单
```
specs_agent/
├── Dockerfile ✅
├── README.md ✅
├── USAGE.md ✅
├── requirements.txt ✅
├── run_api_server.py ✅
└── src/
├── __init__.py ✅
└── server/
├── __init__.py ✅
├── api_server.py ✅
├── mcp_server.py ✅
└── prompts/
├── __init__.py ✅
├── requirements_prompt.py ✅
├── design_prompt.py ✅
└── tasks_prompt.py ✅
```
specs_agent/Dockerfile
+20
View File
@@ -0,0 +1,20 @@
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
RUN apt-get update && apt-get install -y gcc curl && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
CMD ["python", "run_api_server.py"]
specs_agent/README.md
+169
View File
@@ -0,0 +1,169 @@
# Specs Agent
将一句话需求转化为完整的项目规范文档。
## 核心功能
**只做一件事**：将模糊的一句话需求转化为结构化的、可执行的项目规范文档。
专为 **Agent 开发项目** 设计，采用三阶段流程：
```
需求分析 → 设计规划 → 任务拆解
```
### 三阶段流程
| 阶段 | 工具 | 输出 |
|------|------|------|
| 1. 需求分析 | `analyze_requirements` | 用户故事 + 验收标准 |
| 2. 设计规划 | `create_design` | 架构图 + 接口定义 |
| 3. 任务拆解 | `create_tasks` | 任务清单 + 代码要点 |
每个阶段支持人工确认和修改（`refine_*` 工具）。
### 提供能力
| 工具 | 功能 |
|------|------|
| `analyze_requirements` | 分析需求，生成需求文档 |
| `refine_requirements` | 根据反馈修改需求文档 |
| `create_design` | 生成技术设计文档 |
| `refine_design` | 根据反馈修改设计文档 |
| `create_tasks` | 生成任务清单 |
| `refine_tasks` | 根据反馈修改任务清单 |
| `get_full_spec` | 合并生成完整规范文档 |
## 快速开始
### 本地运行
```bash
cd specs_agent
pip install -r requirements.txt
python run_api_server.py
```
### Docker 运行
```bash
docker build -t specs-agent:latest .
docker run -p 8000:8000 -e OPENAI_API_KEY=your-key specs-agent:latest
```
## 使用示例
### 输入
```
创建一个数据去重Agent
```
### 阶段1输出（需求文档）
```markdown
# 需求文档: 数据去重 Agent
## 1. 项目背景
创建一个智能Agent，用于检测和去除数据集中的重复项...
## 2. 用户故事
### US-001: 基础去重功能
**作为** 数据处理人员
**我想要** 对数据列表进行去重
**以便于** 获得无重复的干净数据
**验收标准**:
- **Given** 一个包含重复项的数据列表
**When** 调用去重工具
**Then** 返回去除重复后的数据列表
...
```
### 阶段2输出（设计文档）
```markdown
# 设计文档: 数据去重 Agent
## 1. 系统架构
### 1.1 架构图
[Mermaid 架构图]
## 2. MCP 工具设计
| 工具名称 | 功能描述 |
|---------|---------|
| deduplicate | 精确去重 |
| fuzzy_deduplicate | 模糊去重 |
...
```
### 阶段3输出（任务清单）
```markdown
# 任务清单: 数据去重 Agent
## 任务列表
### Task-001: 创建项目目录结构
**文件**: src/__init__.py, src/server/__init__.py
**验收标准**:
- [ ] 目录结构符合规范
...
```
## 环境变量
| 变量 | 必需 | 说明 |
|------|------|------|
| OPENAI_API_KEY | 是 | OpenAI 或 LiteLLM API Key |
| OPENAI_BASE_URL | 否 | API Base URL |
| MODEL_NAME | 否 | 模型名称，默认 taiji/gpt-4o-mini |
| API_PORT | 否 | 服务端口，默认 8000 |
## 项目结构
```
specs_agent/
├── Dockerfile
├── README.md
├── USAGE.md
├── requirements.txt
├── run_api_server.py
└── src/
├── __init__.py
└── server/
├── __init__.py
├── api_server.py
├── mcp_server.py
└── prompts/
├── __init__.py
├── requirements_prompt.py
├── design_prompt.py
└── tasks_prompt.py
```
## 服务端点
| 端点 | 方法 | 说明 |
|------|------|------|
| `/` | GET | 服务状态 |
| `/health` | GET | 健康检查 |
| `/mcp` | POST | MCP JSON-RPC |
| `/mcp/sse` | GET/POST | MCP SSE 流式 |
| `/api/v1/requirements` | POST | 生成需求文档 |
| `/api/v1/requirements/refine` | POST | 修改需求文档 |
| `/api/v1/design` | POST | 生成设计文档 |
| `/api/v1/design/refine` | POST | 修改设计文档 |
| `/api/v1/tasks` | POST | 生成任务清单 |
| `/api/v1/tasks/refine` | POST | 修改任务清单 |
| `/api/v1/full-spec` | POST | 获取完整规范 |
## 部署信息
| 配置项 | 值 |
|--------|-----|
| 镜像地址 | agnettaiji.azurecr.io/ai-agents/specs-agent:latest |
| 服务端口 | 8000 |
| 健康检查 | /health |
specs_agent/USAGE.md
+249
View File
@@ -0,0 +1,249 @@
# Specs Agent 使用指南
## 概述
Specs Agent 是一个项目规范生成工具，采用三阶段流程将一句话需求转化为完整的项目规范文档。
## API 使用
### 认证
所有 API 请求需要在 Header 中提供 API Key：
```bash
# 方式1: api-key header
-H "api-key: your-api-key"
# 方式2: Authorization header
-H "Authorization: Bearer your-api-key"
```
---
## 阶段一：需求分析
### 1.1 生成需求文档
**POST** `/api/v1/requirements`
```bash
curl -X POST http://localhost:8000/api/v1/requirements \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"brief_description": "创建一个数据去重Agent",
"context": "用于处理大量文本数据的去重需求"
}'
```
**请求参数**:
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| brief_description | string | 是 | 一句话需求描述 |
| context | string | 否 | 额外上下文信息 |
**响应示例**:
```json
{
"success": true,
"phase": "requirements",
"document": "# 需求文档: 数据去重 Agent\n\n## 1. 项目背景\n...",
"next_step": "请审阅需求文档，如需修改请调用 refine_requirements，确认后调用 create_design 进入设计阶段"
}
```
### 1.2 修改需求文档
**POST** `/api/v1/requirements/refine`
```bash
curl -X POST http://localhost:8000/api/v1/requirements/refine \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"current_requirements": "# 需求文档: 数据去重 Agent\n...",
"feedback": "请添加支持JSON格式数据的去重功能"
}'
```
**请求参数**:
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| current_requirements | string | 是 | 当前需求文档 |
| feedback | string | 是 | 修改意见 |
---
## 阶段二：设计规划
### 2.1 生成设计文档
**POST** `/api/v1/design`
```bash
curl -X POST http://localhost:8000/api/v1/design \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"requirements_doc": "# 需求文档: 数据去重 Agent\n...",
"tech_stack": "使用 Python 3.12，优先使用内置库"
}'
```
**请求参数**:
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| requirements_doc | string | 是 | 确认后的需求文档 |
| tech_stack | string | 否 | 技术栈偏好 |
**响应示例**:
```json
{
"success": true,
"phase": "design",
"document": "# 设计文档: 数据去重 Agent\n\n## 1. 系统架构\n...",
"next_step": "请审阅设计文档，如需修改请调用 refine_design，确认后调用 create_tasks 进入任务拆解阶段"
}
```
### 2.2 修改设计文档
**POST** `/api/v1/design/refine`
```bash
curl -X POST http://localhost:8000/api/v1/design/refine \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"current_design": "# 设计文档: 数据去重 Agent\n...",
"feedback": "请添加批量处理的API端点"
}'
```
---
## 阶段三：任务拆解
### 3.1 生成任务清单
**POST** `/api/v1/tasks`
```bash
curl -X POST http://localhost:8000/api/v1/tasks \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"design_doc": "# 设计文档: 数据去重 Agent\n..."
}'
```
**请求参数**:
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| design_doc | string | 是 | 确认后的设计文档 |
**响应示例**:
```json
{
"success": true,
"phase": "tasks",
"document": "# 任务清单: 数据去重 Agent\n\n## 任务总览\n...",
"next_step": "请审阅任务清单，如需修改请调用 refine_tasks，确认后调用 get_full_spec 获取完整规范文档"
}
```
### 3.2 修改任务清单
**POST** `/api/v1/tasks/refine`
```bash
curl -X POST http://localhost:8000/api/v1/tasks/refine \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"current_tasks": "# 任务清单: 数据去重 Agent\n...",
"feedback": "请将Task-002拆分为两个更小的任务"
}'
```
---
## 获取完整规范
### 合并三阶段文档
**POST** `/api/v1/full-spec`
```bash
curl -X POST http://localhost:8000/api/v1/full-spec \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"requirements_doc": "# 需求文档: 数据去重 Agent\n...",
"design_doc": "# 设计文档: 数据去重 Agent\n...",
"tasks_doc": "# 任务清单: 数据去重 Agent\n..."
}'
```
**响应示例**:
```json
{
"success": true,
"document": "# 数据去重 Agent - 完整规范文档\n\n> 本文档由 Specs Agent 自动生成...\n\n---\n\n# 第一部分：需求分析\n...\n\n---\n\n# 第二部分：设计规划\n...\n\n---\n\n# 第三部分：任务拆解\n...",
"message": "完整规范文档已生成，可以开始按任务清单执行开发"
}
```
---
## MCP 协议使用
### 工具列表
```bash
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list"
}'
```
### 调用工具
```bash
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "api-key: your-api-key" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "analyze_requirements",
"arguments": {
"brief_description": "创建一个数据去重Agent"
}
}
}'
```
---
## 错误处理
所有 API 在失败时返回：
```json
{
"success": false,
"error": "错误信息",
"phase": "requirements/design/tasks"
}
```
常见错误：
- `缺少 API Key`: 请在 Header 中提供有效的 API Key
- `需求描述不能为空`: brief_description 参数不能为空
- `需求文档不能为空`: 进入下一阶段前需要提供上一阶段的文档
specs_agent/requirements.txt
+13
View File
@@ -0,0 +1,13 @@
# Pydantic AI
pydantic-ai>=0.0.14
# MCP
mcp>=0.9.0
fastmcp>=0.1.0
# FastAPI
fastapi>=0.109.0
uvicorn[standard]>=0.27.0
# HTTP Client
aiohttp>=3.9.0
specs_agent/run_api_server.py
+22
View File
@@ -0,0 +1,22 @@
#!/usr/bin/env python
"""启动 Specs Agent API 服务器"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
if __name__ == '__main__':
from src.server.api_server import app
import uvicorn
import os
host = os.getenv('API_HOST', '0.0.0.0')
port = int(os.getenv('API_PORT', '8000'))
print(f"🚀 启动 Specs Agent API: http://{host}:{port}")
print("📋 工作流程:")
print(" 1. analyze_requirements - 需求分析")
print(" 2. create_design - 设计规划")
print(" 3. create_tasks - 任务拆解")
print(" 4. get_full_spec - 生成完整规范")
uvicorn.run(app, host=host, port=port, log_level="info")
specs_agent/src/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Specs Agent 源代码包"""
specs_agent/src/server/__init__.py
+1
View File
@@ -0,0 +1 @@
"""服务器模块"""
specs_agent/src/server/api_server.py
+397
View File
@@ -0,0 +1,397 @@
"""
Specs Agent HTTP API 服务器
提供 REST API 和 MCP HTTP/SSE 端点。
"""
import json
import uuid
import os
from typing import Optional, Dict, Any, AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Header, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse, JSONResponse
from pydantic import BaseModel, Field
from .mcp_server import TOOL_MAP, TOOL_LIST
# ==================== 配置 ====================
SERVER_NAME = "Specs Agent API"
# ==================== FastAPI 应用 ====================
@asynccontextmanager
async def lifespan(app: FastAPI):
print(f"🚀 {SERVER_NAME} 启动")
yield
print(f"🛑 {SERVER_NAME} 关闭")
app = FastAPI(
title=SERVER_NAME,
description="将一句话需求转化为完整的项目规范文档",
version="1.0.0",
lifespan=lifespan
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==================== API Key 验证 ====================
async def verify_api_key(
api_key: Optional[str] = Header(None, alias="api-key"),
authorization: Optional[str] = Header(None)
) -> str:
"""验证 API Key"""
if api_key and api_key.strip() and api_key.strip() != "sk":
return api_key.strip()
if authorization:
key = authorization[7:].strip() if authorization.startswith("Bearer ") else authorization.strip()
if key and key != "sk":
return key
raise HTTPException(status_code=401, detail="缺少 API Key")
def get_api_key_from_request(request: Request) -> Optional[str]:
"""从请求头提取 API Key（不验证）"""
api_key = request.headers.get("api-key") or request.headers.get("api_key")
if not api_key:
auth = request.headers.get("Authorization")
if auth:
api_key = auth[7:] if auth.startswith("Bearer ") else auth
return api_key
# ==================== 健康检查 ====================
@app.get("/")
async def root():
return {
"service": SERVER_NAME,
"status": "running",
"description": "将一句话需求转化为完整的项目规范文档",
"tools": list(TOOL_MAP.keys()),
"workflow": [
"1. analyze_requirements - 需求分析",
"2. create_design - 设计规划",
"3. create_tasks - 任务拆解",
"4. get_full_spec - 生成完整规范"
]
}
@app.get("/health")
async def health():
return {"status": "healthy", "service": SERVER_NAME}
# ==================== MCP 端点 ====================
sessions: Dict[str, Dict] = {}
async def handle_mcp_request(data: Dict, session_id: str = None, api_key: str = None) -> Dict:
"""处理 MCP JSON-RPC 请求"""
method = data.get("method")
params = data.get("params", {})
req_id = data.get("id")
# tools/call 需要验证 API Key
if method == "tools/call" and (not api_key or api_key == "sk"):
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32001, "message": "缺少 API Key"}}
try:
if method == "initialize":
session_id = session_id or str(uuid.uuid4())
sessions[session_id] = {"initialized": True}
return {
"jsonrpc": "2.0", "id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": SERVER_NAME, "version": "1.0.0"}
}
}
elif method == "tools/list":
return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOL_LIST}}
elif method == "tools/call":
tool_name = params.get("name")
args = params.get("arguments", {})
if tool_name not in TOOL_MAP:
raise ValueError(f"Unknown tool: {tool_name}")
# 设置 API Key 到环境变量
old_key = os.environ.get('OPENAI_API_KEY')
if api_key:
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP[tool_name](**args)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
return {
"jsonrpc": "2.0", "id": req_id,
"result": {"content": [{"type": "text", "text": str(result)}]}
}
elif method == "ping":
return {"jsonrpc": "2.0", "id": req_id, "result": {}}
else:
raise ValueError(f"Unknown method: {method}")
except Exception as e:
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(e)}}
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""MCP HTTP 端点"""
try:
body = await request.json()
session_id = request.headers.get("x-mcp-session-id")
api_key = get_api_key_from_request(request)
response = await handle_mcp_request(body, session_id, api_key)
return JSONResponse(content=response, headers={"x-mcp-session-id": session_id or ""})
except Exception as e:
return JSONResponse(status_code=400, content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}})
@app.get("/mcp/sse")
async def mcp_sse(request: Request):
"""MCP SSE 端点"""
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
async def stream() -> AsyncGenerator[str, None]:
yield f"data: {json.dumps({'type': 'connection', 'sessionId': session_id})}\n\n"
import asyncio
while True:
await asyncio.sleep(30)
yield f"data: {json.dumps({'type': 'ping'})}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id})
@app.post("/mcp/sse")
async def mcp_sse_post(request: Request):
"""MCP SSE POST 端点"""
try:
body = await request.json()
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
api_key = get_api_key_from_request(request)
async def stream() -> AsyncGenerator[str, None]:
response = await handle_mcp_request(body, session_id, api_key)
yield f"data: {json.dumps(response)}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id})
except Exception as e:
return JSONResponse(status_code=400, content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}})
# ==================== 业务 API ====================
# --- 阶段一：需求分析 ---
class AnalyzeRequirementsRequest(BaseModel):
"""需求分析请求"""
brief_description: str = Field(..., description="一句话需求描述")
context: Optional[str] = Field(None, description="额外上下文信息")
class RefineRequirementsRequest(BaseModel):
"""修改需求请求"""
current_requirements: str = Field(..., description="当前需求文档")
feedback: str = Field(..., description="修改意见")
@app.post("/api/v1/requirements")
async def api_analyze_requirements(request: AnalyzeRequirementsRequest, api_key: str = Depends(verify_api_key)):
"""生成需求文档"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['analyze_requirements'](
brief_description=request.brief_description,
context=request.context
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/requirements/refine")
async def api_refine_requirements(request: RefineRequirementsRequest, api_key: str = Depends(verify_api_key)):
"""修改需求文档"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['refine_requirements'](
current_requirements=request.current_requirements,
feedback=request.feedback
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# --- 阶段二：设计规划 ---
class CreateDesignRequest(BaseModel):
"""设计文档请求"""
requirements_doc: str = Field(..., description="需求文档")
tech_stack: Optional[str] = Field(None, description="技术栈偏好")
class RefineDesignRequest(BaseModel):
"""修改设计请求"""
current_design: str = Field(..., description="当前设计文档")
feedback: str = Field(..., description="修改意见")
@app.post("/api/v1/design")
async def api_create_design(request: CreateDesignRequest, api_key: str = Depends(verify_api_key)):
"""生成设计文档"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['create_design'](
requirements_doc=request.requirements_doc,
tech_stack=request.tech_stack
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/design/refine")
async def api_refine_design(request: RefineDesignRequest, api_key: str = Depends(verify_api_key)):
"""修改设计文档"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['refine_design'](
current_design=request.current_design,
feedback=request.feedback
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# --- 阶段三：任务拆解 ---
class CreateTasksRequest(BaseModel):
"""任务清单请求"""
design_doc: str = Field(..., description="设计文档")
class RefineTasksRequest(BaseModel):
"""修改任务请求"""
current_tasks: str = Field(..., description="当前任务清单")
feedback: str = Field(..., description="修改意见")
@app.post("/api/v1/tasks")
async def api_create_tasks(request: CreateTasksRequest, api_key: str = Depends(verify_api_key)):
"""生成任务清单"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['create_tasks'](
design_doc=request.design_doc
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/tasks/refine")
async def api_refine_tasks(request: RefineTasksRequest, api_key: str = Depends(verify_api_key)):
"""修改任务清单"""
try:
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP['refine_tasks'](
current_tasks=request.current_tasks,
feedback=request.feedback
)
return json.loads(result)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# --- 完整规范 ---
class FullSpecRequest(BaseModel):
"""完整规范请求"""
requirements_doc: str = Field(..., description="需求文档")
design_doc: str = Field(..., description="设计文档")
tasks_doc: str = Field(..., description="任务清单")
@app.post("/api/v1/full-spec")
async def api_get_full_spec(request: FullSpecRequest, api_key: str = Depends(verify_api_key)):
"""获取完整规范文档"""
try:
result = await TOOL_MAP['get_full_spec'](
requirements_doc=request.requirements_doc,
design_doc=request.design_doc,
tasks_doc=request.tasks_doc
)
return json.loads(result)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
specs_agent/src/server/mcp_server.py
+602
View File
@@ -0,0 +1,602 @@
"""
Specs Agent MCP 服务器
核心功能：将一句话需求转化为完整的项目规范文档。
三阶段流程：需求分析 → 设计规划 → 任务拆解
"""
import json
import os
import re
from typing import Optional
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
from .prompts.requirements_prompt import (
REQUIREMENTS_SYSTEM_PROMPT,
ANALYZE_REQUIREMENTS_PROMPT,
REFINE_REQUIREMENTS_PROMPT,
)
from .prompts.design_prompt import (
DESIGN_SYSTEM_PROMPT,
CREATE_DESIGN_PROMPT,
REFINE_DESIGN_PROMPT,
)
from .prompts.tasks_prompt import (
TASKS_SYSTEM_PROMPT,
CREATE_TASKS_PROMPT,
REFINE_TASKS_PROMPT,
)
# ==================== 配置 ====================
# LiteLLM Gateway 配置
_BASE_URL = os.getenv('OPENAI_BASE_URL',
os.getenv('LLM_BASE_URL', 'https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1'))
_API_KEY = os.getenv('OPENAI_API_KEY', 'sk')
os.environ.setdefault('OPENAI_API_KEY', _API_KEY)
os.environ.setdefault('OPENAI_BASE_URL', _BASE_URL)
# 模型名称（pydantic_ai 需要 openai: 前缀）
def _get_model_name() -> str:
model = os.getenv('MODEL_NAME', os.getenv('LITELLM_MODEL', 'taiji/gpt-4o-mini'))
return model if ':' in model else f'openai:{model}'
MODEL_NAME = _get_model_name()
# ==================== MCP 服务器 ====================
server = FastMCP('Specs Agent')
def get_requirements_agent() -> Agent:
"""创建需求分析 Agent"""
return Agent(MODEL_NAME, system_prompt=REQUIREMENTS_SYSTEM_PROMPT)
def get_design_agent() -> Agent:
"""创建设计规划 Agent"""
return Agent(MODEL_NAME, system_prompt=DESIGN_SYSTEM_PROMPT)
def get_tasks_agent() -> Agent:
"""创建任务拆解 Agent"""
return Agent(MODEL_NAME, system_prompt=TASKS_SYSTEM_PROMPT)
def extract_markdown(text: str) -> str:
"""从 AI 输出中提取 Markdown 内容"""
# 尝试提取 ```markdown ... ``` 代码块
pattern = r'```markdown\s*([\s\S]*?)\s*```'
match = re.search(pattern, text)
if match:
return match.group(1).strip()
# 如果没有代码块，返回原文本（去除首尾空白）
return text.strip()
# ==================== 阶段一：需求分析 ====================
@server.tool()
async def analyze_requirements(
brief_description: str,
context: Optional[str] = None
) -> str:
"""
分析用户需求，生成结构化的需求文档。
这是三阶段流程的第一步：需求分析阶段。
将模糊的一句话需求转化为详细的需求文档，包含用户故事和验收标准。
Args:
brief_description: 一句话需求描述（如"创建一个数据去重Agent"）
context: 可选的额外上下文信息
Returns:
需求文档（Markdown 格式）
"""
if not brief_description or not brief_description.strip():
return json.dumps({
"success": False,
"error": "需求描述不能为空",
"phase": "requirements"
}, ensure_ascii=False)
try:
# 构建上下文部分
context_section = f"额外上下文：\n{context}" if context else ""
# 构建提示
prompt = ANALYZE_REQUIREMENTS_PROMPT.format(
brief_description=brief_description,
context_section=context_section
)
# 调用 AI 分析
result = await get_requirements_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "requirements",
"document": output,
"next_step": "请审阅需求文档，如需修改请调用 refine_requirements，确认后调用 create_design 进入设计阶段"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "requirements"
}, ensure_ascii=False)
@server.tool()
async def refine_requirements(
current_requirements: str,
feedback: str
) -> str:
"""
根据用户反馈修改需求文档。
Args:
current_requirements: 当前的需求文档（Markdown 格式）
feedback: 用户的反馈或修改意见
Returns:
修改后的需求文档（Markdown 格式）
"""
if not current_requirements or not feedback:
return json.dumps({
"success": False,
"error": "需求文档和反馈都不能为空",
"phase": "requirements"
}, ensure_ascii=False)
try:
prompt = REFINE_REQUIREMENTS_PROMPT.format(
current_requirements=current_requirements,
feedback=feedback
)
result = await get_requirements_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "requirements",
"document": output,
"next_step": "请审阅修改后的需求文档，如需继续修改请再次调用 refine_requirements，确认后调用 create_design 进入设计阶段"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "requirements"
}, ensure_ascii=False)
# ==================== 阶段二：设计规划 ====================
@server.tool()
async def create_design(
requirements_doc: str,
tech_stack: Optional[str] = None
) -> str:
"""
基于需求文档生成技术设计文档。
这是三阶段流程的第二步：设计规划阶段。
将需求文档转化为技术设计，包含架构图、接口定义、数据模型等。
Args:
requirements_doc: 确认后的需求文档（Markdown 格式）
tech_stack: 可选的技术栈偏好说明
Returns:
设计文档（Markdown 格式，含 Mermaid 图）
"""
if not requirements_doc or not requirements_doc.strip():
return json.dumps({
"success": False,
"error": "需求文档不能为空",
"phase": "design"
}, ensure_ascii=False)
try:
# 构建技术栈部分
tech_stack_section = f"技术栈偏好：\n{tech_stack}" if tech_stack else ""
prompt = CREATE_DESIGN_PROMPT.format(
requirements_doc=requirements_doc,
tech_stack_section=tech_stack_section
)
result = await get_design_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "design",
"document": output,
"next_step": "请审阅设计文档，如需修改请调用 refine_design，确认后调用 create_tasks 进入任务拆解阶段"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "design"
}, ensure_ascii=False)
@server.tool()
async def refine_design(
current_design: str,
feedback: str
) -> str:
"""
根据用户反馈修改设计文档。
Args:
current_design: 当前的设计文档（Markdown 格式）
feedback: 用户的反馈或修改意见
Returns:
修改后的设计文档（Markdown 格式）
"""
if not current_design or not feedback:
return json.dumps({
"success": False,
"error": "设计文档和反馈都不能为空",
"phase": "design"
}, ensure_ascii=False)
try:
prompt = REFINE_DESIGN_PROMPT.format(
current_design=current_design,
feedback=feedback
)
result = await get_design_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "design",
"document": output,
"next_step": "请审阅修改后的设计文档，如需继续修改请再次调用 refine_design，确认后调用 create_tasks 进入任务拆解阶段"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "design"
}, ensure_ascii=False)
# ==================== 阶段三：任务拆解 ====================
@server.tool()
async def create_tasks(
design_doc: str
) -> str:
"""
基于设计文档生成任务清单。
这是三阶段流程的第三步：任务拆解阶段。
将设计文档转化为可执行的任务清单，包含依赖关系和验收标准。
Args:
design_doc: 确认后的设计文档（Markdown 格式）
Returns:
任务清单（Markdown 格式）
"""
if not design_doc or not design_doc.strip():
return json.dumps({
"success": False,
"error": "设计文档不能为空",
"phase": "tasks"
}, ensure_ascii=False)
try:
prompt = CREATE_TASKS_PROMPT.format(
design_doc=design_doc
)
result = await get_tasks_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "tasks",
"document": output,
"next_step": "请审阅任务清单，如需修改请调用 refine_tasks，确认后调用 get_full_spec 获取完整规范文档"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "tasks"
}, ensure_ascii=False)
@server.tool()
async def refine_tasks(
current_tasks: str,
feedback: str
) -> str:
"""
根据用户反馈修改任务清单。
Args:
current_tasks: 当前的任务清单（Markdown 格式）
feedback: 用户的反馈或修改意见
Returns:
修改后的任务清单（Markdown 格式）
"""
if not current_tasks or not feedback:
return json.dumps({
"success": False,
"error": "任务清单和反馈都不能为空",
"phase": "tasks"
}, ensure_ascii=False)
try:
prompt = REFINE_TASKS_PROMPT.format(
current_tasks=current_tasks,
feedback=feedback
)
result = await get_tasks_agent().run(prompt)
output = extract_markdown(result.output)
return json.dumps({
"success": True,
"phase": "tasks",
"document": output,
"next_step": "请审阅修改后的任务清单，如需继续修改请再次调用 refine_tasks，确认后调用 get_full_spec 获取完整规范文档"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"phase": "tasks"
}, ensure_ascii=False)
# ==================== 通用工具 ====================
@server.tool()
async def get_full_spec(
requirements_doc: str,
design_doc: str,
tasks_doc: str
) -> str:
"""
合并三个阶段的文档，生成完整的项目规范文档。
Args:
requirements_doc: 需求文档（Markdown 格式）
design_doc: 设计文档（Markdown 格式）
tasks_doc: 任务清单（Markdown 格式）
Returns:
完整的项目规范文档（Markdown 格式）
"""
if not all([requirements_doc, design_doc, tasks_doc]):
return json.dumps({
"success": False,
"error": "三个阶段的文档都不能为空"
}, ensure_ascii=False)
try:
# 提取项目名称（从需求文档标题）
project_name = "项目"
title_match = re.search(r'#\s*需求文档:\s*(.+)', requirements_doc)
if title_match:
project_name = title_match.group(1).strip()
# 合并文档
full_spec = f"""# {project_name} - 完整规范文档
> 本文档由 Specs Agent 自动生成，包含需求分析、设计规划、任务拆解三个阶段的完整内容。
---
# 第一部分：需求分析
{requirements_doc}
---
# 第二部分：设计规划
{design_doc}
---
# 第三部分：任务拆解
{tasks_doc}
---
# 附录
## 文档版本
- 生成时间: {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
- 生成工具: Specs Agent
## 使用说明
1. 按照任务清单中的顺序执行开发任务
2. 每完成一个任务，检查对应的验收标准
3. 如有问题，参考设计文档中的技术细节
4. 如需修改需求，重新运行 Specs Agent 生成新的规范
"""
return json.dumps({
"success": True,
"document": full_spec,
"message": "完整规范文档已生成，可以开始按任务清单执行开发"
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
}, ensure_ascii=False)
# ==================== 工具映射（供 API 使用）====================
TOOL_MAP = {
'analyze_requirements': analyze_requirements,
'refine_requirements': refine_requirements,
'create_design': create_design,
'refine_design': refine_design,
'create_tasks': create_tasks,
'refine_tasks': refine_tasks,
'get_full_spec': get_full_spec,
}
TOOL_LIST = [
{
"name": "analyze_requirements",
"description": "分析用户需求，生成结构化的需求文档（阶段一）",
"inputSchema": {
"type": "object",
"properties": {
"brief_description": {
"type": "string",
"description": "一句话需求描述（如'创建一个数据去重Agent'）"
},
"context": {
"type": "string",
"description": "可选的额外上下文信息"
}
},
"required": ["brief_description"]
}
},
{
"name": "refine_requirements",
"description": "根据用户反馈修改需求文档",
"inputSchema": {
"type": "object",
"properties": {
"current_requirements": {
"type": "string",
"description": "当前的需求文档（Markdown 格式）"
},
"feedback": {
"type": "string",
"description": "用户的反馈或修改意见"
}
},
"required": ["current_requirements", "feedback"]
}
},
{
"name": "create_design",
"description": "基于需求文档生成技术设计文档（阶段二）",
"inputSchema": {
"type": "object",
"properties": {
"requirements_doc": {
"type": "string",
"description": "确认后的需求文档（Markdown 格式）"
},
"tech_stack": {
"type": "string",
"description": "可选的技术栈偏好说明"
}
},
"required": ["requirements_doc"]
}
},
{
"name": "refine_design",
"description": "根据用户反馈修改设计文档",
"inputSchema": {
"type": "object",
"properties": {
"current_design": {
"type": "string",
"description": "当前的设计文档（Markdown 格式）"
},
"feedback": {
"type": "string",
"description": "用户的反馈或修改意见"
}
},
"required": ["current_design", "feedback"]
}
},
{
"name": "create_tasks",
"description": "基于设计文档生成任务清单（阶段三）",
"inputSchema": {
"type": "object",
"properties": {
"design_doc": {
"type": "string",
"description": "确认后的设计文档（Markdown 格式）"
}
},
"required": ["design_doc"]
}
},
{
"name": "refine_tasks",
"description": "根据用户反馈修改任务清单",
"inputSchema": {
"type": "object",
"properties": {
"current_tasks": {
"type": "string",
"description": "当前的任务清单（Markdown 格式）"
},
"feedback": {
"type": "string",
"description": "用户的反馈或修改意见"
}
},
"required": ["current_tasks", "feedback"]
}
},
{
"name": "get_full_spec",
"description": "合并三个阶段的文档，生成完整的项目规范文档",
"inputSchema": {
"type": "object",
"properties": {
"requirements_doc": {
"type": "string",
"description": "需求文档（Markdown 格式）"
},
"design_doc": {
"type": "string",
"description": "设计文档（Markdown 格式）"
},
"tasks_doc": {
"type": "string",
"description": "任务清单（Markdown 格式）"
}
},
"required": ["requirements_doc", "design_doc", "tasks_doc"]
}
}
]
if __name__ == '__main__':
server.run()
specs_agent/src/server/prompts/__init__.py
+1
View File
@@ -0,0 +1 @@
"""提示词模板模块"""
specs_agent/src/server/prompts/design_prompt.py
+241
View File
@@ -0,0 +1,241 @@
"""
设计规划阶段提示词模板
"""
DESIGN_SYSTEM_PROMPT = '''你是一个专业的软件架构师，专注于 Agent 开发项目的技术设计。
你的核心能力：
1. 设计清晰的系统架构（使用 Mermaid 图）
2. 定义 MCP 工具接口
3. 设计 API 端点
4. 规划数据模型和文件结构
Agent 开发专业知识：
- 精通 Pydantic AI 框架
- 精通 FastMCP 工具定义
- 精通 FastAPI 服务开发
- 熟悉 MCP 协议规范
- 了解 LiteLLM Gateway 集成
输出原则：
- 架构图使用 Mermaid 语法
- 接口定义要完整（参数、返回值、描述）
- 数据模型使用 Pydantic 风格
- 文件结构遵循项目模板规范'''
CREATE_DESIGN_PROMPT = '''请基于以下需求文档，生成技术设计文档。
需求文档：
```markdown
{requirements_doc}
```
{tech_stack_section}
请按以下 Markdown 格式输出设计文档：
```markdown
# 设计文档: [项目名称]
## 1. 系统架构
### 1.1 架构概述
[简要描述系统架构设计思路]
### 1.2 架构图
```mermaid
flowchart TB
subgraph Client["客户端"]
A[MCP Client]
B[HTTP Client]
end
subgraph Agent["Agent 服务"]
C[FastAPI Server]
D[MCP Handler]
E[Tool: xxx]
F[Tool: xxx]
end
subgraph External["外部服务"]
G[LiteLLM Gateway]
end
A -->|MCP Protocol| D
B -->|REST API| C
C --> D
D --> E
D --> F
E --> G
F --> G
```
### 1.3 组件说明
| 组件 | 职责 | 技术选型 |
|------|------|---------|
| FastAPI Server | HTTP 服务入口 | FastAPI + Uvicorn |
| MCP Handler | MCP 协议处理 | FastMCP |
| Tool: xxx | [工具职责] | Pydantic AI |
| LLM Client | 大模型调用 | Pydantic AI Agent |
## 2. MCP 工具设计
### 2.1 工具清单
| 工具名称 | 功能描述 | 输入参数 | 输出格式 |
|---------|---------|---------|---------|
| tool_name_1 | [功能描述] | param1: str, param2: Optional[str] | JSON |
| tool_name_2 | [功能描述] | param1: str | JSON |
### 2.2 工具详细定义
#### tool_name_1
```python
@server.tool()
async def tool_name_1(
param1: str,
param2: Optional[str] = None
) -> str:
"""
[工具描述]
Args:
param1: [参数1描述]
param2: [参数2描述，可选]
Returns:
[返回值描述]（JSON 格式）
"""
```
#### tool_name_2
[继续定义其他工具...]
## 3. API 端点设计
### 3.1 端点清单
| 端点 | 方法 | 描述 | 认证 |
|------|------|------|------|
| / | GET | 服务状态 | 否 |
| /health | GET | 健康检查 | 否 |
| /mcp | POST | MCP JSON-RPC | 是 |
| /api/v1/xxx | POST | [业务接口] | 是 |
### 3.2 请求/响应模型
#### POST /api/v1/xxx
**请求体**:
```json
{{
"field1": "string",
"field2": "string (optional)"
}}
```
**响应体**:
```json
{{
"success": true,
"result": "..."
}}
```
## 4. 数据模型
### 4.1 请求模型
```python
class XxxRequest(BaseModel):
"""请求模型"""
field1: str = Field(..., description="字段1描述")
field2: Optional[str] = Field(None, description="字段2描述")
```
### 4.2 响应模型
```python
class XxxResponse(BaseModel):
"""响应模型"""
success: bool
result: Optional[str] = None
error: Optional[str] = None
```
## 5. 系统提示词设计
```
[Agent 的 System Prompt，描述 Agent 的角色、能力和输出要求]
```
## 6. 项目文件结构
```
[agent_name]/
├── Dockerfile
├── README.md
├── USAGE.md
├── requirements.txt
├── run_api_server.py
└── src/
├── __init__.py
└── server/
├── __init__.py
├── api_server.py # FastAPI + MCP HTTP
└── mcp_server.py # MCP 工具定义
```
## 7. 依赖清单
```
# requirements.txt
# Pydantic AI
pydantic-ai>=0.0.14
# MCP
mcp>=0.9.0
fastmcp>=0.1.0
# FastAPI
fastapi>=0.109.0
uvicorn[standard]>=0.27.0
# HTTP Client
aiohttp>=3.9.0
# [其他依赖]
```
## 8. 环境变量
| 变量 | 必需 | 说明 | 默认值 |
|------|------|------|--------|
| OPENAI_API_KEY | 是 | API Key | - |
| OPENAI_BASE_URL | 否 | API Base URL | LiteLLM Gateway |
| MODEL_NAME | 否 | 模型名称 | taiji/gpt-4o-mini |
| API_PORT | 否 | 服务端口 | 8000 |
```
要求：
1. 架构图必须使用有效的 Mermaid 语法
2. 工具定义要完整，包含所有参数和返回值
3. API 端点要覆盖所有功能需求
4. 数据模型使用 Pydantic 风格
5. 文件结构遵循 agent_templates 规范'''
REFINE_DESIGN_PROMPT = '''请根据用户反馈修改设计文档。
当前设计文档：
```markdown
{current_design}
```
用户反馈/修改意见：
「{feedback}」
请根据反馈修改设计文档，输出完整的修改后文档（Markdown格式）。
要求：
1. 保持文档结构不变
2. 仅修改用户反馈涉及的部分
3. 确保架构图、接口定义、数据模型保持一致
4. 如果修改了工具定义，同步更新 TOOL_MAP 和 TOOL_LIST
5. 确保修改后的文档仍然完整、可实现'''
specs_agent/src/server/prompts/requirements_prompt.py
+118
View File
@@ -0,0 +1,118 @@
"""
需求分析阶段提示词模板
"""
REQUIREMENTS_SYSTEM_PROMPT = '''你是一个专业的软件需求分析专家，专注于 Agent 开发项目。
你的核心能力：
1. 将模糊的需求转化为清晰的用户故事（User Story）
2. 为每个需求设计可验证的验收标准（Given-When-Then）
3. 识别功能需求和非功能需求
4. 发现潜在的约束和假设
Agent 开发专业知识：
- 熟悉 Pydantic AI 框架
- 熟悉 FastMCP 工具定义
- 熟悉 FastAPI 服务开发
- 了解 MCP 协议规范
输出原则：
- 需求必须具体、可测量、可实现
- 验收标准必须使用 Given-When-Then 格式
- 优先级使用 P0（必须）、P1（重要）、P2（可选）
- 输出必须是有效的 Markdown 格式'''
ANALYZE_REQUIREMENTS_PROMPT = '''请分析以下需求，生成结构化的需求文档。
用户需求描述：
「{brief_description}」
{context_section}
请按以下 Markdown 格式输出需求文档：
```markdown
# 需求文档: [项目名称]
## 1. 项目背景
[根据需求描述推断的项目背景、目标和价值]
## 2. 用户故事 (User Stories)
### US-001: [故事标题]
**作为** [角色，如：开发者/数据分析师/系统管理员]
**我想要** [具体功能描述]
**以便于** [业务价值/目标]
**验收标准 (Acceptance Criteria)**:
- **Given** [前置条件]
**When** [触发动作]
**Then** [期望结果]
- **Given** [前置条件2]
**When** [触发动作2]
**Then** [期望结果2]
### US-002: [故事标题]
[继续添加更多用户故事...]
## 3. 功能需求清单
| ID | 功能名称 | 优先级 | 描述 |
|----|---------|--------|------|
| FR-001 | [功能名] | P0 | [详细描述] |
| FR-002 | [功能名] | P1 | [详细描述] |
| FR-003 | [功能名] | P2 | [详细描述] |
## 4. 非功能需求
### 4.1 性能要求
- [具体性能指标]
### 4.2 安全要求
- [安全相关要求]
### 4.3 可用性要求
- [可用性相关要求]
## 5. 约束与假设
### 5.1 技术约束
- 基于 Pydantic AI 框架开发
- 使用 FastMCP 定义工具
- 使用 FastAPI 提供 HTTP 服务
- [其他技术约束]
### 5.2 业务假设
- [业务假设1]
- [业务假设2]
## 6. 术语表
| 术语 | 定义 |
|------|------|
| [术语1] | [定义] |
```
要求：
1. 用户故事至少包含 3-5 个核心功能
2. 每个用户故事至少有 2 条验收标准
3. 功能需求按优先级排序
4. 非功能需求要具体可测量
5. 约束和假设要明确列出'''
REFINE_REQUIREMENTS_PROMPT = '''请根据用户反馈修改需求文档。
当前需求文档：
```markdown
{current_requirements}
```
用户反馈/修改意见：
「{feedback}」
请根据反馈修改需求文档，输出完整的修改后文档（Markdown格式）。
要求：
1. 保持文档结构不变
2. 仅修改用户反馈涉及的部分
3. 如果反馈要求添加新功能，添加对应的用户故事和功能需求
4. 如果反馈要求删除功能，移除相关内容
5. 确保修改后的文档仍然完整、一致'''
specs_agent/src/server/prompts/tasks_prompt.py
+278
View File
@@ -0,0 +1,278 @@
"""
任务拆解阶段提示词模板
"""
TASKS_SYSTEM_PROMPT = '''你是一个专业的项目经理，专注于 Agent 开发项目的任务拆解。
你的核心能力：
1. 将设计文档拆解为可执行的开发任务
2. 确定任务之间的依赖关系
3. 为每个任务定义清晰的验收标准
4. 提供关键代码要点和实现提示
Agent 开发专业知识：
- 精通 Pydantic AI 框架
- 精通 FastMCP 工具定义
- 精通 FastAPI 服务开发
- 熟悉 MCP 协议规范
- 了解 Docker 容器化部署
输出原则：
- 任务粒度适中（每个任务 1-4 小时工作量）
- 任务描述具体、可执行
- 验收标准可检查
- 代码要点提供关键实现思路'''
CREATE_TASKS_PROMPT = r"""请基于以下设计文档，生成任务清单。
设计文档：
```markdown
{design_doc}
```
请按以下 Markdown 格式输出任务清单：
```markdown
# 任务清单: [项目名称]
## 任务总览
- **总任务数**: X
- **预计文件数**: X
- **核心文件**: mcp_server.py, api_server.py
## 执行顺序图
```mermaid
flowchart LR
T001[Task-001] --> T002[Task-002]
T002 --> T003[Task-003]
T002 --> T004[Task-004]
T003 --> T005[Task-005]
T004 --> T005
T005 --> T006[Task-006]
```
## 任务列表
---
### Task-001: 创建项目目录结构
**类型**: 创建文件
**文件**:
- `[agent_name]/src/__init__.py`
- `[agent_name]/src/server/__init__.py`
**依赖**: 无
**预计时间**: 10分钟
**描述**:
创建 Agent 项目的基础目录结构，包含必要的 Python 包初始化文件。
**验收标准**:
- [ ] 目录结构符合 agent_templates 规范
- [ ] 所有 `__init__.py` 文件已创建
- [ ] 可以作为 Python 包导入
**代码要点**:
```python
# src/__init__.py
# [Agent名称] 源代码包 的 docstring
# src/server/__init__.py
# 服务器模块 的 docstring
```
---
### Task-002: 实现 MCP 工具核心逻辑
**类型**: 创建文件
**文件**: `[agent_name]/src/server/mcp_server.py`
**依赖**: Task-001
**预计时间**: 2小时
**描述**:
实现 Agent 的核心 MCP 工具，包括：
- 工具函数定义
- 系统提示词
- TOOL_MAP 和 TOOL_LIST
**验收标准**:
- [ ] 所有工具函数已实现
- [ ] 工具函数有完整的 docstring
- [ ] TOOL_MAP 包含所有工具
- [ ] TOOL_LIST 包含所有工具的 schema
- [ ] 可以独立运行 `python mcp_server.py`
**代码要点**:
```python
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
server = FastMCP('[Agent名称]')
# SYSTEM_PROMPT = 系统提示词字符串
@server.tool()
async def tool_name(param1: str) -> str:
# 工具描述 docstring
# 实现逻辑
pass
TOOL_MAP = {{'tool_name': tool_name}}
TOOL_LIST = [...]
```
---
### Task-003: 实现 API 服务器
**类型**: 创建文件
**文件**: `[agent_name]/src/server/api_server.py`
**依赖**: Task-002
**预计时间**: 1小时
**描述**:
实现 FastAPI 服务器，提供：
- 健康检查端点
- MCP HTTP/SSE 端点
- 业务 API 端点
**验收标准**:
- [ ] 所有端点已实现
- [ ] MCP 协议处理正确
- [ ] API Key 验证正常
- [ ] 可以启动服务器
**代码要点**:
```python
from fastapi import FastAPI
from .mcp_server import TOOL_MAP, TOOL_LIST
app = FastAPI(title="[Agent名称]")
@app.get("/health")
async def health():
return {{"status": "healthy"}}
```
---
### Task-004: 创建启动脚本
**类型**: 创建文件
**文件**: `[agent_name]/run_api_server.py`
**依赖**: Task-003
**预计时间**: 10分钟
**描述**:
创建服务启动脚本。
**验收标准**:
- [ ] 可以通过 `python run_api_server.py` 启动服务
- [ ] 支持环境变量配置端口
**代码要点**:
```python
#!/usr/bin/env python
import uvicorn
from src.server.api_server import app
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=8000)
```
---
### Task-005: 创建配置文件
**类型**: 创建文件
**文件**:
- `[agent_name]/requirements.txt`
- `[agent_name]/Dockerfile`
**依赖**: Task-003
**预计时间**: 15分钟
**描述**:
创建项目依赖和 Docker 配置。
**验收标准**:
- [ ] requirements.txt 包含所有依赖
- [ ] Dockerfile 可以成功构建
- [ ] 容器可以正常运行
**代码要点**:
```dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "run_api_server.py"]
```
---
### Task-006: 编写文档
**类型**: 创建文件
**文件**:
- `[agent_name]/README.md`
- `[agent_name]/USAGE.md`
**依赖**: Task-005
**预计时间**: 30分钟
**描述**:
编写项目文档，包括：
- README: 项目介绍、快速开始
- USAGE: 详细使用说明、API 文档
**验收标准**:
- [ ] README 包含项目介绍和快速开始
- [ ] USAGE 包含所有工具和 API 的使用示例
- [ ] 文档格式正确、内容完整
---
## 检查清单
### 开发完成检查
- [ ] 所有任务已完成
- [ ] 代码可以正常运行
- [ ] 所有工具功能正常
### 测试检查
- [ ] 健康检查端点正常
- [ ] MCP 工具调用正常
- [ ] API 端点正常
### 部署检查
- [ ] Docker 镜像构建成功
- [ ] 容器运行正常
- [ ] 环境变量配置正确
```
要求：
1. 任务按依赖关系排序
2. 每个任务有明确的验收标准
3. 代码要点提供关键实现思路
4. 任务粒度适中，便于执行
5. 包含最终的检查清单"""
REFINE_TASKS_PROMPT = '''请根据用户反馈修改任务清单。
当前任务清单：
```markdown
{current_tasks}
```
用户反馈/修改意见：
「{feedback}」
请根据反馈修改任务清单，输出完整的修改后文档（Markdown格式）。
要求：
1. 保持文档结构不变
2. 仅修改用户反馈涉及的部分
3. 如果添加/删除任务，更新任务总览和执行顺序图
4. 确保任务依赖关系正确
5. 确保修改后的任务清单仍然完整、可执行'''