Compare commits

...

14 Commits

Author SHA1 Message Date
Liu Zhicong f0bdd26b08 build first real PRD for v1.0 2026-03-28 11:08:58 -07:00
Liu Zhicong dfcc5efaa0 disable vector-base knowledge base. 2024-08-19 12:20:56 -07:00
Liu Zhicong c1f3ae4fea fix bug : there could be a default LLM_MODEL_NAME when no system config exist. 2024-08-12 16:37:19 -07:00
Liu Zhicong f0737e3882 change default llm_model to GPT4o 2024-05-13 17:44:46 -07:00
Liu Zhicong a30ca32441 fix agent config bug. 2024-05-10 06:17:28 -07:00
Liu Zhicong 7b128c06c2 Support llm model name link.
1. default
2. plan_llm
3. outline_llm
4. swift_llm
2024-04-23 09:26:44 -07:00
Liu Zhicong 51998d841a Merge branch 'MVP' 2024-04-23 04:56:48 -07:00
Liu Zhicong af1bb80283 update kb design, base on objfs and knowled graph. 2024-04-23 04:56:22 -07:00
Liu Zhicong 0a2f5db91b Merge pull request #125 from fiatrete/MVP
MVP
2024-04-23 19:53:04 +08:00
Liu Zhicong 342464e386 rollback agent memory to "chat session history & session summary" 2024-03-20 18:47:58 -07:00
Liu Zhicong 818b52001c Merge pull request #120 from RISHIKREDDYL/patch-1
Update README.md
2024-02-06 16:08:51 -08:00
Rishik Reddy c98c596d3b Update README.md 2024-02-06 22:07:45 +05:30
Liu Zhicong 4f6a7efbd4 Merge pull request #116 from seabornlee/patch-1
Update README.md
2024-01-16 20:21:06 -08:00
Seaborn 20b37fa716 Update README.md 2024-01-17 12:09:01 +08:00
48 changed files with 2963 additions and 2309 deletions
+2 -2
View File
@@ -123,12 +123,12 @@ Say Hello to your private AI assistant Jarvis !
1. **AI Agent**: Driven by a large language model, having own memory.The AI Agent completes tasks through natural language interaction. 1. **AI Agent**: Driven by a large language model, having own memory.The AI Agent completes tasks through natural language interaction.
2. **AI Workflow**: Organize different AI Agents into an AI Agent Group to complete complex tasks. 2. **AI Workflow**: Organize different AI Agents into an AI Agent Group to complete complex tasks.
![workflow](./doc/res/workflow.png) ![workflow](./doc/res/workflow.png)
3. **AI Envriment**: Supports AI Agents to access file systems, IoT devices, network services, smart contracts, and everything on today's internet once authorized. 3. **AI Environment**: Supports AI Agents to access file systems, IoT devices, network services, smart contracts, and everything on today's internet once authorized.
4. **AI Marketplace**: Offer a solution for one-click installation and use of various AI applications, helping users easily access and manage AI apps. 4. **AI Marketplace**: Offer a solution for one-click installation and use of various AI applications, helping users easily access and manage AI apps.
5. **AI Model Solution**: Provide a unified entry point for model search, download, and access control, making it convenient for users to find and use models suitable for their needs. 5. **AI Model Solution**: Provide a unified entry point for model search, download, and access control, making it convenient for users to find and use models suitable for their needs.
6. **Hardware-specific optimization**: Optimize for specific hardware to enable smooth local running of most open-source AI applications. 6. **Hardware-specific optimization**: Optimize for specific hardware to enable smooth local running of most open-source AI applications.
7. **Strict Privacy Protection and Management**: Strictly manage personal data, ranging from family albums to chat records and social media records, and provide a unified access control interface for AI applications. 7. **Strict Privacy Protection and Management**: Strictly manage personal data, ranging from family albums to chat records and social media records, and provide a unified access control interface for AI applications.
8. **Personal Knowlege Base**: 8. **Personal knowledge Base**:
9. **Integrated AIGC Workflow**: Offer AIGC Agent/Workflow for users to train their own voice models, Lora models, knowledge models, etc., using personal data. Based on these private model data, integrate the most advanced AIGC algorithm to help people release creativity easily and build more COOL and more personalized content. 9. **Integrated AIGC Workflow**: Offer AIGC Agent/Workflow for users to train their own voice models, Lora models, knowledge models, etc., using personal data. Based on these private model data, integrate the most advanced AIGC algorithm to help people release creativity easily and build more COOL and more personalized content.
10. **Development Framework**: Provide a development framework for customizing AI assistants for specific purposes, making it easy for developers to create unique AI applications / service for their customers. 10. **Development Framework**: Provide a development framework for customizing AI assistants for specific purposes, making it easy for developers to create unique AI applications / service for their customers.
+34
View File
@@ -0,0 +1,34 @@
# agent memory
## memory的基本形式
memory的基本形式上是 topic+内容
topic用一个有意义的路径表示 /xxx/xxx/xxx (有点类似脑图的逻辑,可以通过逐级展开遍历浏览所有的memory)
同一个memory可以被多个路径指向
内容则是一个json文件
## Agent 使用memory的
1. 根据当前会话的主题,尝试在known_info中加载必要的memory
2. 提供memory的 list/查询 函数, 允许agent在必要的时候 list / 查询memory
该使用逻辑的本质和kb查询逻辑很像
## Agent 更新/创建memory
1. 在任何llm process的过程中,agent都可以用写文件的形式创建memory
2. 更新memory通常是一个专门的 self-think过程,agent此时会用某种模式整理自己所有的logs和memory,并对memory进行更新、创建、删除
该更新逻辑与Agent 与KB的Self-learning逻辑很像。但根据log->summary的过程基本上是 self-think独有的
## 实现逻辑
基本思路:
1. 核心API是一组通用的文件操作API(有些场景可以是只读的) + 一组特化的对象查询API
路径->ObjectObject中包含ObjectId等信息
ObjectId->Object
Object一定是一个json,里面包含可以打开原始文件的路径(fileId)
2. 通过一组文件系统描述来引导Agent操作特定文件
3. 通过一组搜索API来引导Agent操作特定文件
对象查询API,基本思路是
ObjectId->Object
+231
View File
@@ -0,0 +1,231 @@
# Building Data Scraping Skills for OpenDAN
> 为 OpenDAN Agent 开发数据抓取能力:第三方开发者指南
---
## 0. 为什么在 OpenDAN 上做,而不是自己写脚本?
你当然可以自己写一个 Python 爬虫跑在 crontab 里。但你大概已经发现了这些问题:
| 你遇到的问题 | 自己写脚本 | n8n / Make 等自动化平台 | OpenDAN |
|---|---|---|---|
| 页面结构变了,脚本挂了 | 手动修 selector,重新部署 | 同上,workflow 断掉需要人工介入 | Agent 有上下文理解能力,可以根据 Skill 里的判断标准自行调整抓取策略 |
| 需要过登录墙 / 验证码 | 自己集成 Playwright + 各种 hack | 平台提供的浏览器能力有限 | Agent 可以调用浏览器能力,像人一样操作页面 |
| 抓到的数据只有自己能用 | 数据格式随意,换个项目又得重新写 | 平台内部闭环,数据导出麻烦 | 统一 Schema,本地所有 Agent 共享同一份数据 |
| 想加 AI 处理(摘要/分类/理解) | 自己对接 API,管 token、管 key | 需要额外付费接 AI 节点 | AICC 统一模型调用,开发者不用管供应商差异 |
| 多个抓取任务需要协同 | 自己写调度逻辑 | 平台限制了组合方式 | 多个 Skill 天然可组合,Agent 按目标自主调度 |
| 想让别人也能用你的能力 | 发个 GitHub repo,用户自行部署 | 发布到平台 marketplace,但受平台规则约束 | 发布 Skill + Tool 即可,其他用户的 Agent 直接调用 |
**一句话:你写的不是一个跑一次就扔的脚本,而是一个可以被任何 Agent 反复使用的能力。**
传统爬虫是"写死的流程"——一旦目标网站变化,流程就断。
OpenDAN Skill 是"给 Agent 的操作说明"——Agent 根据说明自主决策,遇到异常可以自行调整。
这意味着你的工作成果有更长的生命周期,也有更大的复用面。
---
## 1. 你要做的三件事
为 OpenDAN Agent 提供数据抓取能力,只需要准备三样东西:
**Tool** — 真正执行抓取的程序。CLI 工具、脚本、小程序都行。
**Skill** — 告诉 Agent 怎么用这些 Tool 的操作说明。不是自动化脚本,是"目标 + 工具 + 判断标准"的组合。
**Schema** — 抓取结果的数据格式。保证不同开发者、不同 Agent 之间的数据可以互通。
---
## 2. 从 Tool 开始,先跑通一个最小抓取
### 选语言
推荐 **TypeScript / tsx**Python 也可以。
### 写什么
你的 Tool 就是一个普通 CLI 程序,做的事情和你过去写爬虫一样:
- 调 API
- 跑 Playwright / 无头浏览器
- 解析页面
- 输出结构化 JSON
### 怎么发布
第一版完全不需要学习 OpenDAN / BuckyOS 的包管理体系。
你过去怎么发 npm 包,现在就怎么发:
- `npx`
- `pnpm`
- 或者就是本地脚本
等跑通以后,再考虑正式打包。
---
## 3. 写 Skill:给 Agent 的操作说明书
Skill 不是"严格一步不差的自动化脚本"。
它更像是你写给一个聪明但不了解具体业务的同事的操作指南。
一个好的 Skill 要写清楚以下内容:
### 目标
> 给定一个用户账号,抓取其公开资料、帖子、媒体内容。
或者:
> 给定一个商品链接,抓取其标题、价格、评论、历史价格。
### 可用工具
列出 Agent 可以调用的 Tool,例如:
- `fetch-x-profile` — 抓取 X 用户资料
- `fetch-x-posts` — 抓取 X 用户帖子
- `fetch-instagram-media` — 抓取 Instagram 媒体
### 判断标准
Agent 需要知道每一步"做对了"是什么样子:
- profile 拿到了且字段完整 → 第一步成功
- 帖子列表为空 → 需要判断是"真的没有"还是"抓取失败"
- 页面返回登录墙 → 切换到浏览器模式重试
### 常见错误和修正建议
- 账号不存在 → 停止,返回明确错误
- 页面结构变化 → 尝试用浏览器模式重新抓取
- 请求频率过高 → 降速重试
- 需要登录 → 提示用户提供凭据或切换策略
**核心原则:给方向、给工具、给判断方法。** Agent Loop 自身有纠错能力,Skill 的重点不是"零错误执行",而是让 Agent 知道目标在哪、手里有什么、怎么判断是否在逼近目标。
---
## 4. 定义数据格式:目录结构就是数据库
抓取结果必须有统一结构,否则不同 Skill、不同 Agent 之间没法复用。
推荐用目录结构直接组织数据:
```
data/
x/
alice/
profile.json # 用户资料
posts/
post_001.json
post_002.json
media/
img_001.jpg
instagram/
alice/
profile.json
posts/
media/
bindings/
person_alice.json # 跨平台身份绑定
```
**规则很简单:**
- `{platform}/{account_id}/` 是基本单元
- `profile.json` 存用户资料
- `posts/` 存帖子
- `media/` 存媒体文件
- `bindings/` 把同一个人在不同平台的账号关联起来
### 统一格式带来的两个直接好处
**多 Skill 协同补数据。** 一个 Skill 抓 profile,一个 Skill 抓 posts,一个 Skill 抓 comments——最终都落到同一个目录结构里,互不干扰。
**不同 Agent 复用结果。** Agent A 已经抓过的数据,Agent B 直接读取就行,不用重复抓。在 OpenDAN / BuckyOS 体系里,Agent 可以暴露服务接口,这意味着同一套 Skill 不只是"教 Agent 怎么抓",还会自然形成**抓取结果的共享网络**。
---
## 5. 什么时候需要用 OpenDAN 的平台能力
不是所有场景都需要学额外的东西。按你的需求分三档:
### 情况 A:普通抓取 → 不需要任何平台能力
如果你的 Tool 只做 HTTP 请求、页面解析、Playwright 抓取、文件处理,就按普通程序写,不用学任何额外概念。
### 情况 BTool 内部需要 AI → 接入 AICC
如果你想在工具内部做语义解析、内容摘要、评论分类、图像理解,接入 AICC 即可。
AICC 的作用是统一模型调用。你给模型名、给 prompt、拿结果。不用自己折腾不同模型供应商的接入。
**只有 Tool 里需要 AI 时,才需要了解 AICC。**
### 情况 C:必须操作浏览器 / 桌面 → 使用 AgentPC
有些网站没有 API 或反爬很强,需要浏览器自动化:获取页面截图、点击、输入、滚动、观察结果、循环执行。
这时使用 OpenDAN 的 AgentPC / 浏览器能力。如果流程特别复杂,建议把浏览器操作任务交给专门的 Agent 处理,而不是全塞进一个 Tool。
---
## 6. 最小可用交付物
你的第一版只需要做到这样:
```
my-scraping-skill/
skill.md # 操作说明书
tools/
fetch-x-profile.ts # CLI 工具
fetch-x-posts.ts
schemas/
profile.schema.json # 数据结构定义
post.schema.json
data-example/ # 实际结果示例
x/
alice/
profile.json
posts/
post_001.json
```
各部分职责:
- **`skill.md`** — 适用范围、输入参数、推荐工具、成功判断、常见错误、输出目录
- **`tools/`** — 你的 CLI 工具或脚本
- **`schemas/`** — 你约定的数据结构(JSON Schema
- **`data-example/`** — 给其他开发者看的实际结果示例
---
## 7. 推荐上手顺序
不要一开始做"大而全"。按这个顺序最容易跑通:
1. **选一个网站** — 比如 X、Instagram、Amazon 其中一个
2. **先定义结果格式** — 想清楚 profile、post、media 怎么存
3. **写一个最小 Tool** — 先抓到最核心的一种数据
4. **写一个最小 Skill** — 把目标、工具、判断标准写清楚
5. **跑通一次完整流程** — 输入一个账号,输出一套目录
6. **再补充能力** — 评论、图片、视频、跨平台绑定
---
## 8. 总结
为 OpenDAN Agent 提供数据抓取能力,本质上就是:
> **写一个抓取工具,写一份 Agent 能看懂的使用说明,再把结果按统一格式存下来。**
三条核心原则:
- **Tool 负责执行** — 做具体的抓取动作
- **Skill 负责说明** — 告诉 Agent 怎么用、怎么判断
- **Schema 负责复用** — 保证结果能被任何 Agent 直接使用
和传统爬虫的本质区别在于:你写的不是一个只能自己用的脚本,而是一个**可以被整个 Agent 生态复用的能力模块**。
+1
View File
@@ -18,6 +18,7 @@ Let's start by introducing the two important processes.
Note that the dependency check during installation allows for the missing packages to be installed into the current environment. Note that the dependency check during installation allows for the missing packages to be installed into the current environment.
# Some Basic Concepts # Some Basic Concepts
- ***env***:A target environment consisting of a series of configuration files, where packages can be loaded/installed. - ***env***:A target environment consisting of a series of configuration files, where packages can be loaded/installed.
- ***pkg***:A Package(pkg) is either a folder or a file that serves the same purpose as a folder (such as zip, iso, etc.). - ***pkg***:A Package(pkg) is either a folder or a file that serves the same purpose as a folder (such as zip, iso, etc.).
- ***pkg_name***:A unique string used to label a package. It's usually a readable package name, but can also include the version number or even the ContentId. - ***pkg_name***:A unique string used to label a package. It's usually a readable package name, but can also include the version number or even the ContentId.
+15
View File
@@ -0,0 +1,15 @@
# prompts
四个循环
1. 立刻处理消息+深度思考整理循环
Process <-> Self Thinking
2. 任务迭代完整循环致力于完成所有的待完成任务(不一定是成功完成)
Task -> Todo -> Check
3. 知识库整理
New Knowledge -> Self-Learning
使用知识库的时机?是否有quick process和deep thing的区别?
4. Self-Improve
根据四元组:输入,提示词,输出,上级意见 (可选),对提示词进行改进
+1293 -828
View File
File diff suppressed because it is too large Load Diff
+34 -12
View File
@@ -2,7 +2,7 @@ instance_id = "Jarvis"
fullname = "Jarvis" fullname = "Jarvis"
max_token = 4000 max_token = 4000
#timeout = 1800 #timeout = 1800
model_name = "gpt-4-turbo-preview" model_name = "default"
#enable_kb = "true" #enable_kb = "true"
enable_timestamp = "true" enable_timestamp = "true"
enable_json_resp = "true" enable_json_resp = "true"
@@ -12,9 +12,31 @@ Your name is Jarvis, the super personal assistant to the Principal. Help the Pri
Only clearly specifying the task you completed can be completed independently. Only clearly specifying the task you completed can be completed independently.
""" """
kb_query_desc = """
$ introduce (have default config)
$ dir descriptions
$ dir1
$ dir2
$ dir3
$ support actions(if enable)
$ support funcitons(if enable)
现有信息以知识图谱的形式保存在存储系统中。
1. 介绍知识图谱的结构
2. 不同部分的规则说明(可选)创建知识图谱的指导思路(目前不允许AI自创结构)
# read (function)
access_knowledge_graph($op_name,$params)
# write (action)
update_knowledge_graph($op_name,$params)
"""
[behavior.on_message] [behavior.on_message]
type="AgentMessageProcess" type="AgentMessageProcess"
# TODO: 是否应该自动记录 inner function和action的执行细节 #model_name="llm_default_model"
mutil_model="gpt-4-vision-preview" mutil_model="gpt-4-vision-preview"
asr_model="openai-whisper" asr_model="openai-whisper"
tts_model="tts-1" tts_model="tts-1"
@@ -46,8 +68,8 @@ known_info_tips = """
tools_tips = """ tools_tips = """
""" """
llm_context.actions.enable = ["agent.workspace.create_task","agent.workspace.cancel_task"] llm_context.actions.enable = ["agent.workspace.create_task","agent.workspace.cancel_task","knowledge_base.knowledge_graph_update"]
llm_context.functions.enable = ["agent.workspace.list_task"] llm_context.functions.enable = ["agent.workspace.list_task","knowledge_base.knowledge_graph_read"]
[behavior.triage_tasks] [behavior.triage_tasks]
@@ -167,7 +189,8 @@ process_description="""
reply_format = """ reply_format = """
The Response must be directly parsed by `python json.loads`. Here is an example: The Response must be directly parsed by `python json.loads`. Here is an example:
{ {
resp:'$think step by step, how to check the todo', resp: '$simport report about what you do',
actions: [{
name: '$action1_name', name: '$action1_name',
$param_name: '$parm' #Optional, fill in only if the action has parameters. $param_name: '$parm' #Optional, fill in only if the action has parameters.
}, ... }, ...
@@ -183,20 +206,19 @@ llm_context.functions.enable = ["agent.workspace.read_file","agent.workspace.li
# self thing的主要目的是对各种chatlog,worklog进行分析,并更新面向人和事的summary。 # self thing的主要目的是对各种chatlog,worklog进行分析,并更新面向人和事的summary。
type="AgentSelfThinking" type="AgentSelfThinking"
process_description=""" process_description="""
You are very good at thinking and summarizing what you have already happened。Your input is a chat history and work record,After you think about it, you will follow the requirements below to generate abstract. You are very good at thinking and summarizing what you have already happened。Your input is chat history and work records,After you think about it, you will follow the requirements below to generate abstract.
1. Try to understand the theme of each sentence, and call the relevant operation to record the relationship between the dialogue and the theme 1. Try to understand the theme of each sentence, and call the relevant operation to record the relationship between the dialogue and the theme
2. Try to analyze the personality of different people involved in information 2. Try to analyze the personality of different people involved in information
3. Try to summarize important events in the information and record it 3. Try to summarize important events in the information and record it
4. Try to understand the attitude of different people on different topics or events 4. Try to understand the attitude of different people on different topics or events
5. Pay attention to the time order when summarizing, and combine the summary you have done to update Summary 5. Pay attention to the time order when summarizing, and combine the summary you have done to update Summary
6. The summary of the generation cannot exceed 400 token 6. The summary of the generation cannot exceed 500 token
7. 思考的目的是让自己未来的工作更加高效
8. 总结中只包含有长期价值和未完成的事情,已经完成的事情不需要总结
""" """
reply_format = """ reply_format = """
The Response must be directly parsed by `python json.loads`. Here is an example: The Response must be directly parsed by `python json.loads`. Here is an example:
{ {
resp:'$Summary in one sentence', resp: '$simport report about what you do',
actions: [{
name: '$action1_name', name: '$action1_name',
$param_name: '$parm' #Optional, fill in only if the action has parameters. $param_name: '$parm' #Optional, fill in only if the action has parameters.
}, ... }, ...
@@ -204,8 +226,8 @@ The Response must be directly parsed by `python json.loads`. Here is an example:
} }
""" """
context="Your Principal is {owner}, now in {location}, time: {now}, weather: {weather}." context="Your Principal is {owner}, now in {location}, time: {now}, weather: {weather}."
llm_context.actions.enable = ["agent.memory.update_summary","agent.memory.update_contact_summary","agent.memory.update_relation_summary","agent.memory.set_experience"] llm_context.actions.enable = ["agent.memory.update_chat_summary"]
llm_context.functions.enable = ["agent.memory.get_summary","agent.memory.get_contact_summary","agent.memory.list_summary","agent.memory.get_relation_summary","agent.memory.get_experience"]
#[behavior.self_improve] #[behavior.self_improve]
@@ -1,8 +0,0 @@
name = "JarvisPlus"
input.module = "scan_local"
input.params.workspace = "${myai_dir}/workspace/JarvisPlus"
input.params.path = "${myai_dir}/data"
parser.module = "parse_local"
parser.params.workspace = "${myai_dir}/workspace/JarvisPlus"
parser.params.assign_to = "JarvisPlus"
@@ -1,9 +0,0 @@
import sys
import os
from aios import KnowledgePipelineEnvironment
directory = os.path.dirname(__file__)
sys.path.append(directory + '/../../../../src/component/')
from mail_environment import LocalEmail
def init(env: KnowledgePipelineEnvironment, params: dict):
return LocalEmail(env, params)
@@ -1,10 +0,0 @@
import sys
import os
from aios import *
directory = os.path.dirname(__file__)
sys.path.append(directory + '/../../../../src/component/')
from mail_environment import IssueParser
def init(env: KnowledgePipelineEnvironment, params: dict):
return IssueParser(env, params)
@@ -1,13 +0,0 @@
name = "Mail.Issue"
input.module = "local.py"
input.params.path = "${myai_dir}/mail"
input.params.watch = true
parser.module = "parser.py"
parser.params.mail_path = "${myai_dir}/mail"
parser.params.issue_path = "${myai_dir}/mail/issue.json"
[parser.params.root_issue]
summary = "巴克云公司推进中的项目"
[[parser.params.root_issue.children]]
summary = "去中心存储项目DMC"
@@ -1,10 +0,0 @@
import sys
import os
from knowledge import KnowledgePipelineEnvironment
directory = os.path.dirname(__file__)
sys.path.append(directory + '/../../../../src/component/')
from mail_environment import EmailSpider
def init(env: KnowledgePipelineEnvironment, params: dict):
return EmailSpider(env, params)
@@ -1,14 +0,0 @@
name = "Mail.Sync"
input.module = "input.py"
[input.params]
path = "${myai_dir}/mail"
imap_server = "imap.qq.com"
imap_port = 993
address = "115620204@qq.com"
password = "zbbjpbukeonqbjja"
[input.params.fields]
from = "from"
to = "to"
subject = "subject"
-211
View File
@@ -1,211 +0,0 @@
import copy
import os
from typing import List
import aiofiles
import chardet
import logging
import string
import docx2txt
from PyPDF2 import PdfReader
from aios import KnowledgePipelineEnvironment, ImageObjectBuilder, DocumentObjectBuilder, KnowledgeStore, RichTextObject
from aios.agent.agent_base import AgentPrompt
from aios.frame.compute_kernel import ComputeKernel
from aios.knowledge.data.writer import split_text
from aios.proto.compute_task import ComputeTaskResult, ComputeTaskResultCode
from aios.storage.storage import AIStorage
from aios.utils import video_utils, image_utils
class KnowledgeDirSource:
def __init__(self, env: KnowledgePipelineEnvironment, config):
self.env = env
path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
config["path"] = path
self.config = config
# @classmethod
# def user_config_items(cls):
# return [("path", "local dir path")]
def path(self):
return self.config["path"]
@staticmethod
async def read_txt_file(file_path:str)->str:
cur_encode = "utf-8"
async with aiofiles.open(file_path,'rb') as f:
cur_encode = chardet.detect(await f.read())['encoding']
async with aiofiles.open(file_path,'r',encoding=cur_encode) as f:
return await f.read()
async def next(self):
while True:
journals = self.env.journal.latest_journals(1)
from_time = 0
if len(journals) == 1:
latest_journal = journals[0]
if latest_journal.is_finish():
yield None
continue
from_time = os.path.getctime(latest_journal.get_input())
if os.path.getmtime(self.path()) <= from_time:
yield (None, None)
continue
file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x)))
for rel_path in file_pathes:
file_path = os.path.join(self.path(), rel_path)
timestamp = os.path.getctime(file_path)
if timestamp <= from_time:
continue
ext = os.path.splitext(file_path)[1].lower()
if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
logging.info(f"knowledge dir source found image file {file_path}")
image = ImageObjectBuilder({}, {}, file_path).build(self.env.get_knowledge_store())
await self.env.get_knowledge_store().insert_object(image)
yield (image.calculate_id(), file_path)
if ext in ['.txt']:
logging.info(f"knowledge dir source found text file {file_path}")
text = await self.read_txt_file(file_path)
document = DocumentObjectBuilder({}, {}, text).build(self.env.get_knowledge_store())
await self.env.get_knowledge_store().insert_object(document)
yield (document.calculate_id(), file_path)
yield (None, None)
def init(env: KnowledgePipelineEnvironment, params: dict) -> KnowledgeDirSource:
return KnowledgeDirSource(env, params)
async def image_to_text(images: List[str]) -> str:
msg_prompt = AgentPrompt()
image_prompt = "What's in this image?"
content = [{"type": "text", "text": image_prompt}]
content.extend([{"type": "image_url", "image_url": {"url": image_utils.to_base64(image)}} for image in images])
msg_prompt.messages = [{"role": "user", "content": content}]
resp: ComputeTaskResult = await (ComputeKernel.get_instance()
.do_llm_completion(prompt=msg_prompt,
resp_mode="text",
mode_name="gpt-4-vision-preview",
max_token=4000,
inner_functions=None,
timeout=None))
if resp.result_code != ComputeTaskResultCode.OK:
raise Exception(f"image_to_text error: {resp.result_code} msg:{resp.error_str}")
return resp.result_str
async def video_to_text(video: str) -> str:
prompt = "These pictures are key frames extracted from the video. Please describe the content of the video based on these key frames."
frames = video_utils.extract_frames(video, (1024, 1024))
msg_prompt = AgentPrompt()
content = [{"type": "text", "text": prompt}]
content.extend([{"type": "image_url", "image_url": {"url": frame}} for frame in frames])
msg_prompt.messages = [{"role": "user", "content": content}]
resp: ComputeTaskResult = await (ComputeKernel.get_instance()
.do_llm_completion(prompt=msg_prompt,
resp_mode="text",
mode_name="gpt-4-vision-preview",
max_token=4000,
inner_functions=None,
timeout=None))
if resp.result_code != ComputeTaskResultCode.OK:
raise Exception(f"video_to_text error: {resp.result_code} msg:{resp.error_str}")
return resp.result_str
async def summary_document(text: str, separators: List[str]=["\n\n", "\n"]) -> str:
chunks = split_text(text, separators=separators, chunk_size=4000, chunk_overlap=200, length_function=len)
prompt = AgentPrompt()
prompt.system_message = {"role":"system","content":"Your job is to generate a summary based on the input."}
if len(chunks) == 1:
prompt.append(AgentPrompt(chunks[0]))
resp = await (ComputeKernel.get_instance()
.do_llm_completion(prompt=prompt,
resp_mode="text",
mode_name="gpt-4-1106-preview",
max_token=4000,
inner_functions=None,
timeout=None))
if resp.result_code != ComputeTaskResultCode.OK:
raise Exception(f"summary_document error: {resp.result_code} msg:{resp.error_str}")
return resp.result_str
segments = []
for i, chunk in enumerate(chunks):
seg_prompt = copy.deepcopy(prompt)
seg_prompt.append(AgentPrompt(chunk))
resp = await (ComputeKernel.get_instance()
.do_llm_completion(prompt=seg_prompt,
resp_mode="text",
mode_name="gpt-4-1106-preview",
max_token=4000,
inner_functions=None,
timeout=None))
if resp.result_code != ComputeTaskResultCode.OK:
raise Exception(f"summary_document error: {resp.result_code} msg:{resp.error_str}")
segments.append(resp.result_str)
segments_str = "\n".join(segments)
prompt.append(AgentPrompt(f"Please combine the summaries of the following paragraphs into one complete summary:\n{segments_str}"))
resp = await (ComputeKernel.get_instance()
.do_llm_completion(prompt=prompt,
resp_mode="text",
mode_name="gpt-4-1106-preview",
max_token=4000,
inner_functions=None,
timeout=None))
if resp.result_code != ComputeTaskResultCode.OK:
raise Exception(f"summary_document error: {resp.result_code} msg:{resp.error_str}")
return resp.result_str
def pdf_to_rich_text_object(pdf: str, store: KnowledgeStore) -> RichTextObject:
base_name = os.path.basename(pdf)
cache_path = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge", "doc_cache", base_name)
if not os.path.exists(cache_path):
os.makedirs(cache_path)
reader = PdfReader(pdf)
rich_text = RichTextObject()
page_texts = []
image_count = 0
for page in reader.pages:
text = page.extract_text()
page_texts.append(text)
for image in page.images:
image_path = os.path.join(cache_path, f"{image_count}_{image.name}")
with open(image_path, "wb") as f:
f.write(image.data)
image_object = ImageObjectBuilder({}, {}, image_path).build(store)
rich_text.add_image(image_object)
document = DocumentObjectBuilder({}, {}, "".join(page_texts)).build(store)
rich_text.add_document(document)
return rich_text
def doc_to_rich_text_object(doc: str, store: KnowledgeStore) -> RichTextObject:
base_name = os.path.basename(doc)
cache_path = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge", "doc_cache", base_name)
if not os.path.exists(cache_path):
os.makedirs(cache_path)
text = docx2txt.process(doc, cache_path)
rich_text = RichTextObject()
for image in os.listdir(cache_path):
image_path = os.path.join(cache_path, image)
image_object = ImageObjectBuilder({}, {}, image_path).build(store)
rich_text.add_image(image_object)
document = DocumentObjectBuilder({}, {}, text).build(store)
rich_text.add_document(document)
return rich_text
-101
View File
@@ -1,101 +0,0 @@
# define a knowledge base class
import json
import string
from aios import *
class EmbeddingParser:
def __init__(self, env: KnowledgePipelineEnvironment, config: dict):
self._default_text_model = "all-MiniLM-L6-v2"
self._default_image_model = "clip-ViT-B-32"
path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
if not os.path.exists(path):
os.makedirs(path)
config["path"] = path
self.env = env
self.config = config
def get_path(self) -> str:
return self.config["path"]
def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
return ChromaVectorStore(self.get_path(), model_name)
async def __embedding_document(self, document: DocumentObject):
for chunk_id in document.get_chunk_list():
chunk = self.env.get_knowledge_store().get_chunk_reader().get_chunk(chunk_id)
if chunk is None:
raise ValueError(f"text chunk not found: {chunk_id}")
text = chunk.read().decode("utf-8")
vector = await ComputeKernel.get_instance().do_text_embedding(text, self._default_text_model)
if vector:
await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id)
async def __embedding_image(self, image: ImageObject):
# desc = {}
# if not not image.get_meta():
# desc["meta"] = image.get_meta()
# if not not image.get_exif():
# desc["exif"] = image.get_exif()
# if not not image.get_tags():
# desc["tags"] = image.get_tags()
# vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model)
vector = await ComputeKernel.get_instance().do_image_embedding(image.calculate_id(), self._default_image_model)
if vector:
await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id())
async def __embedding_video(self, vedio: VideoObject):
desc = {}
if not not vedio.get_meta():
desc["meta"] = vedio.get_meta()
if not not vedio.get_info():
desc["info"] = vedio.get_info()
if not not vedio.get_tags():
desc["tags"] = vedio.get_tags()
vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(desc), self._default_text_model)
await self.__get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id())
async def __embedding_rich_text(self, rich_text: RichTextObject):
for document_id in rich_text.get_documents().values():
document = DocumentObject.decode(self.env.get_knowledge_store().get_object_store().get_object(document_id))
await self.__embedding_document(document)
for image_id in rich_text.get_images().values():
image = ImageObject.decode(self.env.get_knowledge_store().get_object_store().get_object(image_id))
await self.__embedding_image(image)
for video_id in rich_text.get_videos().values():
video = VideoObject.decode(self.env.get_knowledge_store().get_object_store().get_object(video_id))
await self.__embedding_video(video)
for rich_text_id in rich_text.get_rich_texts().values():
rich_text = RichTextObject.decode(self.env.get_knowledge_store().get_object_store().get_object(rich_text_id))
await self.__embedding_rich_text(rich_text)
async def __embedding_email(self, email: EmailObject):
vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(email.get_desc()), self._default_text_model)
await self.__get_vector_store(self._default_text_model).insert(vector, email.calculate_id())
await self.__embedding_rich_text(email.get_rich_text())
async def __do_embedding(self, object: KnowledgeObject):
if object.get_object_type() == ObjectType.Document:
await self.__embedding_document(object)
if object.get_object_type() == ObjectType.Image:
await self.__embedding_image(object)
if object.get_object_type() == ObjectType.Video:
await self.__embedding_video(object)
if object.get_object_type() == ObjectType.RichText:
await self.__embedding_rich_text(object)
if object.get_object_type() == ObjectType.Email:
await self.__embedding_email(object)
else:
pass
async def parse(self, object: ObjectID) -> str:
obj = self.env.get_knowledge_store().load_object(object)
await self.__do_embedding(obj)
return str(object)
def init(env: KnowledgePipelineEnvironment, params: dict) -> EmbeddingParser:
return EmbeddingParser(env, params)
@@ -1,6 +0,0 @@
name = "Mia"
input.module = "input.py"
input.params.path = "${myai_dir}/data"
parser.module = "parser.py"
parser.params.path = "${myai_dir}/knowledge/indices/embedding"
-96
View File
@@ -1,96 +0,0 @@
import os
import logging
import json
from aios import *
class EmbeddingEnvironment(SimpleEnvironment):
def __init__(self, workspace: str) -> None:
super().__init__(workspace)
self.path = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge/indices/embedding")
self._default_text_model = "all-MiniLM-L6-v2"
self._default_image_model = "clip-ViT-B-32"
query_param = {
"tokens": "key words to query",
"types": "prefered knowledge types, one or more of [text, image]",
"index": "index of query result"
}
self.add_ai_function(SimpleAIFunction("query_knowledge",
"vector query content from local knowledge base",
self._query,
query_param))
def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
return ChromaVectorStore(self.path, model_name)
async def query_objects(self, tokens: str, types: list[str], topk: int) -> [ObjectID]:
texts = []
if "text" in types:
vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_text_model)
texts = await self.__get_vector_store(self._default_text_model).query(vector, topk)
images = []
if "image" in types:
vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_image_model)
images = await self.__get_vector_store(self._default_image_model).query(vector, topk)
return texts + images
def tokens_from_objects(self, object_ids: [ObjectID]) -> list[str]:
results = dict()
for object_id in object_ids:
parents = KnowledgeStore().get_relation_store().get_related_root_objects(object_id)
# last parent is the root object
root_object_id = parents[0] if parents else object_id
logging.info(f"object_id: {str(object_id)} root_object_id: {str(root_object_id)}")
if str(root_object_id) in results:
results[str(root_object_id)].append(object_id)
else:
results[str(root_object_id)] = [root_object_id, object_id]
content = ""
result_desc = []
for result in results.values():
# first element in result is the root object
root_object_id = result[0]
if root_object_id.get_object_type() == ObjectType.Email:
email = KnowledgeStore().load_object(root_object_id)
desc = email.get_desc()
desc["type"] = "email"
desc["contents"] = []
result_desc.append(desc)
upper_list = desc["contents"]
result = result[1:]
else:
upper_list = result_desc
for object_id in result:
if object_id.get_object_type() == ObjectType.Chunk:
upper_list.append({"type": "text", "content": KnowledgeStore().get_chunk_reader().get_chunk(object_id).read().decode("utf-8")})
if object_id.get_object_type() == ObjectType.Image:
# image = self.load_object(object_id)
desc = dict()
desc["id"] = str(object_id)
desc["type"] = "image"
upper_list.append(desc)
if object_id.get_object_type() == ObjectType.Video:
video = KnowledgeStore().load_object(object_id)
desc = video.get_desc()
desc["type"] = "video"
upper_list.append(desc)
else:
pass
content += json.dumps(result_desc)
content += ".\n"
return content
async def _query(self, tokens: str, types: list[str] = ["text"], index: str=0):
index = int(index)
object_ids = await self.query_objects(tokens, types, 4)
if len(object_ids) <= index:
return "*** I have no more information for your reference.\n"
else:
content = "*** I have provided the following known information for your reference with json format:\n"
return content + self.tokens_from_objects(object_ids[index:index+1])
def init(workspace: str) -> EmbeddingEnvironment:
return EmbeddingEnvironment(workspace)
@@ -1,3 +0,0 @@
pipelines = [
"JarvisPlus"
]
+3 -2
View File
@@ -6,7 +6,7 @@ from .proto.agent_task import *
from .agent.agent_base import * from .agent.agent_base import *
from .agent.chatsession import AIChatSession from .agent.chatsession import AIChatSession
from .agent.agent import AIAgent,AIAgentTemplete, BaseAIAgent from .agent.agent import AIAgent, BaseAIAgent
from .agent.role import AIRole,AIRoleGroup from .agent.role import AIRole,AIRoleGroup
from .agent.workflow import Workflow from .agent.workflow import Workflow
from .agent.agent_memory import AgentMemory from .agent.agent_memory import AgentMemory
@@ -29,6 +29,7 @@ from .ai_functions.image_2_text_function import Image2TextFunction
from .environment.workspace_env import WorkspaceEnvironment from .environment.workspace_env import WorkspaceEnvironment
from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem
from .storage.objfs import ObjFS
from .net import * from .net import *
from .knowledge import * from .knowledge import *
@@ -36,4 +37,4 @@ from .package_manager import *
from .utils import * from .utils import *
AIOS_Version = "0.5.2, build 2023-12-15" AIOS_Version = "0.5.2, build 2024-3-31"
+4 -46
View File
@@ -31,29 +31,6 @@ from ..proto.compute_task import LLMPrompt,LLMResult
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class AIAgentTemplete:
def __init__(self) -> None:
self.llm_model_name:str = "gpt-4-turbo-preview"
self.max_token_size:int = 0
self.template_id:str = None
self.introduce:str = None
self.author:str = None
self.prompt:LLMPrompt = None
def load_from_config(self,config:dict) -> bool:
if config.get("llm_model_name") is not None:
self.llm_model_name = config["llm_model_name"]
if config.get("max_token_size") is not None:
self.max_token_size = config["max_token_size"]
if config.get("template_id") is not None:
self.template_id = config["template_id"]
if config.get("prompt") is not None:
self.prompt = LLMPrompt()
if self.prompt.load_from_config(config["prompt"]) is False:
logger.error("load prompt from config failed!")
return False
class AIAgent(BaseAIAgent): class AIAgent(BaseAIAgent):
def __init__(self) -> None: def __init__(self) -> None:
@@ -81,19 +58,6 @@ class AIAgent(BaseAIAgent):
self.history_len = 10 self.history_len = 10
self.read_report_prompt = None self.read_report_prompt = None
todo_prompts = {}
todo_prompts[TodoListType.TO_WORK] = {
"do": None,
"check": None,
"review": None,
}
todo_prompts[TodoListType.TO_LEARN] = {
"do": None,
"check": None,
"review": None,
}
self.todo_prompts = todo_prompts
self.base_dir = None self.base_dir = None
#self.memory_db = None #self.memory_db = None
self.unread_msg = Queue() # msg from other agent self.unread_msg = Queue() # msg from other agent
@@ -178,9 +142,6 @@ class AIAgent(BaseAIAgent):
return self.template_id return self.template_id
def get_llm_model_name(self) -> str: def get_llm_model_name(self) -> str:
if self.llm_model_name is None:
return AIStorage.get_instance().get_user_config().get_value("llm_model_name")
return self.llm_model_name return self.llm_model_name
def get_max_token_size(self) -> int: def get_max_token_size(self) -> int:
@@ -248,14 +209,11 @@ class AIAgent(BaseAIAgent):
logger.info(f"agent {self.agent_id} self thinking start!") logger.info(f"agent {self.agent_id} self thinking start!")
context_info = await self._get_context_info() context_info = await self._get_context_info()
known_session_list = AIChatSession.list_session(self.agent_id,self.memory.memory_db) #chat_history = AIChatSession.list_session(self.agent_id,self.memory.memory_db)
known_experience_list = await self.memory.list_experience() #known_experience_list = await self.memory.list_experience()
record_list = await self.memory.load_records(await self.memory.get_last_think_time()) #record_list = await self.memory.load_records(await self.memory.get_last_think_time())
input_parms = { input_parms = {
"record_list":record_list,
"known_session_list":known_session_list,
"known_experience_list":known_experience_list,
"context_info":context_info "context_info":context_info
} }
@@ -266,7 +224,7 @@ class AIAgent(BaseAIAgent):
logger.info(f"llm process self thinking ignore!") logger.info(f"llm process self thinking ignore!")
else: else:
logger.info(f"llm process self thinking ok!,think is:{llm_result.resp}") logger.info(f"llm process self thinking ok!,think is:{llm_result.resp}")
self.memory.set_last_think_time(time.time()) await self.memory.set_last_think_time(time.time())
self.agent_energy -= 2 self.agent_energy -= 2
return return
+267 -246
View File
@@ -9,10 +9,11 @@ import sqlite3
import aiofiles import aiofiles
from ..storage.storage import AIStorage from ..storage.storage import AIStorage
from ..knowledge.knowledge_base import BaseKnowledgeGraph,ObjFSKnowledgeGrpah
from ..frame.compute_kernel import ComputeKernel from ..frame.compute_kernel import ComputeKernel
from ..frame.contact_manager import ContactManager from ..frame.contact_manager import ContactManager
from ..frame.contact import Contact from ..frame.contact import Contact
from ..proto.ai_function import ParameterDefine, SimpleAIAction, SimpleAIFunction from ..proto.ai_function import ParameterDefine, SimpleAIFunction
from ..proto.agent_msg import AgentMsg, AgentMsgType from ..proto.agent_msg import AgentMsg, AgentMsgType
from ..proto.agent_task import AgentWorkLog from ..proto.agent_task import AgentWorkLog
@@ -35,20 +36,36 @@ logger = logging.getLogger(__name__)
class AgentMemory: class AgentMemory:
def __init__(self,agent_id:str,base_dir:str) -> None: def __init__(self,agent_id:str,base_dir:str,enable_knowledge_graph = False) -> None:
self.agent_memory_base_dir = base_dir self.agent_memory_base_dir = base_dir
self.agent_id:str= agent_id self.agent_id:str= agent_id
AIStorage.get_instance().ensure_directory_exists(self.agent_memory_base_dir) AIStorage.get_instance().ensure_directory_exists(self.agent_memory_base_dir)
AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/experience") #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/experience")
AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/contacts") #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/contacts")
AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/relations") #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/relations")
AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/summary") #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/summary")
self.memory_db:str = f"{self.agent_memory_base_dir}/memory.db" self.memory_db:str = f"{self.agent_memory_base_dir}/memory.db"
self.model_name:str = "gp4-1106-preview"
self.threshold_hours = 72 self.threshold_hours = 72
self.last_think_time : float = 0.0 self.last_think_time : float = 0.0
self.enable_knowledge_graph : bool = enable_knowledge_graph
if self.enable_knowledge_graph:
kb_desc = """
The Knowledgegraph is used to store important information obtained by Agent in the conversation.Use the following ways to store information:
- /contacts/$name:Related information of the contact
- /relations/$obj1/$obj2:The relationship between obj2 and obj1
- /summary/$topic:Based on topic summary
"""
self.knowledge_graph = ObjFSKnowledgeGrpah(f"{self.agent_id}.memory",self.memory_db,kb_desc)
BaseKnowledgeGraph.add_kb(self.knowledge_graph)
self.simple_memory_sentences = None
else:
self.knowledge_graph = None
self.simple_memory_sentences : List[str] = []
self.load_memory_meta() self.load_memory_meta()
@@ -84,7 +101,7 @@ class AgentMemory:
return chatsession return chatsession
# return last record time # return last record time
async def load_records(self,starttime,tokenlimit=8000)->float: async def load_records(self,starttime,tokenlimit=8000,model_name=None)->float:
# 专用思路:做聊天记录/工作经验的整理 # 专用思路:做聊天记录/工作经验的整理
# 通用思路:没有具体的目的,让Agent根据提示词自己工作(可能效果很差也可能很好) # 通用思路:没有具体的目的,让Agent根据提示词自己工作(可能效果很差也可能很好)
# 先实现通用思路 # 先实现通用思路
@@ -92,49 +109,31 @@ class AgentMemory:
work_records = self.load_worklogs(self.agent_id,token_limit=tokenlimit) work_records = self.load_worklogs(self.agent_id,token_limit=tokenlimit)
pass pass
async def load_chatlogs(self,msg:AgentMsg,n:int=6,m:int=64,token_limit=800)->str: async def load_chatlogs(self,msg:AgentMsg,token_limit=800,model_name=""):
chatsession = self.get_session_from_msg(msg) chatsession = self.get_session_from_msg(msg)
# Must load n (n> = 2), and hope to load the M # Must load n (n> = 2), and hope to load the M
# The information in the # M is gradually added, knowing that it is less than 72 hours from the current time, and consumes enough tokens # The information in the # M is gradually added, knowing that it is less than 72 hours from the current time, and consumes enough tokens
messages_n = chatsession.read_history(n) # read messages_n = chatsession.read_history() # read
if len(messages_n) >= n:
messages_m = chatsession.read_history(m,n)
else:
messages_m = []
histroy_str = "" histroy_str = ""
read_count = 0 read_count = 0
is_all = True
for msg in messages_n: for msg in messages_n:
dt = datetime.fromtimestamp(float(msg.create_time)) dt = datetime.fromtimestamp(float(msg.create_time))
formatted_time = dt.strftime('%y-%m-%d %H:%M:%S') formatted_time = dt.strftime('%y-%m-%d %H:%M:%S')
record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n" record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n"
token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str,self.model_name) token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str)
if token_limit <= 32: if token_limit <= 32:
is_all = False
break break
read_count += 1 read_count += 1
histroy_str = record_str + histroy_str histroy_str = record_str + histroy_str
if len(messages_n) > 2: return histroy_str,is_all
if read_count < 3:
logging.warning(f"read history {read_count} < 3, will not load more") async def get_chat_summary(self,msg:AgentMsg) -> str:
chatsession : AIChatSession = self.get_session_from_msg(msg)
now = datetime.now() return chatsession.summary
for msg in messages_m:
dt = datetime.fromtimestamp(float(msg.create_time))
time_diff = now - dt
if time_diff > timedelta(hours=self.threshold_hours):
break
formatted_time = dt.strftime('%y-%m-%d %H:%M:%S')
record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n"
token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str,self.model_name)
if token_limit <= 32:
break
read_count += 1
histroy_str = record_str + histroy_str
return histroy_str
# async def action_chatlog_append(self,params:Dict) -> str: # async def action_chatlog_append(self,params:Dict) -> str:
# #
@@ -174,7 +173,7 @@ class AgentMemory:
rows = c.fetchall() rows = c.fetchall()
return [self.from_db_row(row) for row in rows] return [self.worklog_from_db_row(row) for row in rows]
def _create_table(self,conn): def _create_table(self,conn):
c = conn.cursor() c = conn.cursor()
@@ -194,7 +193,7 @@ class AgentMemory:
#conn.close() #conn.close()
@classmethod @classmethod
def from_db_row(self,row): def worklog_from_db_row(self,row):
log = AgentWorkLog() log = AgentWorkLog()
# 这里高度依赖表结构的顺序 # 这里高度依赖表结构的顺序
log.logid, log.owner_id, log.work_type, log.timestamp, log.content, log.result, meta_str, log.operator = row log.logid, log.owner_id, log.work_type, log.timestamp, log.content, log.result, meta_str, log.operator = row
@@ -220,6 +219,7 @@ class AgentMemory:
def load_meta(self,Dict): def load_meta(self,Dict):
self.last_think_time = Dict.get("last_think_time",0.0) self.last_think_time = Dict.get("last_think_time",0.0)
self.simple_memory_sentences = Dict.get("simple_memory_sentences",[])
def load_memory_meta(self): def load_memory_meta(self):
meta_file_path = f"{self.agent_memory_base_dir}/meta.json" meta_file_path = f"{self.agent_memory_base_dir}/meta.json"
@@ -248,7 +248,12 @@ class AgentMemory:
self.last_think_time = last_time self.last_think_time = last_time
self.save_memory_meta() self.save_memory_meta()
async def get_contact_summary(self,contact_id:str) -> str: async def get_contact_summary(self,contact_id:str) -> str:
# There is two part of contact summary
# Part 1. user defined summary (set by owner or by contac) , global , imutable
# Part 2. auto generated summary, local in agent memory , mutable
if contact_id is None: if contact_id is None:
return "Contact id is None" return "Contact id is None"
@@ -259,245 +264,261 @@ class AgentMemory:
result["relation"] = contact_info.relationship result["relation"] = contact_info.relationship
result["notes"] = contact_info.notes result["notes"] = contact_info.notes
summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" # summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='r') as file: # async with aiofiles.open(summary_path, mode='r') as file:
result["summary"] = await file.read() # result["summary"] = await file.read()
except Exception as e: # except Exception as e:
logger.error(f"read contact summary failed: {e}") # logger.error(f"read contact summary failed: {e}")
return json.dumps(result,ensure_ascii=False) return json.dumps(result,ensure_ascii=False)
async def update_contact_summary(self,contact_id:str,summary:str): # async def update_contact_summary(self,contact_id:str,summary:str):
summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" # summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='w') as file: # async with aiofiles.open(summary_path, mode='w') as file:
await file.write(summary) # await file.write(summary)
return "OK" # return "OK"
except Exception as e: # except Exception as e:
logger.error(f"write contact summary failed: {e}") # logger.error(f"write contact summary failed: {e}")
return "write contact summary failed: {e}" # return "write contact summary failed: {e}"
async def get_summary(self,object_name:str) -> str: # async def get_summary(self,object_name:str) -> str:
summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" # summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='r') as file: # async with aiofiles.open(summary_path, mode='r') as file:
return await file.read() # return await file.read()
except Exception as e: # except Exception as e:
logger.error(f"read summary failed: {e}") # logger.error(f"read summary failed: {e}")
return f"read summary failed: {e}" # return f"read summary failed: {e}"
async def update_summary(self,object_name:str,summary:str) -> str: # async def update_summary(self,object_name:str,summary:str) -> str:
summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" # summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='w') as file: # async with aiofiles.open(summary_path, mode='w') as file:
await file.write(summary) # await file.write(summary)
return "OK" # return "OK"
except Exception as e: # except Exception as e:
logger.error(f"write summary failed: {e}") # logger.error(f"write summary failed: {e}")
return f"write summary failed: {e}" # return f"write summary failed: {e}"
async def list_summary_object_names(self) -> List[str]: # async def list_summary_object_names(self) -> List[str]:
# list dir # # list dir
try: # try:
contents = os.listdir(self.agent_memory_base_dir) # contents = os.listdir(self.agent_memory_base_dir)
return [x for x in contents if x.endswith(".summary")] # return [x for x in contents if x.endswith(".summary")]
except Exception as e: # except Exception as e:
logger.error(f"list summary object names failed: {e}") # logger.error(f"list summary object names failed: {e}")
return [] # return []
# means object1 feel object2 is ... # means object1 feel object2 is ...
async def get_relation_summary(self,object_name1:str,object_name2:str) -> str: # async def get_relation_summary(self,object_name1:str,object_name2:str) -> str:
summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" # summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='r') as file: # async with aiofiles.open(summary_path, mode='r') as file:
await file.read() # await file.read()
except FileNotFoundError: # except FileNotFoundError:
return "no summary" # return "no summary"
except Exception as e: # except Exception as e:
logger.error(f"read relation summary failed: {e}") # logger.error(f"read relation summary failed: {e}")
return f"read relation summary failed: {e}" # return f"read relation summary failed: {e}"
async def update_relation_summary(self,object_name1:str,object_name2:str,summary:Dict): # async def update_relation_summary(self,object_name1:str,object_name2:str,summary:Dict):
summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" # summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary"
try: # try:
async with aiofiles.open(summary_path, mode='w') as file: # async with aiofiles.open(summary_path, mode='w') as file:
await file.write(json.dumps(summary)) # await file.write(json.dumps(summary))
return "OK" # return "OK"
except Exception as e: # except Exception as e:
logger.error(f"write relation summary failed: {e}") # logger.error(f"write relation summary failed: {e}")
return "write relation summary failed: {e}" # return "write relation summary failed: {e}"
async def get_experience(self,topic_name:str) -> str: # async def get_experience(self,topic_name:str) -> str:
experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" # experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience"
try: # try:
async with aiofiles.open(experience_path, mode='r') as file: # async with aiofiles.open(experience_path, mode='r') as file:
await file.read() # await file.read()
except FileNotFoundError: # except FileNotFoundError:
return "no experience" # return "no experience"
except Exception as e: # except Exception as e:
logger.error(f"read experience failed: {e}") # logger.error(f"read experience failed: {e}")
return f"read experience failed: {e}" # return f"read experience failed: {e}"
async def set_experience(self,topic_name:str,summary:str) -> str: # async def set_experience(self,topic_name:str,summary:str) -> str:
experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" # experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience"
try: # try:
async with aiofiles.open(experience_path, mode='w') as file: # async with aiofiles.open(experience_path, mode='w') as file:
await file.write(summary) # await file.write(summary)
return "OK" # return "OK"
except Exception as e: # except Exception as e:
logger.error(f"write experience failed: {e}") # logger.error(f"write experience failed: {e}")
return "write experience failed: {e}" # return "write experience failed: {e}"
async def list_experience(self) -> List[str]: # async def list_experience(self) -> List[str]:
dir_path = f"{self.agent_memory_base_dir}/experience" # dir_path = f"{self.agent_memory_base_dir}/experience"
try: # try:
contents = os.listdir(dir_path) # contents = os.listdir(dir_path)
return [x for x in contents if x.endswith(".experience")] # return [x for x in contents if x.endswith(".experience")]
except Exception as e: # except Exception as e:
logger.error(f"list experience failed: {e}") # logger.error(f"list experience failed: {e}")
return [] # return []
@staticmethod @staticmethod
def register_ai_functions(): def register_ai_functions():
async def get_contact_summary(parameters): async def update_chat_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") agent_memory:AgentMemory = parameters.get("_memory")
contact_name = parameters.get("contact_name") chatsession = AIChatSession.get_session_by_id(parameters.get("session_id"),agent_memory.memory_db)
return await agent_memory.get_contact_summary(contact_name)
parameters = ParameterDefine.create_parameters({
"contact_name": {"type": "string", "description": "contact name"}
})
get_contact_summary_func = SimpleAIFunction("agent.memory.get_contact_summary",
"get contact summary",
get_contact_summary,
parameters)
GlobaToolsLibrary.register_tool_function(get_contact_summary_func)
async def update_contact_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory")
contact_name = parameters.get("contact_name")
summary = parameters.get("summary") summary = parameters.get("summary")
return await agent_memory.update_contact_summary(contact_name,summary) chatsession.update_summary(summary)
return "OK"
parameters = ParameterDefine.create_parameters({ parameters = ParameterDefine.create_parameters({
"contact_name": {"type": "string", "description": "contact name"}, "session_id": {"type": "string", "description": "session id"},
"summary": {"type": "string", "description": "new summary"} "summary": {"type": "string", "description": "new summary"}
}) })
update_contact_summary_func = SimpleAIFunction("agent.memory.update_contact_summary", update_chat_summary_func = SimpleAIFunction("agent.memory.update_chat_summary",
"update contact summary", "update chat summary",
update_contact_summary, update_chat_summary,
parameters) parameters)
GlobaToolsLibrary.register_tool_function(update_contact_summary_func) GlobaToolsLibrary.get_instance().register_tool_function(update_chat_summary_func)
async def get_summary(parameters): # async def get_contact_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
object_name = parameters.get("object_name") # contact_name = parameters.get("contact_name")
return await agent_memory.get_summary(object_name) # return await agent_memory.get_contact_summary(contact_name)
parameters = ParameterDefine.create_parameters({ # parameters = ParameterDefine.create_parameters({
"object_name": {"type": "string", "description": "object name"} # "contact_name": {"type": "string", "description": "contact name"}
}) # })
get_summary_func = SimpleAIFunction("agent.memory.get_summary", # get_contact_summary_func = SimpleAIFunction("agent.memory.get_contact_summary",
"get summary of sth", # "get contact summary",
get_summary, # get_contact_summary,
parameters) # parameters)
GlobaToolsLibrary.register_tool_function(get_summary_func) # GlobaToolsLibrary.register_tool_function(get_contact_summary_func)
async def update_summary(parameters): # async def update_contact_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
object_name = parameters.get("object_name") # contact_name = parameters.get("contact_name")
summary = parameters.get("summary") # summary = parameters.get("summary")
return await agent_memory.update_summary(object_name,summary) # return await agent_memory.update_contact_summary(contact_name,summary)
parameters = ParameterDefine.create_parameters({ # parameters = ParameterDefine.create_parameters({
"object_name": {"type": "string", "description": "object name"}, # "contact_name": {"type": "string", "description": "contact name"},
"summary": {"type": "string", "description": "new summary"} # "summary": {"type": "string", "description": "new summary"}
}) # })
update_summary_func = SimpleAIFunction("agent.memory.update_summary", # update_contact_summary_func = SimpleAIFunction("agent.memory.update_contact_summary",
"update summary of sth", # "update contact summary",
update_summary, # update_contact_summary,
parameters) # parameters)
GlobaToolsLibrary.register_tool_function(update_summary_func) # GlobaToolsLibrary.register_tool_function(update_contact_summary_func)
async def list_summary_object_names(parameters): # async def get_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
return await agent_memory.list_summary_object_names() # object_name = parameters.get("object_name")
parameters = ParameterDefine.create_parameters({}) # return await agent_memory.get_summary(object_name)
list_summary_object_names_func = SimpleAIFunction("agent.memory.list_summary", # parameters = ParameterDefine.create_parameters({
"list summary object names", # "object_name": {"type": "string", "description": "object name"}
list_summary_object_names, # })
parameters) # get_summary_func = SimpleAIFunction("agent.memory.get_summary",
GlobaToolsLibrary.register_tool_function(list_summary_object_names_func) # "get summary of sth",
# get_summary,
# parameters)
# GlobaToolsLibrary.register_tool_function(get_summary_func)
async def get_relation_summary(parameters): # async def update_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
object_name1 = parameters.get("object1_name") # object_name = parameters.get("object_name")
object_name2 = parameters.get("object2_name") # summary = parameters.get("summary")
return await agent_memory.get_relation_summary(object_name1,object_name2) # return await agent_memory.update_summary(object_name,summary)
parameters = ParameterDefine.create_parameters({ # parameters = ParameterDefine.create_parameters({
"object1_name": {"type": "string", "description": "object name1"}, # "object_name": {"type": "string", "description": "object name"},
"object2_name": {"type": "string", "description": "object name2"} # "summary": {"type": "string", "description": "new summary"}
}) # })
get_relation_summary_func = SimpleAIFunction("agent.memory.get_relation_summary", # update_summary_func = SimpleAIFunction("agent.memory.update_summary",
"object1 feel object2 is ...", # "update summary of sth",
get_relation_summary, # update_summary,
parameters) # parameters)
GlobaToolsLibrary.register_tool_function(get_relation_summary_func) # GlobaToolsLibrary.register_tool_function(update_summary_func)
async def update_relation_summary(parameters): # async def list_summary_object_names(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
object_name1 = parameters.get("object1_name") # return await agent_memory.list_summary_object_names()
object_name2 = parameters.get("object2_name") # parameters = ParameterDefine.create_parameters({})
summary = parameters.get("summary") # list_summary_object_names_func = SimpleAIFunction("agent.memory.list_summary",
return await agent_memory.update_relation_summary(object_name1,object_name2,summary) # "list summary object names",
parameters = ParameterDefine.create_parameters({ # list_summary_object_names,
"object1_name": {"type": "string", "description": "object name1"}, # parameters)
"object2_name": {"type": "string", "description": "object name2"}, # GlobaToolsLibrary.register_tool_function(list_summary_object_names_func)
"summary": {"type": "string", "description": "new summary"}
})
update_relation_summary_func = SimpleAIFunction("agent.memory.update_relation_summary",
"object1 feel object2 is ...",
update_relation_summary,
parameters)
GlobaToolsLibrary.register_tool_function(update_relation_summary_func)
async def get_experience(parameters): # async def get_relation_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
topic_name = parameters.get("topic_name") # object_name1 = parameters.get("object1_name")
return await agent_memory.get_experience(topic_name) # object_name2 = parameters.get("object2_name")
parameters = ParameterDefine.create_parameters({ # return await agent_memory.get_relation_summary(object_name1,object_name2)
"topic_name": {"type": "string", "description": "topic name"} # parameters = ParameterDefine.create_parameters({
}) # "object1_name": {"type": "string", "description": "object name1"},
get_experience_func = SimpleAIFunction("agent.memory.get_experience", # "object2_name": {"type": "string", "description": "object name2"}
"get experience", # })
get_experience, # get_relation_summary_func = SimpleAIFunction("agent.memory.get_relation_summary",
parameters) # "object1 feel object2 is ...",
GlobaToolsLibrary.register_tool_function(get_experience_func) # get_relation_summary,
# parameters)
# GlobaToolsLibrary.register_tool_function(get_relation_summary_func)
async def set_experience(parameters): # async def update_relation_summary(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
topic_name = parameters.get("topic_name") # object_name1 = parameters.get("object1_name")
summary = parameters.get("summary") # object_name2 = parameters.get("object2_name")
return await agent_memory.set_experience(topic_name,summary) # summary = parameters.get("summary")
parameters = ParameterDefine.create_parameters({ # return await agent_memory.update_relation_summary(object_name1,object_name2,summary)
"topic_name": {"type": "string", "description": "topic name"}, # parameters = ParameterDefine.create_parameters({
"summary": {"type": "string", "description": "new summary"} # "object1_name": {"type": "string", "description": "object name1"},
}) # "object2_name": {"type": "string", "description": "object name2"},
set_experience_func = SimpleAIFunction("agent.memory.set_experience", # "summary": {"type": "string", "description": "new summary"}
"set experience", # })
set_experience, # update_relation_summary_func = SimpleAIFunction("agent.memory.update_relation_summary",
parameters) # "object1 feel object2 is ...",
GlobaToolsLibrary.register_tool_function(set_experience_func) # update_relation_summary,
# parameters)
# GlobaToolsLibrary.register_tool_function(update_relation_summary_func)
async def list_experience(parameters): # async def get_experience(parameters):
agent_memory:AgentMemory = parameters.get("_agent_memory") # agent_memory:AgentMemory = parameters.get("_memory")
return await agent_memory.list_experience() # topic_name = parameters.get("topic_name")
parameters = ParameterDefine.create_parameters({}) # return await agent_memory.get_experience(topic_name)
list_experience_func = SimpleAIFunction("agent.memory.list_experience", # parameters = ParameterDefine.create_parameters({
"list exist experience topics", # "topic_name": {"type": "string", "description": "topic name"}
list_experience, # })
parameters) # get_experience_func = SimpleAIFunction("agent.memory.get_experience",
GlobaToolsLibrary.register_tool_function(list_experience_func) # "get experience",
# get_experience,
# parameters)
# GlobaToolsLibrary.register_tool_function(get_experience_func)
# async def set_experience(parameters):
# agent_memory:AgentMemory = parameters.get("_memory")
# topic_name = parameters.get("topic_name")
# summary = parameters.get("summary")
# return await agent_memory.set_experience(topic_name,summary)
# parameters = ParameterDefine.create_parameters({
# "topic_name": {"type": "string", "description": "topic name"},
# "summary": {"type": "string", "description": "new summary"}
# })
# set_experience_func = SimpleAIFunction("agent.memory.set_experience",
# "set experience",
# set_experience,
# parameters)
# GlobaToolsLibrary.register_tool_function(set_experience_func)
# async def list_experience(parameters):
# agent_memory:AgentMemory = parameters.get("_memory")
# return await agent_memory.list_experience()
# parameters = ParameterDefine.create_parameters({})
# list_experience_func = SimpleAIFunction("agent.memory.list_experience",
# "list exist experience topics",
# list_experience,
# parameters)
# GlobaToolsLibrary.register_tool_function(list_experience_func)
+21 -5
View File
@@ -198,6 +198,9 @@ class ChatSessionDB:
try: try:
conn = self._get_conn() conn = self._get_conn()
cursor = conn.cursor() cursor = conn.cursor()
if limit == 0:
limit = 1024
cursor.execute(""" cursor.execute("""
SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages
WHERE SessionID = ? WHERE SessionID = ?
@@ -234,6 +237,8 @@ class ChatSessionDB:
try: try:
conn = self._get_conn() conn = self._get_conn()
cursor = conn.cursor() cursor = conn.cursor()
if limit == 0:
limit = 1024
cursor.execute(""" cursor.execute("""
SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages
WHERE SessionID = ? WHERE SessionID = ?
@@ -296,6 +301,7 @@ class ChatSessionDB:
# chat session might be large, so can read / write at stream mode. # chat session might be large, so can read / write at stream mode.
class AIChatSession: class AIChatSession:
_dbs = {} _dbs = {}
_sessions = {}
#@classmethod #@classmethod
#async def get_session_by_id(cls,session_id:str,db_path:str): #async def get_session_by_id(cls,session_id:str,db_path:str):
# db = cls._dbs.get(db_path) # db = cls._dbs.get(db_path)
@@ -345,18 +351,24 @@ class AIChatSession:
cls._dbs[db_path] = db cls._dbs[db_path] = db
result = None result = None
for session in cls._sessions.values():
if session.owner_id == owner_id and session.topic == session_topic:
return session
session = db.get_chatsession_by_owner_topic(owner_id,session_topic) session = db.get_chatsession_by_owner_topic(owner_id,session_topic)
if session is None: if session is None:
if auto_create: if auto_create:
session_id = "CS#" + uuid.uuid4().hex session_id = "CS#" + uuid.uuid4().hex
db.insert_chatsession(session_id,owner_id,session_topic,datetime.datetime.now()) db.insert_chatsession(session_id,owner_id,session_topic,datetime.datetime.now())
result = AIChatSession(owner_id,session_id,db) result = AIChatSession(owner_id,session_id,db)
cls._sessions[session_id] = result
else: else:
result = AIChatSession(owner_id,session[0],db) result = AIChatSession(owner_id,session[0],db)
result.topic = session_topic result.topic = session_topic
result.summarize_pos = session[4] result.summarize_pos = session[4]
result.summary = session[5] result.summary = session[5]
result.openai_thread_id = session[6] result.openai_thread_id = session[6]
cls._sessions[result.session_id] = result
return result return result
@@ -368,6 +380,10 @@ class AIChatSession:
cls._dbs[db_path] = db cls._dbs[db_path] = db
result = None result = None
session = cls._sessions.get(session_id)
if session:
return session
session = db.get_chatsession_by_id(session_id) session = db.get_chatsession_by_id(session_id)
if session is None: if session is None:
return None return None
@@ -377,6 +393,7 @@ class AIChatSession:
result.summarize_pos = session[4] result.summarize_pos = session[4]
result.summary = session[5] result.summary = session[5]
result.openai_thread_id = session[6] result.openai_thread_id = session[6]
cls._sessions[session_id] = result
return result return result
@@ -402,13 +419,13 @@ class AIChatSession:
self.topic : str = None self.topic : str = None
self.start_time : str = None self.start_time : str = None
self.summarize_pos : int = 0 self.summarize_pos : int = 0
self.summary = None self.summary : str = None
self.openai_thread_id = None self.openai_thread_id = None
def get_owner_id(self) -> str: def get_owner_id(self) -> str:
return self.owner_id return self.owner_id
def read_history(self, number:int=10,offset=0,order="revers") -> [AgentMsg]: def read_history(self, number:int=0,offset=0,order="revers") -> [AgentMsg]:
if order == "revers": if order == "revers":
msgs = self.db.get_messages(self.session_id, number, offset) msgs = self.db.get_messages(self.session_id, number, offset)
else: else:
@@ -444,9 +461,8 @@ class AIChatSession:
self.db.insert_message(msg,tags) self.db.insert_message(msg,tags)
def update_think_progress(self,progress:int,new_summary:str) -> None: def update_summary(self,new_summary:str) -> None:
self.db.update_session_summary(self.session_id,progress,new_summary) self.db.update_session_summary(self.session_id,self.summarize_pos,new_summary)
self.summarize_pos = progress
self.summary = new_summary self.summary = new_summary
def update_openai_thread_id(self,thread_id:str) -> None: def update_openai_thread_id(self,thread_id:str) -> None:
+124 -129
View File
@@ -14,6 +14,7 @@ from .workspace import AgentWorkspace
from .llm_context import LLMProcessContext,GlobaToolsLibrary, SimpleLLMContext from .llm_context import LLMProcessContext,GlobaToolsLibrary, SimpleLLMContext
from ..frame.compute_kernel import ComputeKernel from ..frame.compute_kernel import ComputeKernel
from ..knowledge.knowledge_base import BaseKnowledgeGraph
from abc import ABC,abstractmethod from abc import ABC,abstractmethod
import copy import copy
@@ -39,8 +40,9 @@ class BaseLLMProcess(ABC):
#None means system default, #None means system default,
# TODO: support abcstract model name like: local-hight,local-low,local-medium,remote-hight,remote-low,remote-medium # TODO: support abcstract model name like: local-hight,local-low,local-medium,remote-hight,remote-low,remote-medium
self.model_name = None self.model_name = None
self.max_token = 1000 # result_token self.max_token = 2000 # result_token
self.max_prompt_token = 1000 # not include input prompt self.max_prompt_token = 2000 # not include input prompt
self.chat_summary_token_len = 500
self.timeout = 1800 # 30 min self.timeout = 1800 # 30 min
self.llm_context:LLMProcessContext = None self.llm_context:LLMProcessContext = None
@@ -64,6 +66,9 @@ class BaseLLMProcess(ABC):
async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool:
pass pass
def get_remain_prompt_length(self,prompt:LLMPrompt,will_append_str:str) -> int:
return self.max_prompt_token - ComputeKernel.llm_num_tokens(prompt,self.model_name)
@abstractmethod @abstractmethod
async def load_from_config(self,config:dict) -> bool: async def load_from_config(self,config:dict) -> bool:
#self.behavior = config.get("behavior") #self.behavior = config.get("behavior")
@@ -166,6 +171,10 @@ class BaseLLMProcess(ABC):
# Action define in prompt, will be execute after llm compute # Action define in prompt, will be execute after llm compute
prompt = await self.prepare_prompt(input) prompt = await self.prepare_prompt(input)
if prompt is None:
logger.warn(f"prepare_prompt return None, break llm_process")
return LLMResult.from_error_str("prepare_prompt return None")
max_result_token = self.max_token - ComputeKernel.llm_num_tokens(prompt,self.get_llm_model_name()) max_result_token = self.max_token - ComputeKernel.llm_num_tokens(prompt,self.get_llm_model_name())
#if max_result_token < MIN_PREDICT_TOKEN_LEN: #if max_result_token < MIN_PREDICT_TOKEN_LEN:
# return LLMResult.from_error_str(f"prompt too long,can not predict") # return LLMResult.from_error_str(f"prompt too long,can not predict")
@@ -221,8 +230,7 @@ class LLMAgentBaseProcess(BaseLLMProcess):
self.workspace : AgentWorkspace = None # If Workspace is not none , enable Agent Tasklist self.workspace : AgentWorkspace = None # If Workspace is not none , enable Agent Tasklist
self.memory : AgentMemory = None self.memory : AgentMemory = None
self.enable_kb : bool = False self.enable_kb_list : List[str] = None
self.kb = None
async def initial(self,params:Dict = None) -> bool: async def initial(self,params:Dict = None) -> bool:
self.memory = params.get("memory") self.memory = params.get("memory")
@@ -257,26 +265,49 @@ class LLMAgentBaseProcess(BaseLLMProcess):
if config.get("context"): if config.get("context"):
self.context = config.get("context") self.context = config.get("context")
if config.get("knowledge_grpah_introduce"):
self.knowledge_grpah_introduce = config.get("knowledge_grpah_introduce")
self.llm_context = SimpleLLMContext() self.llm_context = SimpleLLMContext()
if config.get("llm_context"): if config.get("llm_context"):
self.llm_context.load_from_config(config.get("llm_context")) self.llm_context.load_from_config(config.get("llm_context"))
if config.get("enable_kb"): def prepare_knowledge_grpah_prompt(self) -> Dict:
self.enable_kb = config.get("enable_kb") == "true" result = {}
result["introduce"] = BaseKnowledgeGraph.get_kb_default_desc_str()
result["knowledge_graph_list"] = {}
have_kb = False
if self.memory.enable_knowledge_graph:
result["knowledge_graph_list"][self.memory.knowledge_graph.kb_id] = self.memory.knowledge_graph.get_description()
have_kb = True
if self.enable_kb_list:
for kb_id in self.enable_kb_list:
kb = BaseKnowledgeGraph.get_kb(kb_id)
if kb:
have_kb = True
result["knowledge_graph_list"][kb_id] = kb.get_description()
else:
logger.error(f"knowledge base {kb_id} not found")
if have_kb is False:
return None
return result
def prepare_role_system_prompt(self,context_info:Dict) -> Dict: def prepare_role_system_prompt(self,context_info:Dict) -> Dict:
system_prompt_dict = {} system_prompt_dict = {}
# System Prompt
## LLM的身份说明
system_prompt_dict["role_description"] = self.role_description
#prompt.append_system_message(self.role_description)
## 处理信息的流程说明 system_prompt_dict["role_description"] = self.role_description
system_prompt_dict["process_rule"] = self.process_description system_prompt_dict["process_rule"] = self.process_description
#prompt.append_system_message(self.process_description)
### 回复的格式
system_prompt_dict["reply_format"] = self.reply_format system_prompt_dict["reply_format"] = self.reply_format
#prompt.append_system_message(self.reply_format)
kb_prompt = self.prepare_knowledge_grpah_prompt()
if kb_prompt:
system_prompt_dict["knowledge_graph"] = kb_prompt
## Context ## Context
if self.context: if self.context:
@@ -293,9 +324,13 @@ class LLMAgentBaseProcess(BaseLLMProcess):
def get_action_desc(self) -> Dict: def get_action_desc(self) -> Dict:
result = {} result = {}
actions_list = self.llm_context.get_all_ai_action() actions_list = []
actions_list.extend(self.llm_context.get_all_ai_action())
for action in actions_list: for action in actions_list:
result[action.get_name()] = action.get_description() result[action.get_name()] = action.get_description()
return result return result
async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: async def get_inner_function_for_exec(self,func_name:str) -> AIFunction:
@@ -429,14 +464,15 @@ class AgentMessageProcess(LLMAgentBaseProcess):
#TODO Is sender an agent? #TODO Is sender an agent?
return await self.memory.get_contact_summary(sender_id) return await self.memory.get_contact_summary(sender_id)
async def load_chatlogs(self,msg:AgentMsg)->str: async def load_chatlogs(self,msg:AgentMsg,max_length_by_token:int)->str:
## like ## like
#sender,[2023-11-1 12:00:00] #sender,[2023-11-1 12:00:00]
#content #content
return await self.memory.load_chatlogs(msg) return await self.memory.load_chatlogs(msg,max_length_by_token)
async def get_chat_summary(self,msg:AgentMsg)->str:
return await self.memory.get_chat_summary(msg)
async def get_log_summary(self,msg:AgentMsg)->str:
return None
async def get_extend_known_info(self,msg:AgentMsg,prompt:LLMPrompt)->str: async def get_extend_known_info(self,msg:AgentMsg,prompt:LLMPrompt)->str:
@@ -466,18 +502,7 @@ class AgentMessageProcess(LLMAgentBaseProcess):
### 信息发送者资料 ### 信息发送者资料
known_info["sender_info"] = await self.sender_info(msg) known_info["sender_info"] = await self.sender_info(msg)
#prompt.append_system_message(await self.sender_info(self,msg)) #prompt.append_system_message(await self.sender_info(self,msg))
### 近期的聊天记录
chat_record = await self.load_chatlogs(msg)
if chat_record:
if len(chat_record) > 4:
known_info["chat_record"] = chat_record
#prompt.append_system_message(await self.load_chatlogs(self,msg))
### 交流总结
summary = await self.get_log_summary(msg)
if summary:
if len(summary) > 4:
known_info["summary"] = summary
#prompt.append_system_message(await self.get_log_summary(self,msg))
system_prompt_dict["known_info"] = known_info system_prompt_dict["known_info"] = known_info
prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions())
@@ -485,15 +510,25 @@ class AgentMessageProcess(LLMAgentBaseProcess):
#TODO eanble workspace functions? #TODO eanble workspace functions?
logger.info(f"workspace is not none,enable workspace functions") logger.info(f"workspace is not none,enable workspace functions")
## 给予查询KB的权限
if self.enable_kb:
logger.info(f"enable kb")
### 根据Token Limit加载聊天记录
remain_token = self.get_remain_prompt_length(prompt,json.dumps(system_prompt_dict,ensure_ascii=False))
chat_record,is_all = await self.load_chatlogs(msg,remain_token - self.chat_summary_token_len)
if chat_record:
if len(chat_record) > 4:
known_info["chat_record"] = chat_record
if not is_all :
### 如果出触发了Token Limit,则删除几条信息后,加载summary (summary的长度基本是固定的)
summary = await self.get_chat_summary(msg)
if summary:
if len(summary) > 4:
known_info["chat_summary"] = summary
# TODO: extend known info
#prompt.append_system_message(await self.get_extend_known_info(msg,prompt))
prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False))
## 扩展已知信息 (这可能是一个LLM过程)
prompt.append_system_message(await self.get_extend_known_info(msg,prompt))
return prompt return prompt
@@ -527,117 +562,77 @@ class AgentMessageProcess(LLMAgentBaseProcess):
class AgentSelfThinking(LLMAgentBaseProcess): class AgentSelfThinking(LLMAgentBaseProcess):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__() super().__init__()
async def load_from_config(self, config: dict) -> Coroutine[Any, Any, bool]: async def load_from_config(self, config: dict) -> Coroutine[Any, Any, bool]:
if await super().load_from_config(config) is False: if await super().load_from_config(config) is False:
return False return False
async def _load_chat_history(self,token_limit:int):
chat_history = {}
session_list = AIChatSession.list_session(self.memory.agent_id ,self.memory.memory_db)
total_read_msg = 0
for session_id in session_list:
chatsession = AIChatSession.get_session_by_id(session_id,self.memory.memory_db)
session_history = {}
session_history["summary"] = chatsession.summary
session_history["id"] = chatsession.session_id
token_limit -= ComputeKernel.llm_num_tokens_from_text(chatsession.summary,self.model_name)
read_history_msg = 0
if token_limit > 8:
# load session chat history
cur_pos = chatsession.summarize_pos
messages = chatsession.read_history(0,cur_pos,"natural") # read
history_str = ""
for msg in messages:
read_history_msg += 1
total_read_msg += 1
cur_pos += 1
dt = datetime.fromtimestamp(float(msg.create_time))
formatted_time = dt.strftime('%y-%m-%d %H:%M:%S')
record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n"
token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str,self.model_name)
if token_limit < 8:
break
async def _get_history_prompt_for_think(self,chatsession,summary:str,system_token_len:int,pos:int)->(LLMPrompt,int): history_str = history_str + record_str
history_len = (self.max_token_size * 0.7) - system_token_len
messages = chatsession.read_history(self.history_len,pos,"natural") # read if ComputeKernel.llm_num_tokens_from_text(history_str,self.model_name) > self.chat_summary_token_len:
result_token_len = 0 session_history["history"] = history_str
result_prompt = LLMPrompt() chat_history[session_id] = session_history
have_summary = False chatsession.summarize_pos = cur_pos
if summary is not None:
if len(summary) > 1:
have_summary = True
if have_summary:
result_prompt.messages.append({"role":"user","content":summary})
result_token_len -= len(summary)
else:
result_prompt.messages.append({"role":"user","content":"There is no summary yet."})
result_token_len -= 6
read_history_msg = 0
history_str : str = ""
for msg in messages:
read_history_msg += 1
dt = datetime.datetime.fromtimestamp(float(msg.create_time))
formatted_time = dt.strftime('%y-%m-%d %H:%M:%S')
record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n"
history_str = history_str + record_str
history_len -= len(msg.body)
result_token_len += len(msg.body)
if history_len < 0:
logger.warning(f"_get_prompt_from_session reach limit of token,just read {read_history_msg} history message.")
break
result_prompt.messages.append({"role":"user","content":history_str})
return result_prompt,pos+read_history_msg
async def _think_chatsession(self,session_id):
if self.agent_think_prompt is None:
return
logger.info(f"agent {self.agent_id} think session {session_id}")
chatsession = AIChatSession.get_session_by_id(session_id,self.chat_db)
while True:
cur_pos = chatsession.summarize_pos
summary = chatsession.summary
prompt:LLMPrompt = LLMPrompt()
#prompt.append(self._get_agent_prompt())
prompt.append(await self._get_agent_think_prompt())
system_prompt_len = ComputeKernel.llm_num_tokens(prompt)
#think env?
history_prompt,next_pos = await self._get_history_prompt_for_think(chatsession,summary,system_prompt_len,cur_pos)
prompt.append(history_prompt)
is_finish = next_pos - cur_pos < 2
if is_finish:
logger.info(f"agent {self.agent_id} think session {session_id} is finished!,no more history")
break
#3) llm summarize chat history
task_result:ComputeTaskResult = await self.do_llm_complection(prompt)
if task_result.result_code != ComputeTaskResultCode.OK:
logger.error(f"think_chatsession llm compute error:{task_result.error_str}")
break
else: else:
new_summary= task_result.result_str logger.info(f"load_chat_history reach token limit,load {total_read_msg} history messages.")
logger.info(f"agent {self.agent_id} think session {session_id} from {cur_pos} to {next_pos} summary:{new_summary}") return chat_history
chatsession.update_think_progress(next_pos,new_summary)
return if total_read_msg < 2:
logger.info(f"load_chat_history: no history messages,return NONE")
return None
return chat_history
async def prepare_prompt(self,input:Dict) -> LLMPrompt: async def prepare_prompt(self,input:Dict) -> LLMPrompt:
prompt = LLMPrompt() prompt = LLMPrompt()
record_list = input.get("record_list")
context_info = input.get("context_info") context_info = input.get("context_info")
if record_list is None:
logger.error(f"AgentSelfThinking prepare_prompt failed! input not found")
return None
prompt.append_user_message(json.dumps(record_list,ensure_ascii=False))
system_prompt_dict = self.prepare_role_system_prompt(context_info) system_prompt_dict = self.prepare_role_system_prompt(context_info)
# Known_info is the SESSION summary of the existence, the current task work record summary, # Known_info is the SESSION summary of the existence, the current task work record summary,
known_info = {} token_remain = self.get_remain_prompt_length(prompt,json.dumps(system_prompt_dict,ensure_ascii=False))
have_known_info = False chat_history = await self._load_chat_history(token_remain)
known_session_list = input.get("known_session_list") if chat_history is None:
known_task_list = input.get("known_task_list") logger.info(f"prepare_prompt: no history messages,return NONE")
known_contact_list = input.get("known_contact_list") return None
known_experience_list = input.get("known_experience_list")
if known_session_list:
known_info["known_session_list"] = known_session_list
have_known_info = True
if known_task_list:
known_info["known_task_list"] = known_task_list
have_known_info = True
if known_contact_list:
known_info["known_contact_list"] = known_contact_list
have_known_info = True
if known_experience_list:
known_info["known_experience_list"] = known_experience_list
have_known_info = True
if have_known_info:
system_prompt_dict["known_info"] = known_info
prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions())
prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False))
prompt.append_user_message(json.dumps(chat_history,ensure_ascii=False))
return prompt
async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool:
action_params = {} action_params = {}
+5 -3
View File
@@ -8,6 +8,7 @@ from asyncio import Queue
from ..proto.compute_task import * from ..proto.compute_task import *
from ..knowledge import ObjectID from ..knowledge import ObjectID
from ..storage.storage import AIStorage
from .compute_node import ComputeNode from .compute_node import ComputeNode
@@ -105,7 +106,7 @@ class ComputeKernel:
return True return True
@staticmethod @staticmethod
def llm_num_tokens_from_text(text:str,model:str) -> int: def llm_num_tokens_from_text(text:str,model:str = None) -> int:
if model is None: if model is None:
model = "gpt-4-turbo-preview" model = "gpt-4-turbo-preview"
@@ -122,12 +123,13 @@ class ComputeKernel:
def llm_num_tokens(prompt: LLMPrompt, model_name: str = None) -> int: def llm_num_tokens(prompt: LLMPrompt, model_name: str = None) -> int:
return ComputeKernel.llm_num_tokens_from_text(prompt.as_str(), model_name) return ComputeKernel.llm_num_tokens_from_text(prompt.as_str(), model_name)
# friendly interface for use: # friendly interface for use:
def llm_completion(self, prompt: LLMPrompt, resp_mode:str="text",mode_name: Optional[str] = None, max_token: int = 0,inner_functions = None): def llm_completion(self, prompt: LLMPrompt, resp_mode:str="text",model_name: Optional[str] = None, max_token: int = 0,inner_functions = None):
# craete a llm_work_task ,push on queue's end # craete a llm_work_task ,push on queue's end
# then task_schedule would run this task.(might schedule some work_task to another host) # then task_schedule would run this task.(might schedule some work_task to another host)
task_req = ComputeTask() task_req = ComputeTask()
task_req.set_llm_params(prompt,resp_mode,mode_name, max_token,inner_functions) task_req.set_llm_params(prompt,resp_mode,model_name, max_token,inner_functions)
self.run(task_req) self.run(task_req)
return task_req return task_req
+3 -2
View File
@@ -1,6 +1,7 @@
from .object import * from .object import *
from .vector import * #from .vector import *
from .data import * from .data import *
from .store import KnowledgeStore from .store import KnowledgeStore
from .core_object import * from .core_object import *
from .pipeline import * #from .pipeline import *
from .knowledge_base import *
+371
View File
@@ -0,0 +1,371 @@
from abc import ABC, abstractmethod
import json
import os
import uuid
from typing import List
from ..proto.ai_function import ParameterDefine, SimpleAIAction, SimpleAIFunction
from ..agent.llm_context import GlobaToolsLibrary
from ..storage.objfs import ObjFS
import logging
logger = logging.getLogger(__name__)
class BaseKnowledgeGraph(ABC):
_all_knowledge_bases = {}
_default_kb = None
@classmethod
def get_kb(cls, kb_id:str):
if kb_id is None:
return cls._default_kb
return cls._all_knowledge_bases.get(kb_id)
@classmethod
def add_kb(cls,kb:'BaseKnowledgeGraph',is_default=False):
cls._all_knowledge_bases[kb.kb_id] = kb
if is_default:
cls._default_kb = kb
@classmethod
def remove_kb(cls,kb_id:str):
if cls._default_kb is not None and cls._default_kb.kb_id == kb_id:
cls._default_kb = None
if cls._all_knowledge_bases.get(kb_id) is not None:
del cls._all_knowledge_bases[kb_id]
def __init__(self, kb_id: str,kb_desc:str=None):
self.kb_id = kb_id
if kb_desc is None:
self.kb_desc = """
"""
else:
self.kb_desc = kb_desc
def get_description(self)->str:
return self.kb_desc
# 读接口: 查询,浏览
@abstractmethod
async def serach(self, query: str,query_type:str):
pass
@abstractmethod
async def get_obj_by_path(self,path)->str:
pass
@abstractmethod
async def get_obj_by_id(self,obj_id)->str:
pass
@abstractmethod
async def list_by_path(self,base_path)->List[str]:
pass
@abstractmethod
def list_source(self) -> List[str]:
pass
@abstractmethod
async def add_obj(self,obj_id,obj_name,obj_content,paths) -> bool:
pass
@abstractmethod
async def remove(self,remove_path) -> bool:
pass
@abstractmethod
async def remove_obj(self,objid):
pass
@abstractmethod
async def link(self,obj_id,paths) -> bool:
pass
@abstractmethod
async def unlink(self,paths) -> bool:
pass
@abstractmethod
async def update_obj(self,obj_id,new_content) -> bool:
pass
@staticmethod
def get_kb_default_desc_str():
return """The basic design of the Knowledge Graph is
1. Each object can be described in JSON, and have a unique obj_id.
2. The object can be accessed through the PATH, and multiple paths can point to the same object.
3. Carefully understand the semantics of the path, and follow the description of the knowledge graph.You can list all the sub-paths of a path through the LIST operation
All Knowledge Graph APIs return are json format string."""
# 写接口:通常由KnowledgePipeline调用
@staticmethod
def register_ai_functions():
async def knowledge_graph_access(parameters):
kb_id = parameters['kb_id']
op_name = parameters['op']
param = parameters['param']
if op_name is None:
logger.error("Operation type is not specified")
return "Operation type is not specified"
if param is None:
logger.error("Operation parameters is not specified")
return "Error! Operation parameters is not specified"
param = json.loads(param)
kb = BaseKnowledgeGraph.get_kb(kb_id)
if kb is None:
logger.error(f"Knowledge base is not found id:{kb_id}")
return "Error! Knowledge base is not found"
if op_name == "list":
root_path = param.get("path")
if root_path is None:
logger.error("Path is not specified")
return "Error! Path is not specified"
return json.dumps(await kb.list_by_path(root_path), ensure_ascii=False)
if op_name == "tree":
root_path = param.get("path")
if root_path is None:
logger.error("Path is not specified")
return "Error! Path is not specified"
depth = param.get("depth")
if depth is None:
depth = 3
return json.dumps(await kb.tree(root_path,depth), ensure_ascii=False)
if op_name == "read":
obj_path = param.get("path")
if obj_path is None:
logger.error("Path is not specified")
return "Error! Path is not specified"
return json.dumps(await kb.get_obj_by_path(obj_path), ensure_ascii=False)
if op_name == "get_obj":
obj_id = param.get("obj_id")
if obj_id is None:
logger.error("Object ID is not specified")
return "Error! Object ID is not specified"
return json.dumps(await kb.get_obj_by_id(obj_id), ensure_ascii=False)
return "Error! Operation type is not supported"
# search is not supported currently
func_desc = "Read knowledge graph, op_param format is as follows: list:{'path':$path}, read:{'path':$path}, get_obj:{'obj_id':$obj_id}, tree:{'path':$path,'depth':$depth}"
parameters = ParameterDefine.create_parameters({
"kb_id": "Knowledge Base ID",
"op": "Operation Type,could be [list, read, get_obj]",
"op_param": "Operation Param, must be a json string"
})
knowledge_graph_access_func = SimpleAIFunction("knowledge_base.knowledge_graph_read",
func_desc,
knowledge_graph_access,
parameters)
GlobaToolsLibrary.get_instance().register_tool_function(knowledge_graph_access_func)
async def knwoledge_graph_update(parameters):
kb_id = parameters['kb_id']
op_name = parameters['op']
param = parameters['param']
result = {}
if op_name is None:
logger.error("Operation type is not specified")
result["result"] = "Error! Operation type is not specified"
return json.dumps(result, ensure_ascii=False)
if param is None:
logger.error("Operation parameters is not specified")
result["result"] = "Error! Operation parameters is not specified"
return json.dumps(result, ensure_ascii=False)
param = json.loads(param)
kb = BaseKnowledgeGraph.get_kb(kb_id)
if kb is None:
logger.error(f"Knowledge base is not found id:{kb_id}")
result["result"] = "Error! Knowledge base is not found"
return json.dumps(result, ensure_ascii=False)
if op_name == "write":
write_path = param.get("path")
if write_path is None:
logger.error("Path is not specified")
result["result"] = "Error! Path is not specified"
return json.dumps(result, ensure_ascii=False)
obj_content = param.get("obj_json")
if obj_content is None:
logger.error("Object content is not specified")
result["result"] = "Error! Object content is not specified"
return json.dumps(result, ensure_ascii=False)
objid = uuid.uuid4()
objname = os.path.basename(write_path)
paths = []
paths.append(write_path)
if await kb.add_obj(objid,objname,obj_content['content'],paths):
result["result"] = "OK"
result['obj_id'] = objid
else:
result["result"] = "Error! Add object failed"
if op_name == "remove":
remove_path = param.get("path")
if remove_path is None:
logger.error("Path is not specified")
result["result"] = "Error! Path is not specified"
return json.dumps(result, ensure_ascii=False)
if await kb.remove(remove_path):
result["result"] = "OK"
else:
result["result"] = "Error! Remove path failed"
if op_name == "remove_obj":
obj_id = param.get("obj_id")
if obj_id is None:
logger.error("Object ID is not specified")
result["result"] = "Error! Object ID is not specified"
return result
obj = await kb.get_obj_by_id(obj_id)
if obj is None:
logger.error(f"Object is not found id:{obj_id}")
result["result"] = "Error! Object is not found"
return result
await kb.remove_obj(obj_id)
result["result"] = "OK"
if op_name == "set_obj":
obj_id = param.get("obj_id")
if obj_id is None:
logger.error("Object ID is not specified")
result["result"] = "Error! Object ID is not specified"
return json.dumps(result, ensure_ascii=False)
obj = await kb.get_obj_by_id(obj_id)
if obj is None:
logger.error(f"Object is not found id:{obj_id}")
result["result"] = "Error! Object is not found"
return result
obj_content = param.get("obj_json")
if obj_content is None:
logger.error("new object is not specified")
result["result"] = "Error! new object is not specified"
return json.dumps(result, ensure_ascii=False)
await kb.update_obj(obj_id,obj_content)
result["result"] = "OK"
if op_name == "link":
path_from = param.get("path")
path_to = param.get("target")
if path_from is None or path_to is None:
logger.error("Path is not specified")
result["result"] = "Error! Path is not specified"
return json.dumps(result, ensure_ascii=False)
objid = await kb.get_obj_by_path(path_to)
if objid is None:
logger.error(f"Object is not found path:{path_to}")
result["result"] = "Error!Target Object is not found"
return json.dumps(result, ensure_ascii=False)
await kb.link(objid,[path_from])
result["result"] = "OK"
if op_name == "unlink":
path_will_remove = param.get("path")
if path_will_remove is None:
logger.error("Path is not specified")
result["result"] = "Error! Path is not specified"
return json.dumps(result, ensure_ascii=False)
await kb.unlink([path_will_remove])
result["result"] = "OK"
return json.dumps(result, ensure_ascii=False)
OperationParames = """Parameters is a json string, the format is as follows:
write:{'path':$path,'obj_json':$obj_json},
remove:{'path':$path},
remove_obj:{'obj_id':$obj_id},
set_obj:{'obj_id':$obj_id,'obj_json':$new_obj_json},
link:{'path':$path,'target':$target_obj_path},
unlink:{'path':$path}
"""
parameters = ParameterDefine.create_parameters({
"kb_id": "Knowledge Base ID",
"op": "Operation Type,could be [write, remove, remove_obj, set_obj, link, unlink",
"param": OperationParames
})
knowledge_graph_update_func = SimpleAIFunction("knowledge_base.knowledge_graph_update",
"Update Knowledge Graph APIs",
knwoledge_graph_update,
parameters)
GlobaToolsLibrary.get_instance().register_tool_function(knowledge_graph_update_func)
class ObjFSKnowledgeGrpah(BaseKnowledgeGraph):
def __init__(self, kb_id:str,db_path:str,kb_desc:str=None):
super().__init__(kb_id,kb_desc)
self.db_path = db_path
self.obj_storage : ObjFS = ObjFS(db_path)
async def serach(self, query: str,query_type:str):
pass
def list_source(self):
pass
async def get_obj_by_path(self,path)->str:
return self.obj_storage.get_obj_by_path(path)
async def get_obj_by_id(self,obj_id)->str:
return self.obj_storage.get_obj_by_id(obj_id)
async def list_by_path(self,base_path)->List[str]:
return self.obj_storage.list_paths(base_path)
async def tree(self,base_path,depth:int)->str:
return self.obj_storage.tree(base_path,depth)
async def add_obj(self,obj_id,obj_name,obj_content,paths)->bool:
self.obj_storage.add_obj(obj_id,obj_name,obj_content,paths)
#todo 更新默认是做dict的merge
async def update_obj(self, obj_id, new_content)->bool:
return self.obj_storage.update_obj(obj_id,new_content)
async def remove(self,remove_path)->bool:
self.obj_storage.remove_path(remove_path)
async def remove_obj(self,objid)->bool:
self.obj_storage.remove_obj(objid)
async def link(self,from_path,target_path)->bool:
objid = self.obj_storage.get_obj_by_path(target_path)
if objid is None:
return False
self.obj_storage.add_path(objid,from_path)
return True
async def unlink(self,paths)->bool:
self.obj_storage.remove_path(paths)
-133
View File
@@ -1,133 +0,0 @@
import datetime
import sqlite3
import os
import logging
from . import ObjectID, KnowledgeStore
from enum import Enum
class KnowledgePipelineJournal:
def __init__(self, time: datetime.datetime, input: str, parser: str):
self.time = time
self.input = input
self.parser = parser
def is_finish(self) -> bool:
return self.input is None
def get_input(self) -> str:
return self.input
def get_parser(self) -> str:
return self.parser
def __str__(self) -> str:
if self.is_finish():
return f"{self.time}: finished)"
else:
return f"{self.time}: input:{self.input}, parser:{self.parser})"
# init sqlite3 client
class KnowledgePipelineJournalClient:
def __init__(self, pipeline_path: str = None):
if not os.path.exists(pipeline_path):
os.makedirs(pipeline_path)
self.journal_path = os.path.join(pipeline_path, "journal.db")
conn = sqlite3.connect(self.journal_path)
conn.execute(
'''CREATE TABLE IF NOT EXISTS journal (
id INTEGER PRIMARY KEY AUTOINCREMENT,
time DATETIME DEFAULT CURRENT_TIMESTAMP,
input TEXT,
parser TEXT)'''
)
conn.commit()
def insert(self, input: str, parser: str, timestamp: datetime.datetime = None):
timestamp = datetime.datetime.now() if timestamp is None else timestamp
conn = sqlite3.connect(self.journal_path)
conn.execute(
"INSERT INTO journal (time, input, parser) VALUES (?, ?, ?)",
(timestamp, input, parser),
)
conn.commit()
def latest_journals(self, topn) -> [KnowledgePipelineJournal]:
conn = sqlite3.connect(self.journal_path)
cursor = conn.cursor()
cursor.execute("SELECT * FROM journal ORDER BY id DESC LIMIT ?", (topn,))
return [KnowledgePipelineJournal(time, input, parser) for (_, time, input, parser) in cursor.fetchall()]
class KnowledgePipelineEnvironment:
def __init__(self, pipeline_path: str):
self.knowledge_store = KnowledgeStore()
if not os.path.exists(pipeline_path):
os.makedirs(pipeline_path)
self.pipeline_path = pipeline_path
self.journal = KnowledgePipelineJournalClient(pipeline_path)
self.logger = logging.getLogger()
def get_journal(self) -> KnowledgePipelineJournalClient:
return self.journal
def get_knowledge_store(self) -> KnowledgeStore:
return self.knowledge_store
def get_logger(self) -> logging.Logger:
return self.logger
class KnowledgePipelineState(Enum):
INIT = 0
RUNNING = 1
STOPPED = 2
FINISHED = 3
class NullParser:
async def parse(self, object_id):
return ""
class KnowledgePipeline:
def __init__(self, name: str, env: KnowledgePipelineEnvironment, input_init, input_params=None, parser_init=None, parser_params=None):
self.name = name
self.state = KnowledgePipelineState.INIT
self.input_init = input_init
self.input_params = input_params
self.parser_init = parser_init
self.parser_params = parser_params
self.env = env
self.input = None
self.parser = None
def get_name(self):
return self.name
def get_journal(self) -> KnowledgePipelineJournalClient:
return self.env.journal
async def run(self):
if self.state == KnowledgePipelineState.INIT:
self.input = self.input_init(self.env, self.input_params)
if self.parser_init is None:
self.parser = NullParser()
else:
self.parser = self.parser_init(self.env, self.parser_params)
self.state = KnowledgePipelineState.RUNNING
if self.state == KnowledgePipelineState.RUNNING:
async for input in self.input.next():
if input is None:
self.state = KnowledgePipelineState.FINISHED
self.env.journal.insert(None, None)
return
(object_id, input_journal) = input
if object_id is not None:
parser_journal = await self.parser.parse(object_id)
self.env.journal.insert(input_journal, parser_journal)
else:
return
if self.state == KnowledgePipelineState.STOPPED:
return
if self.state == KnowledgePipelineState.FINISHED:
return
-2
View File
@@ -1,2 +0,0 @@
from .vector_base import VectorBase
from .chroma_store import ChromaVectorStore
-51
View File
@@ -1,51 +0,0 @@
from .vector_base import VectorBase
from ..object import ObjectID
import chromadb
import logging
import os
class ChromaVectorStore(VectorBase):
def __init__(self, root_dir, model_name: str) -> None:
super().__init__(model_name)
logging.info(
"will init chroma vector store, model={}".format(model_name)
)
directory = os.path.join(root_dir, "vector")
logging.info("will use vector store: {}".format(directory))
client = chromadb.PersistentClient(
path=directory, settings=chromadb.Settings(anonymized_telemetry=False)
)
# client = chromadb.Client()
collection_name = "coll_{}".format(model_name)
logging.info("will init chroma colletion: %s", collection_name)
collection = client.get_or_create_collection(collection_name)
self.collection = collection
async def insert(self, vector: [float], id: ObjectID):
logging.info(f"will insert vector: {len(vector)} id: {str(id)}")
logging.debug(f"vector is {vector}")
self.collection.add(
embeddings=vector,
ids=str(id),
)
async def query(self, vector: [float], top_k: int) -> [ObjectID]:
ret = self.collection.query(
query_embeddings=vector,
n_results=top_k,
)
logging.info(f"query result {ret}")
if len(ret['ids']) == 0:
return []
return list(map(ObjectID.from_base58, ret["ids"][0]))
async def delete(self, id: ObjectID):
self.collection.delete(
ids=id,
)
-16
View File
@@ -1,16 +0,0 @@
# import the ObjectID class
from ..object import ObjectID
# define a vector base class
class VectorBase:
def __init__(self, model_name) -> None:
self.model_name = model_name
async def insert(self, vector: [float], id: ObjectID):
pass
async def query(self, vector: [float], top_k: int) -> [ObjectID]:
pass
async def delete(self, id: ObjectID):
pass
+1 -1
View File
@@ -61,7 +61,7 @@ class AgentMsg:
self.mentions:[] = None #use in group chat only self.mentions:[] = None #use in group chat only
#self.title:str = None #self.title:str = None
self.body:str = None self.body:str = None
self.body_mime:str = None #//default is "text/plain",encode is utf8 self.body_mime:str = "text/plain" #//default is "text/plain",encode is utf8
#type is call / action #type is call / action
self.func_name = None self.func_name = None
+1 -1
View File
@@ -8,7 +8,7 @@ class ParameterDefine:
self.type:str = "string" self.type:str = "string"
self.enum:List[str] = None self.enum:List[str] = None
self.description = desc self.description = desc
self.is_required = False self.is_required = True
@classmethod @classmethod
def create_parameters(cls,json_obj:dict) -> Dict[str,'ParameterDefine']: def create_parameters(cls,json_obj:dict) -> Dict[str,'ParameterDefine']:
+3 -3
View File
@@ -287,9 +287,9 @@ class ComputeTask:
self.callchain_id = callchain_id self.callchain_id = callchain_id
self.params["prompts"] = prompts.to_message_list() self.params["prompts"] = prompts.to_message_list()
self.params["resp_mode"] = resp_mode self.params["resp_mode"] = resp_mode
if model_name is None:
model_name = AIStorage.get_instance().get_user_config().get_value("llm_model_name") self.params["model_name"] = AIStorage.get_instance().get_user_config().llm_get_real_model_name(model_name)
self.params["model_name"] = model_name
if max_token_size is None: if max_token_size is None:
self.params["max_token_size"] = 4000 self.params["max_token_size"] = 4000
else: else:
+18
View File
@@ -0,0 +1,18 @@
from typing import List
class NamedObjectStorage:
def __init__(self, storage, name: str):
self.storage = storage
self.name = name
async def get(self, key: str) -> bytes:
return await self.storage.get(self.name, key)
async def put(self, key: str, data: bytes):
await self.storage.put(self.name, key, data)
async def delete(self, key: str):
await self.storage.delete(self.name, key)
async def list(self) -> List[str]:
return await self.storage.list(self.name)
+217
View File
@@ -0,0 +1,217 @@
from abc import ABC, abstractmethod
import sqlite3
from sqlite3 import Error
from typing import List
import threading
import time
import uuid
import logging
logger = logging.getLogger(__name__)
class ObjFSReader(ABC):
@abstractmethod
def get_obj_by_path(self,path):
pass
@abstractmethod
def get_obj_by_id(self,obj_id):
pass
@abstractmethod
def list_paths(self,base_path):
pass
#ObjFS provides structured data storage similar to brain-like, as an object storage layer of Agent Friendly
class ObjFS(ObjFSReader):
def __init__(self, db_file):
""" initialize db connection """
self.db_file = db_file
self._get_conn()
def _get_conn(self):
""" get db connection """
local = threading.local()
if not hasattr(local, 'conn'):
local.conn = self._create_connection(self.db_file)
return local.conn
def _create_connection(self, db_file):
""" create a database connection to a SQLite database """
conn = None
try:
conn = sqlite3.connect(db_file)
except Error as e:
logger.error("Error occurred while connecting to database: %s", e)
return None
if conn:
self._create_table(conn)
return conn
def _create_table(self, conn):
try:
conn.execute('''CREATE TABLE IF NOT EXISTS objects
(id TEXT PRIMARY KEY, name TEXT, content TEXT, created_at REAL, modified_at REAL, size INTEGER)''')
conn.execute('''CREATE TABLE IF NOT EXISTS paths
(id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT UNIQUE, obj_id TEXT, FOREIGN KEY(obj_id) REFERENCES objects(id))''')
except Error as e:
logger.error("Error occurred while creating tables: %s", e)
def close(self):
local = threading.local()
if not hasattr(local, 'conn'):
return
local.conn.close()
def add_obj(self,obj_uuid, name, content, paths) -> bool:
conn = self._get_conn()
c = conn.cursor()
#obj id是guid,由外部生成
# 获取当前时间戳
current_time = time.time()
# 计算内容大小
content_size = len(content.encode('utf-8'))
try:
# 插入对象
c.execute("INSERT INTO objects (id, name, content, created_at, modified_at, size) VALUES (?, ?, ?, ?, ?, ?)", (obj_uuid, name, content, current_time, current_time, content_size))
# 插入路径
for path in paths:
c.execute("INSERT OR IGNORE INTO paths (path, obj_id) VALUES (?, ?)", (path, obj_uuid))
conn.commit()
except Error as e:
logger.warning("Error occurred while adding object: %s", e)
return False
return True
def update_obj(self,obj_id, new_content) -> bool:
#UPDATE orders
#SET data = json_set(
# data,
# '$.items[1].price',
# 0.35
#)
#WHERE id = 1;
try:
conn = self._get_conn()
c = conn.cursor()
# 获取当前时间戳
current_time = time.time()
# 计算新内容大小
new_content_size = len(new_content.encode('utf-8'))
c.execute("UPDATE objects SET content = ?, modified_at = ?, size = ? WHERE id = ?", (new_content, current_time, new_content_size, obj_id))
conn.commit()
return True
except Error as e:
logger.warning("Error occurred while updating object: %s", e)
return False
def add_path(self,obj_id, new_path) -> bool:
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("INSERT OR IGNORE INTO paths (path, obj_id) VALUES (?, ?)", (new_path, obj_id))
conn.commit()
return True
except Error as e:
logger.warning("Error occurred while adding path: %s", e)
return False
def remove_path(self,path) -> bool:
try:
conn = self._get_conn()
c = conn.cursor()
#TODO
c.execute("DELETE FROM paths WHERE path = ?", (path,))
conn.commit()
return True
except Error as e:
logger.warning("Error occurred while removing path: %s", e)
return False
def remove_obj(self,obj_id) -> bool:
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("DELETE FROM objects WHERE id = ?", (obj_id,))
# 删除所有与该对象相关的路径
c.execute("DELETE FROM paths WHERE obj_id = ?", (obj_id,))
conn.commit()
return True
except Error as e:
logger.warning("Error occurred while removing object: %s", e)
return False
def get_obj_by_path(self,path) -> str:
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("SELECT objects.id, objects.name, objects.content FROM objects JOIN paths ON objects.id = paths.obj_id WHERE paths.path = ?", (path,))
obj_row = c.fetchone()
if obj_row:
return obj_row[2]
return None
except Error as e:
logger.warning("Error occurred while getting object by path: %s", e)
return None
def get_obj_by_id(self,obj_id) -> str:
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("SELECT id, name, content FROM objects WHERE id = ?", (obj_id,))
obj_row = c.fetchone()
if obj_row:
return obj_row[2]
return None
except Error as e:
logger.warning("Error occurred while getting object by id: %s", e)
return None
def list_paths(self,base_path)->List[str]:
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("SELECT path FROM paths WHERE path LIKE ? ESCAPE '/'", (base_path + "/%",))
return [row[0] for row in c.fetchall()]
except Error as e:
logger.warning("Error occurred while listing paths: %s", e)
return None
def tree(self, base_path,max_depth=3):
try:
conn = self._get_conn()
c = conn.cursor()
c.execute("SELECT path FROM paths WHERE path LIKE ? ESCAPE '/'", (base_path + "/%",))
paths = [row[0] for row in c.fetchall()]
tree = {}
for path in paths:
parts = path.split("/")
node = tree
for part in parts:
if part not in node:
node[part] = {}
node = node[part]
return tree
except Error as e:
logger.warning("Error occurred while listing paths: %s", e)
return None
+21 -1
View File
@@ -40,7 +40,10 @@ class UserConfig:
self.config_table = {} self.config_table = {}
self.user_config_path:str = None self.user_config_path:str = None
self._init_default_value("llm_model_name","gpt-4-turbo-preview") self._init_default_value("llm_default_model","gpt-4o")
self._init_default_value("llm_plan_model","gpt-4o")
self._init_default_value("llm_outline_model","gpt-3.5-turbo")
self._init_default_value("llm_swift_model","gpt-3.5-turbo")
def _init_default_value(self,key:str,value:Any) -> None: def _init_default_value(self,key:str,value:Any) -> None:
if self.config_table.get(key) is not None: if self.config_table.get(key) is not None:
@@ -52,6 +55,23 @@ class UserConfig:
self.config_table[key] = new_config_item self.config_table[key] = new_config_item
def llm_get_real_model_name(self,mode_name:str) -> str:
default_model_name = self.get_value("llm_default_model")
plan_llm_model_name = self.get_value("llm_plan_model")
outline_model_name = self.get_value("llm_outline_model")
swift_model_name = self.get_value("llm_swift_model")
if mode_name is None:
return default_model_name
if mode_name == "default":
return default_model_name
if mode_name == "plan_llm":
return plan_llm_model_name
if mode_name == "outline_llm":
return outline_model_name
if mode_name == "swift_llm":
return swift_model_name
return mode_name
def add_user_config(self,key:str,desc:str,is_optional:bool,default_value:Any=None,item_type="str") -> None: def add_user_config(self,key:str,desc:str,is_optional:bool,default_value:Any=None,item_type="str") -> None:
if self.config_table.get(key) is not None: if self.config_table.get(key) is not None:
logger.warning("user config key %s already exist, will be overrided",key) logger.warning("user config key %s already exist, will be overrided",key)
+2 -9
View File
@@ -6,7 +6,7 @@ import sys
import runpy import runpy
from typing import Any, Callable, Dict, List, Optional, Union from typing import Any, Callable, Dict, List, Optional, Union
from aios import AIAgent,AIAgentTemplete,AIStorage,BaseAIAgent,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask,WorkspaceEnvironment from aios import AIAgent,AIStorage,BaseAIAgent,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask,WorkspaceEnvironment
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -82,12 +82,6 @@ class AgentManager:
def remove(self,agent_id:str)->int: def remove(self,agent_id:str)->int:
pass pass
async def get_templete(self,templete_id) -> AIAgentTemplete:
template_media_info = self.agent_templete_env.get(templete_id)
if template_media_info is None:
return None
return self._load_templete_from_media(template_media_info)
def install(self,templete_id) -> PackageInstallTask: def install(self,templete_id) -> PackageInstallTask:
installer = self.agent_templete_env.get_installer() installer = self.agent_templete_env.get_installer()
return installer.install(templete_id) return installer.install(templete_id)
@@ -95,8 +89,7 @@ class AgentManager:
def uninstall(self,templete_id) -> int: def uninstall(self,templete_id) -> int:
pass pass
async def _load_templete_from_media(self,templete_media:PackageMediaInfo) -> AIAgentTemplete:
pass
async def _load_agent_from_media(self,agent_media:PackageMediaInfo) -> BaseAIAgent: async def _load_agent_from_media(self,agent_media:PackageMediaInfo) -> BaseAIAgent:
reader = self.agent_env._create_media_loader(agent_media) reader = self.agent_env._create_media_loader(agent_media)
+1 -1
View File
@@ -1,3 +1,3 @@
from .local_document import LocalKnowledgeBase, ScanLocalDocument, ParseLocalDocument #from .local_document import LocalKnowledgeBase, ScanLocalDocument, ParseLocalDocument
from .local_file_system import FilesystemEnvironment from .local_file_system import FilesystemEnvironment
from .shell import ShellEnvironment from .shell import ShellEnvironment
+21 -21
View File
@@ -9,7 +9,7 @@ import aiofiles
from urllib.parse import urlparse from urllib.parse import urlparse
from typing import Optional from typing import Optional
from aios import KnowledgeStore, ObjectType #from aios import KnowledgeStore, ObjectType
from aios.frame.tunnel import AgentTunnel from aios.frame.tunnel import AgentTunnel
from aios.proto.agent_msg import AgentMsg, AgentMsgType from aios.proto.agent_msg import AgentMsg, AgentMsgType
import discord import discord
@@ -165,26 +165,26 @@ class DiscordTunnel(AgentTunnel):
if len(resp_msg.body) < 1: if len(resp_msg.body) < 1:
return return
knownledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body) # knownledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body)
if knownledge_object is not None: # if knownledge_object is not None:
if knownledge_object.get_object_type() == ObjectType.Image: # if knownledge_object.get_object_type() == ObjectType.Image:
image = KnowledgeStore().bytes_from_object(knownledge_object) # image = KnowledgeStore().bytes_from_object(knownledge_object)
try: # try:
async with aiofiles.open("image.jpg", "wb") as f: # async with aiofiles.open("image.jpg", "wb") as f:
await f.write(image) # await f.write(image)
await message.channel.send(file=discord.File("image.jpg")) # await message.channel.send(file=discord.File("image.jpg"))
except Exception as e: # except Exception as e:
logger.error(f"save image error:{e}") # logger.error(f"save image error:{e}")
logger.exception(e) # logger.exception(e)
return # return
else: # else:
pos = resp_msg.body.find("audio file") # pos = resp_msg.body.find("audio file")
if pos != -1: # if pos != -1:
audio_file = resp_msg.body[pos+11:].strip() # audio_file = resp_msg.body[pos+11:].strip()
if audio_file.startswith("\""): # if audio_file.startswith("\""):
audio_file = audio_file[1:-1] # audio_file = audio_file[1:-1]
await message.channel.send(file=discord.File(audio_file)) # await message.channel.send(file=discord.File(audio_file))
return # return
await message.channel.send(resp_msg.body) await message.channel.send(resp_msg.body)
else: else:
if resp_msg.is_image_msg(): if resp_msg.is_image_msg():
+1 -1
View File
@@ -27,7 +27,7 @@ class OpenAI_ComputeNode(ComputeNode):
@classmethod @classmethod
def declare_user_config(cls): def declare_user_config(cls):
if os.getenv("OPENAI_API_KEY_") is None: if os.getenv("OPENAI_API_KEY") is None:
user_config = AIStorage.get_instance().get_user_config() user_config = AIStorage.get_instance().get_user_config()
user_config.add_user_config("openai_api_key","openai api key",False,None) user_config.add_user_config("openai_api_key","openai api key",False,None)
+21 -21
View File
@@ -12,7 +12,7 @@ import aiohttp
from slack_bolt.adapter.socket_mode.websockets import AsyncSocketModeHandler from slack_bolt.adapter.socket_mode.websockets import AsyncSocketModeHandler
from slack_bolt.app.async_app import AsyncApp from slack_bolt.app.async_app import AsyncApp
from aios import KnowledgeStore, ObjectType #from aios import KnowledgeStore, ObjectType
from aios.frame.tunnel import AgentTunnel from aios.frame.tunnel import AgentTunnel
from aios.proto.agent_msg import AgentMsg, AgentMsgType from aios.proto.agent_msg import AgentMsg, AgentMsgType
from aios.storage.storage import AIStorage from aios.storage.storage import AIStorage
@@ -189,26 +189,26 @@ class SlackTunnel(AgentTunnel):
if len(resp_msg.body) < 1: if len(resp_msg.body) < 1:
return return
knownledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body) # knownledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body)
if knownledge_object is not None: # if knownledge_object is not None:
if knownledge_object.get_object_type() == ObjectType.Image: # if knownledge_object.get_object_type() == ObjectType.Image:
image = KnowledgeStore().bytes_from_object(knownledge_object) # image = KnowledgeStore().bytes_from_object(knownledge_object)
try: # try:
async with aiofiles.open("image.jpg", "wb") as f: # async with aiofiles.open("image.jpg", "wb") as f:
await f.write(image) # await f.write(image)
await app.client.files_upload_v2(channel=event["channel"], file="image.jpg") # await app.client.files_upload_v2(channel=event["channel"], file="image.jpg")
except Exception as e: # except Exception as e:
logger.error(f"save image error:{e}") # logger.error(f"save image error:{e}")
logger.exception(e) # logger.exception(e)
return # return
else: # else:
pos = resp_msg.body.find("audio file") # pos = resp_msg.body.find("audio file")
if pos != -1: # if pos != -1:
audio_file = resp_msg.body[pos+11:].strip() # audio_file = resp_msg.body[pos+11:].strip()
if audio_file.startswith("\""): # if audio_file.startswith("\""):
audio_file = audio_file[1:-1] # audio_file = audio_file[1:-1]
await app.client.files_upload_v2(channel=event["channel"], file=audio_file) # await app.client.files_upload_v2(channel=event["channel"], file=audio_file)
return # return
await app.client.chat_postMessage(channel=event["channel"], text=resp_msg.body) await app.client.chat_postMessage(channel=event["channel"], text=resp_msg.body)
else: else:
if resp_msg.is_image_msg(): if resp_msg.is_image_msg():
-1
View File
@@ -1 +0,0 @@
from .local_st_compute_node import *
@@ -1,212 +0,0 @@
import logging
import requests
from typing import Optional, List
from pydantic import BaseModel
from typing import Union
from PIL import Image
import io
from aios import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType,ComputeTaskResultCode,ComputeNode,AIStorage,UserConfig,ObjectID,Queue_ComputeNode
logger = logging.getLogger(__name__)
class LocalSentenceTransformer_Text_ComputeNode(Queue_ComputeNode):
# For valid pretrained models, see https://www.sbert.net/docs/pretrained_models.html
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
super().__init__()
self.node_id = "local_sentence_transformer_text_embedding_node"
self.model_name = model_name
self.model = None
def initial(self) -> bool:
logger.info(
f"LocalSentenceTransformer_Text_ComputeNode init, model_name: {self.model_name}"
)
assert self.model_name is not None
assert self.model is None
try:
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(self.model_name)
except Exception as err:
logger.error(f"load model {self.model} failed: {err}")
return False
self.start()
return True
async def execute_task(self, task: ComputeTask) :
result = ComputeTaskResult()
result.result_code = ComputeTaskResultCode.ERROR
result.set_from_task(task)
result.worker_id = self.node_id
try:
# logger.debug(f"LocalSentenceTransformer_Text_ComputeNode task: {task}")
if task.task_type == ComputeTaskType.TEXT_EMBEDDING:
input = task.params["input"]
logger.debug(
f"LocalSentenceTransformer_Text_ComputeNode task input: {input}"
)
sentence_embeddings = self.model.encode(input, show_progress_bar=False).tolist()
# logger.debug(f"LocalSentenceTransformer_Text_ComputeNode task sentence_embeddings: {sentence_embeddings}")
result.result_code = ComputeTaskResultCode.OK
result.result["content"] = sentence_embeddings
else:
result.error_str = f"unsupport embedding task type: {task.task_type}"
except Exception as err:
import traceback
logger.error(f"{traceback.format_exc()}, error: {err}")
result.error_str = f"{traceback.format_exc()}, error: {err}"
return result
def display(self) -> str:
return f"LocalSentenceTransformer_Text_ComputeNode: {self.node_id}, {self.model_name}"
def get_capacity(self):
pass
def is_support(self, task: ComputeTask) -> bool:
return task.task_type == ComputeTaskType.TEXT_EMBEDDING and task.params["model_name"] == "all-MiniLM-L6-v2"
def is_local(self) -> bool:
return True
class LocalSentenceTransformer_Image_ComputeNode(Queue_ComputeNode):
# For valid pretrained models, see https://www.sbert.net/docs/pretrained_models.html
def __init__(
self,
model_name: str = "clip-ViT-B-32",
multi_model_name: str = "clip-ViT-B-32-multilingual-v1",
):
super().__init__()
self.node_id = "local_sentence_transformer_image_embedding_node"
self.model_name = model_name
self.multi_model_name = multi_model_name
self.model = None
self.multi_model = None
def initial(self) -> bool:
logger.info(
f"LocalSentenceTransformer_Image_ComputeNode init, model_name: {self.model_name} {self.multi_model_name}"
)
assert self.model_name is not None
assert self.multi_model_name is not None
assert self.model is None
assert self.multi_model is None
try:
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(self.model_name)
self.multi_model = SentenceTransformer(self.multi_model_name)
except Exception as err:
logger.error(f"load model {self.model} failed: {err}")
return False
self.start()
return True
def _load_image(self, source: Union[ObjectID, bytes]) -> Optional[Image]:
image_data = None
if isinstance(source, ObjectID):
from aios import KnowledgeStore, ImageObject
buf = KnowledgeStore().get_object_store().get_object(source)
if buf is None:
logger.error(f"load image object but not found! {source}")
return None
try:
image_obj = ImageObject.decode(buf)
except Exception as err:
logger.error(f"decode ImageObject from buffer failed: {source}, {err}")
return None
file_size = image_obj.get_file_size()
# print(f"got image object: {source.to_base58()}, size: {file_size}")
image_data = (
KnowledgeStore()
.get_chunk_reader()
.read_chunk_list_to_single_bytes(image_obj.get_chunk_list())
)
elif isinstance(source, bytes):
image_data = source
else:
logger.error(f"unsupport image source type: {type(source)}, {source}")
return None
try:
img = Image.open(io.BytesIO(image_data))
return img
except Exception as err:
logger.error(f"load image from buffer failed: {source}, {err}")
return None
async def execute_task(
self, task: ComputeTask
) -> ComputeTaskResult:
result = ComputeTaskResult()
result.result_code = ComputeTaskResultCode.ERROR
result.set_from_task(task)
result.worker_id = self.node_id
try:
# logger.debug(f"LocalSentenceTransformer_Text_ComputeNode task: {task}")
if task.task_type == ComputeTaskType.TEXT_EMBEDDING:
input = task.params["input"]
logger.debug(
f"LocalSentenceTransformer_Text_ComputeNode task text input: {input}"
)
sentence_embeddings = self.multi_model.encode(input, show_progress_bar=False).tolist()
# logger.debug(f"LocalSentenceTransformer_Text_ComputeNode task sentence_embeddings: {sentence_embeddings}")
result.result_code = ComputeTaskResultCode.OK
result.result["content"] = sentence_embeddings
elif task.task_type == ComputeTaskType.IMAGE_EMBEDDING:
input = task.params["input"]
logger.debug(
f"LocalSentenceTransformer_Image_ComputeNode task image input: {input}"
)
img = self._load_image(input)
if img is None:
result.error_str = f"load image failed: {input}"
return result
sentence_embeddings = self.model.encode(img, show_progress_bar=False).tolist()
result.result_code = ComputeTaskResultCode.OK
result.result["content"] = sentence_embeddings
else:
result.error_str = f"unsupport embedding task type: {task.task_type}"
except Exception as err:
import traceback
logger.error(f"{traceback.format_exc()}, error: {err}")
result.error_str = f"{traceback.format_exc()}, error: {err}"
return result
def display(self) -> str:
return f"LocalSentenceTransformer_Image_ComputeNode: {self.node_id}, {self.model_name}"
def get_capacity(self):
pass
def is_support(self, task: ComputeTask) -> bool:
return (
(task.task_type == ComputeTaskType.TEXT_EMBEDDING and task.params["model_name"] == "clip-ViT-B-32")
or task.task_type == ComputeTaskType.IMAGE_EMBEDDING
)
def is_local(self) -> bool:
return True
+66 -58
View File
@@ -12,7 +12,7 @@ from telegram import Bot
from telegram.ext import Updater from telegram.ext import Updater
from telegram.error import Forbidden, NetworkError from telegram.error import Forbidden, NetworkError
from aios import ObjectType, KnowledgeStore,AgentTunnel,AIStorage,ContactManager,Contact,AgentMsg,AgentMsgType from aios import AgentTunnel,AIStorage,ContactManager,Contact,AgentMsg,AgentMsgType
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -160,6 +160,68 @@ class TelegramTunnel(AgentTunnel):
if not os.path.exists(path): if not os.path.exists(path):
os.makedirs(path) os.makedirs(path)
return path return path
async def conver_agent_msg_to_tg_msg(self,resp_msg:AgentMsg,update: Update):
if resp_msg.body_mime is None:
if resp_msg.body is None:
return
if len(resp_msg.body) < 1:
await update.message.reply_text("")
return
# knowledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body)
# knowledge_object = None
# if knowledge_object is not None:
# if knowledge_object.get_object_type() == ObjectType.Image:
# image = KnowledgeStore().bytes_from_object(knowledge_object)
# try:
# async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file:
# if local_file:
# await local_file.write(image)
# await update.message.reply_photo("tg_send_temp.png")
# except Exception as e:
# logger.error(f"save image error: {e}")
# return
# else:
# pos = resp_msg.body.find("audio file")
# if pos != -1:
# audio_file = resp_msg.body[pos+11:].strip()
# if audio_file.startswith("\""):
# audio_file = audio_file[1:-1]
# await update.message.reply_voice(audio_file)
# return
await update.message.reply_text(resp_msg.body)
else:
if resp_msg.is_image_msg():
text, images = resp_msg.get_image_body()
if text is not None:
await update.message.reply_text(text)
for image in images:
if os.path.exists(image):
await update.message.reply_photo(image)
else:
await update.message.reply_text(image)
elif resp_msg.is_video_msg():
text, video_file = resp_msg.get_video_body()
if text is not None:
await update.message.reply_text(text)
if os.path.exists(video_file):
await update.message.reply_video(video_file)
else:
await update.message.reply_text(video_file)
elif resp_msg.is_audio_msg():
text, audio_file = resp_msg.get_audio_body()
if text is not None:
await update.message.reply_text(text)
if os.path.exists(audio_file):
await update.message.reply_voice(audio_file)
else:
await update.message.reply_text(audio_file)
else:
await update.message.reply_text(resp_msg.body)
async def conver_tg_msg_to_agent_msg(self,message:Message) -> AgentMsg: async def conver_tg_msg_to_agent_msg(self,message:Message) -> AgentMsg:
agent_msg = AgentMsg() agent_msg = AgentMsg()
@@ -246,6 +308,7 @@ class TelegramTunnel(AgentTunnel):
return False return False
# main entry for telegram message
async def on_message(self, bot:Bot, update: Update) -> None: async def on_message(self, bot:Bot, update: Update) -> None:
message = update.message message = update.message
logger.info(f"on_message: {message.message_id} from {message.from_user.username} ({update.effective_user.username}) to {message.chat.title}({message.chat.id})") logger.info(f"on_message: {message.message_id} from {message.from_user.username} ({update.effective_user.username}) to {message.chat.title}({message.chat.id})")
@@ -312,63 +375,8 @@ class TelegramTunnel(AgentTunnel):
if resp_msg is None: if resp_msg is None:
await update.message.reply_text(f"System Error: Timeout,{self.target_id} no resopnse! Please check logs/aios.log for more details!") await update.message.reply_text(f"System Error: Timeout,{self.target_id} no resopnse! Please check logs/aios.log for more details!")
else: else:
if resp_msg.body_mime is None: await self.conver_agent_msg_to_tg_msg(resp_msg,update)
if resp_msg.body is None:
return
if len(resp_msg.body) < 1:
await update.message.reply_text("")
return
knowledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body)
if knowledge_object is not None:
if knowledge_object.get_object_type() == ObjectType.Image:
image = KnowledgeStore().bytes_from_object(knowledge_object)
try:
async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file:
if local_file:
await local_file.write(image)
await update.message.reply_photo("tg_send_temp.png")
except Exception as e:
logger.error(f"save image error: {e}")
return
else:
pos = resp_msg.body.find("audio file")
if pos != -1:
audio_file = resp_msg.body[pos+11:].strip()
if audio_file.startswith("\""):
audio_file = audio_file[1:-1]
await update.message.reply_voice(audio_file)
return
await update.message.reply_text(resp_msg.body)
else:
if resp_msg.is_image_msg():
text, images = resp_msg.get_image_body()
if text is not None:
await update.message.reply_text(text)
for image in images:
if os.path.exists(image):
await update.message.reply_photo(image)
else:
await update.message.reply_text(image)
elif resp_msg.is_video_msg():
text, video_file = resp_msg.get_video_body()
if text is not None:
await update.message.reply_text(text)
if os.path.exists(video_file):
await update.message.reply_video(video_file)
else:
await update.message.reply_text(video_file)
elif resp_msg.is_audio_msg():
text, audio_file = resp_msg.get_audio_body()
if text is not None:
await update.message.reply_text(text)
if os.path.exists(audio_file):
await update.message.reply_voice(audio_file)
else:
await update.message.reply_text(audio_file)
else:
await update.message.reply_text(resp_msg.body)
-1
View File
@@ -1 +0,0 @@
# might implement by Rust in the future
Binary file not shown.
+159
View File
@@ -0,0 +1,159 @@
aiofiles>=23.2.1
aiohttp>=3.8.5
aioimaplib>=1.0.1
aiosignal>=1.3.1
aiosmtplib>=2.0.2
anyio>=4.0.0
async-timeout>=4.0.3
attrs>=23.1.0
backoff>=2.2.1
base36>=0.1.1
base58>=2.1.1
beautifulsoup4>=4.12.2
cachetools>=5.3.1
certifi>=2023.7.22
charset-normalizer>=3.2.0
chroma-hnswlib>=0.7.1
chromadb>=0.4.0
click>=8.1.7
colorama>=0.4.6
coloredlogs>=15.0.1
decorator>=4.4.2
fastapi
filelock>=3.12.3
flatbuffers>=23.5.26
frozenlist>=1.4.0
fsspec>=2023.9.0
google>=3.0.0
google-api-core>=2.11.1
google-auth>=2.23.0
google-cloud>=0.34.0
google-cloud-texttospeech>=2.14.1
googleapis-common-protos>=1.60.0
h11>=0.14.0
httpcore>=0.17.3
httptools>=0.6.0
httpx>=0.24.1
huggingface-hub>=0.16.4
humanfriendly>=10.0
idna>=3.4
imageio>=2.31.3
imageio-ffmpeg>=0.4.8
importlib-resources>=6.0.1
mail-parser>=3.15.0
monotonic>=1.6
moviepy>=1.0.0
mpmath>=1.3.0
multidict>=6.0.4
numpy>=1.25.2
onnxruntime>=1.15.1
overrides>=7.4.0
packaging>=23.1
pandas>=2.1.0
Pillow>=10.0.0
posthog>=3.0.2
proglog>=0.1.10
prompt-toolkit>=3.0.39
proto-plus>=1.22.3
pulsar-client>=3.3.0
pyasn1>=0.5.0
pyasn1-modules>=0.3.0
pydantic
PyPika>=0.48.9
pyreadline3>=3.4.1
python-dateutil>=2.8.2
python-dotenv>=1.0.0
python-telegram-bot>=20.5
pytz>=2023.3.post1
PyYAML>=6.0.1
requests>=2.31.0
rsa>=4.9
simplejson>=3.19.1
six>=1.16.0
sniffio>=1.3.0
soupsieve>=2.5
starlette>=0.27.0
sympy>=1.12
tokenizers>=0.14.0
toml>=0.10.0
protobuf
grpcio
grpcio-status
h11==0.14.0
httpcore==0.17.3
httptools==0.6.0
httpx==0.24.1
huggingface-hub==0.16.4
humanfriendly==10.0
idna==3.4
imageio==2.31.3
imageio-ffmpeg==0.4.8
importlib-resources==6.0.1
mail-parser==3.15.0
monotonic==1.6
moviepy==1.0.0
mpmath==1.3.0
multidict==6.0.4
numpy==1.25.2
onnxruntime==1.15.1
overrides==7.4.0
packaging==23.1
pandas==2.1.0
Pillow==10.0.0
posthog==3.0.2
proglog==0.1.10
prompt-toolkit==3.0.39
proto-plus==1.22.3
protobuf
pulsar-client==3.3.0
pyasn1==0.5.0
pyasn1-modules==0.3.0
pydantic==1.10.12
PyPika==0.48.9
pyreadline3==3.4.1
python-dateutil==2.8.2
python-dotenv==1.0.0
python-telegram-bot==20.5
pytz==2023.3.post1
PyYAML==6.0.1
requests==2.31.0
rsa==4.9
simplejson==3.19.1
six==1.16.0
sniffio==1.3.0
soupsieve==2.5
starlette==0.27.0
sympy==1.12
telegram==0.0.1
tokenizers==0.14.0
toml==0.10.0
pysocks
chardet
pydub
aiosqlite
python-telegram-bot
pydub
stability_sdk
sentence-transformers==2.2.2
tiktoken
markdown
PyPDF2
srt
webvtt-py
openai
docker
generic_escape
duckduckgo-search
SQLAlchemy
mysqlclient
psycopg2-binary
pyodbc
oracledb
html2text
docx2txt
opencv-python
discord.py
slack_bolt
wget
moviepy
+23 -20
View File
@@ -38,16 +38,16 @@ from google_node import *
from llama_node import * from llama_node import *
from openai_node import * from openai_node import *
from sd_node import * from sd_node import *
from st_node import *
from agent_manager import AgentManager from agent_manager import AgentManager
from workflow_manager import WorkflowManager from workflow_manager import WorkflowManager
from knowledge_manager import KnowledgePipelineManager #from knowledge_manager import KnowledgePipelineManager
from tg_tunnel import TelegramTunnel from tg_tunnel import TelegramTunnel
from email_tunnel import EmailTunnel from email_tunnel import EmailTunnel
from discord_tunnel import DiscordTunnel from discord_tunnel import DiscordTunnel
from slack_tunnel import SlackTunnel from slack_tunnel import SlackTunnel
from common_environment import LocalKnowledgeBase, FilesystemEnvironment, ShellEnvironment, ScanLocalDocument, ParseLocalDocument from common_environment import FilesystemEnvironment, ShellEnvironment
#from common_environment import ScanLocalDocument, ParseLocalDocument
from compute_node_config import * from compute_node_config import *
@@ -152,7 +152,10 @@ class AIOS_Shell:
#AgentManager.get_instance().register_environment("fs", FilesystemEnvironment) #AgentManager.get_instance().register_environment("fs", FilesystemEnvironment)
#AgentManager.get_instance().register_environment("knowledge", LocalKnowledgeBase) #AgentManager.get_instance().register_environment("knowledge", LocalKnowledgeBase)
AgentWorkspace.register_ai_functions() AgentWorkspace.register_ai_functions()
AgentMemory.register_ai_functions()
BaseKnowledgeGraph.register_ai_functions()
ShellEnvironment.register_ai_functions() ShellEnvironment.register_ai_functions()
if await AgentManager.get_instance().initial() is not True: if await AgentManager.get_instance().initial() is not True:
logger.error("agent manager initial failed!") logger.error("agent manager initial failed!")
@@ -211,17 +214,17 @@ class AIOS_Shell:
local_st_text_compute_node = LocalSentenceTransformer_Text_ComputeNode() # local_st_text_compute_node = LocalSentenceTransformer_Text_ComputeNode()
if local_st_text_compute_node.initial() is not True: # if local_st_text_compute_node.initial() is not True:
logger.error("local sentence transformer text embedding node initial failed!") # logger.error("local sentence transformer text embedding node initial failed!")
else: # else:
ComputeKernel.get_instance().add_compute_node(local_st_text_compute_node) # ComputeKernel.get_instance().add_compute_node(local_st_text_compute_node)
local_st_image_compute_node = LocalSentenceTransformer_Image_ComputeNode() # local_st_image_compute_node = LocalSentenceTransformer_Image_ComputeNode()
if local_st_image_compute_node.initial() is not True: # if local_st_image_compute_node.initial() is not True:
logger.error("local sentence transformer image embedding node initial failed!") # logger.error("local sentence transformer image embedding node initial failed!")
else: # else:
ComputeKernel.get_instance().add_compute_node(local_st_image_compute_node) # ComputeKernel.get_instance().add_compute_node(local_st_image_compute_node)
await ComputeKernel.get_instance().start() await ComputeKernel.get_instance().start()
@@ -230,12 +233,12 @@ class AIOS_Shell:
#AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) #AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
pipelines = KnowledgePipelineManager.initial(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines")) #pipelines = KnowledgePipelineManager.initial(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines"))
pipelines.register_input("scan_local", ScanLocalDocument) #pipelines.register_input("scan_local", ScanLocalDocument)
pipelines.register_parser("parse_local", ParseLocalDocument) #pipelines.register_parser("parse_local", ParseLocalDocument)
pipelines.load_dir(os.path.join(AIStorage().get_instance().get_system_app_dir(), "knowledge_pipelines")) #pipelines.load_dir(os.path.join(AIStorage().get_instance().get_system_app_dir(), "knowledge_pipelines"))
pipelines.load_dir(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge_pipelines")) #pipelines.load_dir(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge_pipelines"))
asyncio.create_task(pipelines.run()) #asyncio.create_task(pipelines.run())
TelegramTunnel.register_to_loader() TelegramTunnel.register_to_loader()
EmailTunnel.register_to_loader() EmailTunnel.register_to_loader()
@@ -256,7 +259,7 @@ class AIOS_Shell:
def get_version(self) -> str: def get_version(self) -> str:
return "0.5.1" return "0.5.2"
async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None, msg_mime:str=None) -> str: async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None, msg_mime:str=None) -> str:
if sender == self.username: if sender == self.username: