20260514更新

This commit is contained in:
liubiren 2026-05-14 21:52:16 +08:00
parent f55bec1c80
commit bb9d50dece
12 changed files with 839 additions and 57 deletions

View File

@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
"""
主运行模块
"""
# 列举导入模块
import asyncio
from utils.agent import BaseAgent
# 主函数
async def main():
print("进入会话模式,输入 exit 结束会话")
# 实例智能体
agent = BaseAgent(instructions="请用一句话简洁回复。")
while True:
user_prompt = input("用户:").strip().lower()
if user_prompt == "exit":
print("会话结束")
break
result = await agent.run(user_prompt=user_prompt)
print("智能体:", result.output)
if __name__ == "__main__":
asyncio.run(main=main())

View File

@ -4,46 +4,16 @@
"""
# 列举导入模块
import asyncio
from uuid import uuid4
import uvicorn
from utils.agent import BaseAgent
from utils.memory import Memory
# 主函数
async def main():
print("进入会话模式,输入 exit 结束会话")
# 实例智能体
agent = BaseAgent(instructions="请用一句话简洁回复。")
# 实例记忆体
memory = Memory()
# 生成会话唯一标识
session_id = uuid4().hex.lower()
dialogue_round = 1
while True:
user_prompt = input("用户:").strip().lower()
if user_prompt == "exit":
print("会话结束")
break
# 查询会话历史消息
message_history = memory.read(session_id=session_id)
result = await agent.run(
user_prompt=user_prompt, message_history=message_history
)
# 记录会话历史消息
memory.create(
session_id=session_id,
dialogue_round=dialogue_round,
dialogue_message=result.new_messages(),
)
print("智能体:", result.output)
dialogue_round += 1
# 实例智能体
agent = BaseAgent(
instructions="使用提供的技能回答问题。", skill_name="requirements-analysis"
)
# 实例 Starlette 应用
application = agent.to_web()
if __name__ == "__main__":
asyncio.run(main=main())
uvicorn.run(application, host="127.0.0.1", port=7932)

View File

@ -0,0 +1,139 @@
---
name: "requirements-analysis"
description: "资深需求分析师.通过多轮对话将想法生成产品需求文档。使用此技能当用户需要:(1) 需求分析,(2) 将史诗需求拆分为功能和用户故事,(3) 就多个需求进行优先级排序,(4) 识别利益相关者和依赖关系,(5) 定义验收标准,(6) 创建需求文档。支持单个需求分析、史诗级需求拆分、MoSCoW优先级排序方法。"
---
# 需求分析专家
## 核心能力
1. **对话式需求提取** - 通过系统化问题将一句话需求转化为全面规格说明
2. **分层需求分解** - 将复杂需求从 EPIC → 需求 → 用户故事进行分解
3. **优先级排序** - 支持 MoSCoW、RICE、价值 vs 成本、Kano、加权评分等方法
4. **文档生成** - 创建结构化的专业需求文档
## 单个需求分析工作流程
### 步骤 1初始需求捕获
提出问题:
1. 用一句话描述这个需求是什么?
2. 这解决了什么问题?
3. 谁提出了这个需求?
### 步骤 2利益相关者识别
提出问题:
1. 谁是最终用户?
2. 谁是业务负责人/赞助人?
3. 谁是技术利益相关者?
4. 是否有合规或安全利益相关者?
### 步骤 3详细需求规格
**功能性需求**
- 需要哪些具体功能?
- 用户工作流程是什么?
- 需要捕获/显示哪些数据?
**非功能性需求**
- 性能期望(响应时间、吞吐量)
- 安全要求(认证、授权、加密)
- 可扩展性需求(并发用户数、数据量)
- 可用性要求(正常运行时间、灾难恢复)
**业务背景**
- 业务价值是什么?
- 成功指标是什么?
- 预期的投资回报率是多少?
### 步骤 4时间线和依赖关系
提出问题:
1. 什么时候需要?(硬性截止日期还是灵活的?)
2. 是否依赖其他项目/系统?
3. 是否有外部因素(监管截止日期、市场事件)?
4. 首选的交付方式是什么?(一次性交付还是分阶段?)
### 步骤 5验收标准定义
使用 Given-When-Then 格式:
```
Given [前置条件]
When [操作]
Then [预期结果]
```
### 步骤 6文档编写
参考 `references/requirement-template.md` 获取完整文档模板。
## EPIC 到用户故事的分解
### 层级结构
```
EPIC业务计划
├── 需求 1功能/能力)
│ ├── 用户故事 1.1
│ ├── 用户故事 1.2
│ └── 用户故事 1.3
├── 需求 2功能/能力)
│ ├── 用户故事 2.1
│ └── 用户故事 2.2
└── 需求 3功能/能力)
└── 用户故事 3.1
```
### 分解流程
1. **从 EPIC 开始** - 识别高层次业务计划,定义愿景和目标
2. **分解为需求** - 将 EPIC 分解为逻辑功能或能力
3. **将需求分解为用户故事** - 从用户角度编写用户故事,使用 INVEST 标准
4. **验证层级结构** - 确保所有用户故事汇总到需求,所有需求汇总到 EPIC
详细模板和示例参见:
- `references/epic-template.md` - EPIC 文档模板
- `references/user-story-template.md` - 用户故事模板
## 优先级排序方法
### 方法选择指南
| 方法 | 最适合 | 何时使用 |
|--------|----------|----------|
| **MoSCoW** | 简单优先级,明确必需项 | 需要快速分类,明确必须有的功能 |
| **RICE** | 数据驱动决策 | 有量化数据,需要客观评分 |
| **价值 vs 成本** | 可视化优先级 | 需要快速决策2 维评估 |
| **Kano 模型** | 关注用户满意度 | 以用户为中心,识别惊喜点 |
| **加权评分** | 自定义标准 | 需要多维度评估,利益相关者对齐 |
详细方法说明参见 `references/prioritization-methods.md`
## 最佳实践
1. **主动倾听** - 提出开放式问题,复述以确认理解,不要假设
2. **迭代细化** - 从高层次理解开始,逐步添加细节
3. **文档标准** - 使用一致的模板,包含所有必要信息
4. **利益相关者管理** - 尽早识别所有利益相关者,保持定期沟通
5. **可追溯性** - 将用户故事链接到需求,将需求链接到 EPIC
6. **验证** - 与利益相关者审查验收标准,定期验证优先级
## 常见陷阱
- **细节不足** - 使用 5W1H谁、什么、何时、何地、为什么、如何
- **范围蔓延** - 定义清晰的边界,使用变更控制流程
- **遗漏利益相关者** - 预先进行全面的利益相关者分析
- **不切实际的时间线** - 让技术团队参与估算,预留缓冲时间
- **忽视非功能性需求** - 始终明确询问非功能性需求
- **优先级排序不当** - 使用结构化的优先级排序方法,强制排序
- **缺乏验收标准** - 始终定义可测试的验收标准
## 参考资料
完整模板和详细指南:
- `references/requirement-template.md` - 需求文档完整模板
- `references/epic-template.md` - EPIC 文档模板
- `references/user-story-template.md` - 用户故事模板
- `references/prioritization-methods.md` - 优先级排序方法详解
- `references/complete-example.md` - 完整需求分析示例

View File

@ -0,0 +1,6 @@
{
"ownerId": "kn7795tvpv9b8mq9wvh01g2e3n8167q2",
"slug": "requirements-analysis",
"version": "1.0.0",
"publishedAt": 1772418037477
}

View File

@ -0,0 +1,220 @@
# 完整需求分析示例
## 初始请求
```
用户:"我们需要一个管理者仪表板"
```
## 分析对话
### 第 1 轮:基本理解
```
AI"我会帮你分析这个需求。让我先问几个问题:
1. 管理者面临什么具体问题,这个仪表板将解决?
2. 谁在请求这个仪表板?
3. 这是用于现有系统还是新系统?"
用户:"管理者无法轻松查看团队绩效指标。运营副总裁在请求它。这是用于我们现有的项目管理系统。"
```
### 第 2 轮:利益相关者识别
```
AI"很好,让我了解利益相关者:
1. 谁是将使用这个仪表板的管理者?
2. 除了运营副总裁,还有谁需要参与?
3. 是否有合规或安全要求?"
用户:"所有部门经理(约 50 人)。我们需要让 IT 总监和数据隐私官参与,因为我们要显示员工数据。"
```
### 第 3 轮:详细需求
```
AI"现在让我们深入细节:
1. 管理者需要看到哪些具体指标?
2. 数据应该多久更新一次?
3. 管理者需要过滤或自定义视图吗?
4. 是否有性能要求?"
用户:"他们需要看到:项目完成率、团队利用率、预算 vs 实际、即将到来的截止日期。数据应该每天更新。是的,他们应该能按团队、项目和日期范围过滤。仪表板应该在 3 秒内加载。"
```
### 第 4 轮:时间线和依赖关系
```
AI"让我们讨论时间:
1. 你什么时候需要这个仪表板?
2. 是否依赖其他项目?
3. 是否有特定事件或截止日期驱动这个需求?"
用户:"我们需要在第二季度末6 月 30 日之前完成。它依赖于数据仓库项目首先完成。CEO 想在 7 月审查第二季度绩效,所以我们需要准备好。"
```
### 第 5 轮:验收标准
```
AI"最后,让我们定义成功:
1. 我们如何知道这个仪表板满足你的需求?
2. 哪些是必须有的功能,哪些是最好有的?
3. 你将如何测试?"
用户:"成功意味着管理者可以做出数据驱动的决策,而无需向 IT 请求报告。必须有:我提到的 4 个指标、过滤和导出到 Excel。最好有深入到单个项目、趋势图表。我们将首先与 5 位试点经理测试。"
```
---
## 最终需求文档
# 需求:管理者绩效仪表板
## 概述
- **ID**REQ-2026-001
- **状态**:已批准
- **优先级**:高
- **创建日期**2026-02-10
- **最后更新**2026-02-10
## 利益相关者
- **业务负责人**运营副总裁Jane Smith
- **最终用户**部门经理50 位用户)
- **技术负责人**IT 总监John Doe
- **其他利益相关者**:数据隐私官(合规审查)
## 业务背景
### 问题陈述
部门经理目前缺乏对团队绩效指标的可见性,必须向 IT 请求自定义报告,导致决策延迟和 IT 资源使用效率低下。
### 业务价值
- 将 IT 报告请求量减少 80%
- 实现实时数据驱动决策
- 提高经理生产力 20%
- 支持 CEO 的第二季度绩效审查流程
### 成功指标
- 1 个月内 90% 经理采用率
- IT 报告请求减少 80%
- < 3 秒仪表板加载时间
- 85% 用户满意度评分
## 需求详情
### 功能性需求
#### 核心指标显示
1. **项目完成率**
- 显示按时完成的项目百分比
- 按团队和整体显示
- 颜色编码:绿色(>90%、黄色70-90%)、红色(<70%
2. **团队利用率**
- 显示团队容量利用百分比
- 按团队成员和团队平均值显示
- 包括可计费 vs 不可计费分解
3. **预算 vs 实际**
- 按项目显示预算差异
- 显示为百分比和绝对值
- 如果差异 > 10% 则警报
4. **即将到来的截止日期**
- 列出未来 30 天内有截止日期的项目
- 按紧急程度排序
- 以红色突出显示逾期项目
#### 过滤和自定义
- 按团队过滤(多选)
- 按项目过滤(多选)
- 按日期范围过滤(最近 7/30/90 天,自定义范围)
- 保存每个用户的过滤偏好
- 重置为默认视图
#### 数据导出
- 导出到 Excel.xlsx 格式)
- 包含基于当前过滤器的所有可见数据
- 保持格式和颜色编码
#### 最好有的功能
- 深入到单个项目详情
- 趋势图表(显示随时间变化的指标的折线图)
- 电子邮件定时报告
### 非功能性需求
- **性能**
- 仪表板加载时间 < 3
- 过滤器应用 < 1
- 支持 50 个并发用户
- **安全**
- 基于角色的访问控制(经理只能看到他们的团队)
- 数据访问审计日志
- 仅 HTTPS
- **数据新鲜度**
- 数据每天早上 6 点更新
- 显示最后更新时间戳
- **可用性**
- 移动响应式设计
- 无障碍(符合 WCAG 2.1 AA
- 无需培训的直观界面
- **可用性**
- 工作时间(早上 6 点 - 晚上 8 点99% 正常运行时间
- 计划维护窗口:周日凌晨 2-4 点
## 验收标准
1. **Given** 我是一名经理,**When** 我登录仪表板,**Then** 我看到截至今天早上 6 点更新的我的团队绩效指标
2. **Given** 我正在查看仪表板,**When** 我应用过滤器(团队、项目、日期范围),**Then** 指标在 1 秒内更新以反映过滤后的数据
3. **Given** 我正在查看过滤后的数据,**When** 我点击"导出到 Excel"**Then** 下载一个 Excel 文件,包含所有可见数据并保持颜色编码
4. **Given** 我是一名经理,**When** 我尝试查看另一个团队的数据,**Then** 我被拒绝访问并看到适当的错误消息
5. **Given** 仪表板正在加载,**When** 页面加载时,**Then** 所有指标在 3 秒内可见
6. **Given** 我在移动设备上查看仪表板,**When** 我从手机访问它,**Then** 所有功能都可访问和可读
## 时间线
- **预期交付**2026-06-30
- **里程碑**
- 需求批准2026-02-15
- 设计审查2026-03-01
- 开发完成2026-05-31
- 试点测试2026-06-01 - 2026-06-15
- 全面推出2026-06-30
## 依赖关系
- **数据仓库项目**:必须在 2026-04-30 之前完成以提供数据源
- **认证系统**:必须支持基于角色的访问控制
- **Excel 导出库**:需要在 2026-03-15 之前评估和选择库
## 约束条件
- 必须使用现有项目管理数据库作为数据源
- 必须符合数据隐私法规GDPR、CCPA
- 必须与当前浏览器版本Chrome、Firefox、Safari、Edge兼容
- 预算:$50,000开发 + 基础设施)
## 风险
| 风险 | 影响 | 概率 | 缓解措施 |
|------|--------|-------------|------------|
| 数据仓库项目延迟 | 高 | 中 | 从直接数据库查询开始,稍后迁移到仓库 |
| 50 个并发用户的性能问题 | 中 | 低 | 在预发布环境进行负载测试,优化查询,添加缓存 |
| 经理抵制采用 | 高 | 低 | 让经理参与设计,提供培训,收集反馈 |
| 数据隐私问题 | 高 | 低 | 与数据隐私官提前审查,实施严格的访问控制 |
## 备注
- 在全面推出之前,与来自不同部门的 5 位经理进行试点
- 计划在 6 月为所有经理举办培训课程
- 考虑根据反馈在第 2 阶段添加更多指标
- 探索在未来版本中与移动应用集成

View File

@ -0,0 +1,60 @@
# EPIC 文档模板
## EPIC[标题]
### 愿景
[高层次业务目标或能力]
### 业务目标
- [目标 1]
- [目标 2]
### 目标用户
[用户画像或细分]
### 成功指标
- [KPI 1][目标]
- [KPI 2][目标]
### 时间线
- **开始日期**[日期]
- **目标完成**[日期]
- **预计时长**[周/月]
### 需求列表
1. [需求 1]
2. [需求 2]
3. [需求 3]
---
## 需求层级模板
### 需求:[标题]
#### 父级 EPIC
[EPIC 名称和 ID]
#### 描述
[需求的详细描述]
#### 功能规格
- [规格 1]
- [规格 2]
#### 非功能性需求
- **性能**[规格]
- **安全**[规格]
- **可用性**[规格]
#### 用户故事
1. [用户故事 1]
2. [用户故事 2]
3. [用户故事 3]
#### 依赖关系
- [依赖 1]
#### 验收标准
- [标准 1]
- [标准 2]

View File

@ -0,0 +1,208 @@
# 优先级排序方法详解
## 方法 1MoSCoW 优先级排序
### 分类
- **必须有Must Have**:发布的关键功能,不可协商
- **应该有Should Have**:重要但不关键,必要时可以推迟
- **可以有Could Have**:最好有,时间允许时包含
- **不会有Won't Have**:本次发布不在范围内
### 流程
1. 列出所有需求
2. 对每个需求进行分类
3. 在每个类别内按重要性排序
4. 与利益相关者验证
### 示例
```
必须有:
1. 使用邮箱/密码的用户登录
2. 密码重置功能
3. 基本用户资料
应该有:
1. 社交登录Google、Facebook
2. 双因素认证
3. 头像上传
可以有:
1. 使用手机号登录
2. 生物识别认证
3. 活动日志
不会有(本次发布):
1. 单点登录SSO
2. LDAP 集成
3. OAuth 提供商
```
---
## 方法 2RICE 评分
### 公式
RICE = (覆盖面 × 影响力 × 信心度) / 工作量
### 组成部分
- **覆盖面Reach**:这将影响多少用户?(每季度)
- **影响力Impact**:对每个用户的影响有多大?
- 3 = 巨大
- 2 = 高
- 1 = 中
- 0.5 = 低
- 0.25 = 最小
- **信心度Confidence**:我们对估算有多自信?
- 100% = 高
- 80% = 中
- 50% = 低
- **工作量Effort**:这需要多少人月?
### 流程
1. 对每个需求在所有四个维度上评分
2. 计算 RICE 分数
3. 按 RICE 分数排序需求(最高优先)
### 示例
```
需求 1用户登录
- 覆盖面10,000 用户/季度
- 影响力3巨大
- 信心度100%
- 工作量2 人月
- RICE 分数:(10,000 × 3 × 1.0) / 2 = 15,000
需求 2社交登录
- 覆盖面5,000 用户/季度
- 影响力2
- 信心度80%
- 工作量1 人月
- RICE 分数:(5,000 × 2 × 0.8) / 1 = 8,000
需求 3双因素认证
- 覆盖面10,000 用户/季度
- 影响力2
- 信心度90%
- 工作量1.5 人月
- RICE 分数:(10,000 × 2 × 0.9) / 1.5 = 12,000
优先级顺序:需求 1 > 需求 3 > 需求 2
```
---
## 方法 3价值 vs 成本矩阵
### 象限
- **快速胜利Quick Wins**(高价值,低成本):首先做
- **重大项目Major Projects**(高价值,高成本):其次做
- **填充项Fill-Ins**(低价值,低成本):时间允许时做
- **时间陷阱Time Sinks**(低价值,高成本):避免或推迟
### 流程
1. 对每个需求的价值评分1-10
2. 对每个需求的成本评分1-10
3. 在 2×2 矩阵上绘制
4. 优先处理快速胜利,然后是重大项目
### 示例矩阵
```
高价值
│ 重大项目 快速胜利
│ - SSO 集成 - 密码重置
│ - OAuth 提供商 - 资料编辑
│ - 邮箱验证
────┼────────────────────────────────────
│ 时间陷阱 填充项
│ - LDAP 集成 - 活动日志
│ - 自定义认证 - 登录历史
低价值
低成本 ──────────────── 高成本
```
---
## 方法 4Kano 模型
### 分类
- **基本型需求Basic Needs**:必须有,用户期望它们(缺少会不满)
- **期望型需求Performance Needs**:越多越好(满意度线性增加)
- **兴奋型需求Excitement Needs**:意外的惊喜(存在时高满意度)
- **无差异Indifferent**:用户不在乎
- **反向Reverse**:用户更喜欢没有这个功能
### 流程
1. 通过功能性和非功能性问题调查用户
2. 对每个功能进行分类
3. 优先级:基本型需求 → 期望型需求 → 兴奋型需求
### 示例
```
基本型需求(必须有):
- 用户登录
- 密码安全
- 账户恢复
期望型需求(应该有):
- 快速登录(< 2
- 多种登录选项
- 会话管理
兴奋型需求(可以有):
- 生物识别登录
- 无密码认证
- 社交登录
无差异:
- 登录自定义主题
```
---
## 方法 5加权评分
### 流程
1. 定义评估标准(例如:业务价值、技术可行性、用户影响、战略一致性)
2. 为每个标准分配权重(总计 = 100%
3. 对每个需求在每个标准上评分1-10
4. 计算加权分数
5. 按加权分数排序
### 示例
```
标准权重:
- 业务价值40%
- 用户影响30%
- 技术可行性20%
- 战略一致性10%
需求 1用户登录
- 业务价值10 × 0.4 = 4.0
- 用户影响10 × 0.3 = 3.0
- 技术可行性8 × 0.2 = 1.6
- 战略一致性9 × 0.1 = 0.9
- 总分9.5
需求 2社交登录
- 业务价值7 × 0.4 = 2.8
- 用户影响8 × 0.3 = 2.4
- 技术可行性6 × 0.2 = 1.2
- 战略一致性7 × 0.1 = 0.7
- 总分7.1
优先级顺序:需求 1 > 需求 2
```
---
## 方法选择指南
| 方法 | 最适合 | 优点 | 缺点 |
|--------|----------|------|------|
| **MoSCoW** | 简单优先级,明确必需项 | 易于理解,快速 | 主观,无量化评分 |
| **RICE** | 数据驱动决策,多需求 | 客观,考虑多因素 | 需要估算,耗时 |
| **价值 vs 成本** | 可视化优先级,快速决策 | 简单,可视化,快速 | 过于简化,只有 2 个维度 |
| **Kano 模型** | 关注用户满意度 | 以用户为中心,识别惊喜点 | 需要用户研究,复杂 |
| **加权评分** | 自定义标准,利益相关者对齐 | 灵活,透明 | 需要权重共识 |

View File

@ -0,0 +1,63 @@
# 需求文档模板
## 概述
- **ID**REQ-001
- **状态**:草稿/已批准/进行中/已完成
- **优先级**:高/中/低
- **创建日期**[日期]
- **最后更新**[日期]
## 利益相关者
- **业务负责人**[姓名、部门]
- **最终用户**[用户画像]
- **技术负责人**[姓名、团队]
- **其他利益相关者**[列表]
## 业务背景
### 问题陈述
[这解决了什么问题?]
### 业务价值
[为什么这很重要?]
### 成功指标
- [指标 1][目标]
- [指标 2][目标]
## 需求详情
### 功能性需求
1. [需求 1]
2. [需求 2]
### 非功能性需求
- **性能**[规格]
- **安全**[要求]
- **可扩展性**[期望]
- **可用性**[SLA]
## 验收标准
1. Given [前置条件], When [操作], Then [结果]
2. Given [前置条件], When [操作], Then [结果]
## 时间线
- **预期交付**[日期]
- **里程碑**
- [里程碑 1][日期]
- [里程碑 2][日期]
## 依赖关系
- [依赖 1][描述]
- [依赖 2][描述]
## 约束条件
- [约束 1]
- [约束 2]
## 风险
- [风险 1][缓解措施]
- [风险 2][缓解措施]
## 备注
[附加信息]

View File

@ -0,0 +1,48 @@
# 用户故事模板
## 用户故事:[标题]
### 故事
作为 [用户角色]
我想要 [操作]
以便 [收益]
### 父级需求
[需求名称和 ID]
### 验收标准
1. Given [前置条件], When [操作], Then [结果]
2. Given [前置条件], When [操作], Then [结果]
3. Given [前置条件], When [操作], Then [结果]
### 技术说明
- [实现细节 1]
- [实现细节 2]
### 估算
- **故事点数**[点数]
- **预计小时数**[小时]
### 依赖关系
- [依赖 1]
### 完成定义
- [ ] 代码完成并审查
- [ ] 单元测试编写并通过
- [ ] 集成测试通过
- [ ] 文档更新
- [ ] 部署到预发布环境
- [ ] 验收标准验证
---
## INVEST 标准检查清单
用户故事应该满足 INVEST 标准:
- **I - Independent独立**:故事应该尽可能独立,可以任意顺序开发
- **N - Negotiable可协商**:故事不是合同,细节可以协商
- **V - Valuable有价值**:故事必须为用户或业务提供价值
- **E - Estimable可估算**:团队必须能够估算故事的大小
- **S - Small小型**:故事应该足够小,可以在一个迭代中完成
- **T - Testable可测试**:故事必须有明确的验收标准,可以测试

View File

@ -4,12 +4,20 @@
"""
# 列举导入模块
from typing import List
from pydantic_ai import Agent, AgentRunResult, ModelMessage
from pathlib import Path
import sys
from typing import List, Optional
from uuid import uuid4
from starlette.applications import Starlette
from pydantic_ai import Agent, AgentRunResult
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.output import OutputSpec
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai_skills import SkillsToolset
sys.path.append(Path(__file__).parent.as_posix())
from memory import Memory
class BaseAgent:
@ -19,27 +27,52 @@ class BaseAgent:
2异步运行
"""
def __init__(self, instructions: str, output_type: OutputSpec = str):
def __init__(
self,
instructions: str,
output_type: OutputSpec = str,
skill_name: Optional[str] = None,
):
"""
初始化智能体
:param instructions: 指令
:param output_type: 输出类型
:param skill_name: 技能名称默认不使用技能
:return: 智能体实例
"""
# 生成会话唯一标识
self.session_id = uuid4().hex.lower()
# 实例智能体
self.agent = self._instantiate(
self.agent = self._instantiate_agent(
skill_name=skill_name,
instructions=instructions,
output_type=output_type,
)
# 实例记忆体
self.memory = Memory()
def _instantiate(self, instructions: str, output_type: OutputSpec) -> Agent:
def _instantiate_agent(
self, skill_name: Optional[str], instructions: str, output_type: OutputSpec
) -> Agent:
"""
实例智能体
:param skill_name: 技能名称
:param instructions: 指令
:param output_type: 输出类型
:return: 智能体实例
"""
# 若技能名称为空则技能集合为 None若技能名称非空则构建技能路径
if not skill_name:
toolsets = None
else:
toolsets = [
SkillsToolset(
directories=[Path(__file__).parent.parent / "skills" / skill_name]
)
]
agent = Agent(
toolsets=toolsets,
model=OpenAIChatModel(
model_name="deepseek-v4-flash",
provider=OpenAIProvider(
@ -52,15 +85,27 @@ class BaseAgent:
)
return agent
async def run(
self, user_prompt: str | List[str], message_history: List[ModelMessage] = []
) -> AgentRunResult:
async def run(self, user_prompt: str | List[str]) -> AgentRunResult:
"""
异步运行
:param user_prompt: 用户提示词
:param message_history: 历史消息
:return: 智能体回复
"""
return await self.agent.run(
# 查询会话历史消息
message_history = self.memory.read(session_id=self.session_id)
result = await self.agent.run(
user_prompt=user_prompt, message_history=message_history
)
# 记录会话历史消息
self.memory.create(
session_id=self.session_id,
dialogue_message=result.new_messages(),
)
return result
def to_web(self) -> Starlette:
"""
实例 Starlette 应用
:return: Starlette 应用实例
"""
return self.agent.to_web()

View File

@ -43,8 +43,6 @@ class Memory(SQLite):
id TEXT PRIMARY KEY,
--会话唯一标识
session_id TEXT NOT NULL,
--对话轮次
dialogue_round INTEGER NOT NULL,
--对话消息
dialogue_message TEXT NOT NULL,
--时间戳毫秒
@ -55,13 +53,10 @@ class Memory(SQLite):
except Exception as exception:
raise RuntimeError(f"初始化记忆体发生异常:{str(exception)}") from exception
def create(
self, session_id: str, dialogue_round: int, dialogue_message: List[ModelMessage]
) -> bool:
def create(self, session_id: str, dialogue_message: List[ModelMessage]) -> bool:
"""
新增对话消息
:param session_id: 会话唯一标识
:param dialogue_round: 对话轮次
:param dialogue_message: 对话消息
:return: 新增是否成功
"""
@ -69,12 +64,11 @@ class Memory(SQLite):
with self:
return self.execute(
sql="""
INSERT INTO messages (id, session_id, dialogue_round, dialogue_message, timestamp) VALUES (?, ?, ?, ?, ?)
INSERT INTO messages (id, session_id, dialogue_message, timestamp) VALUES (?, ?, ?, ?)
""",
parameters=(
uuid4().hex.lower(),
session_id,
dialogue_round,
ModelMessagesTypeAdapter.dump_json(dialogue_message),
int(time.time() * 1000),
),