tangger/matagent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
Yutang Li
2025-01-18 17:53:58 +08:00
parent 938c486ddf
commit 981209d677
175 changed files with 195 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
_backend/tools.py Normal file → Executable file
View File

@@ -3,7 +3,7 @@ import requests
def retrieval_from_knowledge_base(
query: str,
topk: int
topk: int = 3
) -> str:
"""
Retrieval for knowledge from the knowledge base based on the specified query and returns the topk results.
@@ -91,44 +91,80 @@ def scheme_convert_to_json():
转换合成方案时，必须严格遵守如下预定义的JSON格式，每个JSON结构的字段必须填充完整，即使有些字段留空：
```json
{
"workflow": [
{
"step_id": <step_number>,
"description": "<brief_description>",
"actions": [
{
"action_id": <action_number>,
"action_type": "<action_type>",
"materials": [{"name": "<material_name>", "amount": <amount>, "unit": "<unit>"}],
"containers": [{"type": "<container_type>", "capacity": "<capacity>", "unit": "<unit>", "additional_parameters": {"material_of_construction": "<material_type>", "shape": "<shape>"}}],
"equipment": [{"name": "<equipment_name>", "parameters": {"<param_name>": <param_value>, "<param_name>": <param_value>}}],
"output": "<action_output>"
}
],
"dependencies": ["<previous_step_ids>"],
"step_output": "<step_output_identifier>"
"task_id": "",
"experiment_name": "",
"materials": [
{ "material_id": "", "name": "", "amount": "", "unit": "", "purity": "", "state": ""}
// ...可在此处添加更多material对象
],
"containers": [{
"container_id": "", "name": "", "capacity": "", "unit": "", "material_of_construction": "", "shape": "", "heat_resistant": "", "pressure_rating": "",
// ...可在此处添加更多container对象
],
"equipments": [{
"equipment_id": "", "name": "",
"parameters": {
// 具体设备参数（例如 rpm 范围, 温度范围等）
},
}
...
// ...可在此处添加更多equipment对象
],
"robot_workflow": [{
"step_id": "", "description": "",
"actions": [{
"action_type": "", // limited robot action: "pick_container""place_container""pick_container_with_material""place_container_into_equipment""remove_container_from_equipment"
"container_id": "",
"material_id": "",
"equipment_id": "",
}
// ...可在此处添加更多子动作
],
"dependencies": [
// 若需要依赖之前的若干 step_id，可列在这里，如 ["1", "2"]
],
"step_output": {
"container_id": "",
"contents": [
{
"material_id": "",
"amount": "",
"unit": ""
}
// ...可在此处列出执行完本步骤后容器中的产物或状态
]
}
}
// ...可在此处添加更多step对象
]
}
}
```
### JSON结构字段说明
1. workflow 类型: 数组; 说明: 包含所有步骤的列表; 限制: 每个步骤都是一个对象，且顺序重要。
2. step_id 类型: 整数; 说明: 步骤的唯一标识符，用于区分不同的步骤; 限制: 必须唯一，不能重复。
3. description 类型: 字符串; 说明: 对步骤的简要描述，说明步骤的目的或内容。限制: 描述应清晰简洁，避免冗长。
4. actions 类型: 数组; 说明: 包含该步骤中所有动作的列表。限制: 每个动作都是一个对象，且顺序可能影响执行顺序。
5. action_id 类型: 字符串; 说明: 动作的唯一标识符，用于区分不同的动作。限制: 在同一步骤内必须唯一。
6. action_type 类型: 字符串; 说明: 动作的类型，例如 "add_material", "ultrasonicate", "centrifuge"。限制: 必须是预定义的类型之一。
7. materials 类型: 数组; 说明: 使用的材料列表，每个材料包含名称、数量和单位。限制: 每个材料对象必须包含 "name", "amount", 和 "unit" 字段。
8. containers 类型: 数组; 说明: 使用的容器列表，每个容器包含类型、容量、单位和附加参数。限制: 每个容器对象必须包含 "type", "capacity", 和 "unit" 字段，"additional_parameters" 为可选。
9. equipment 类型: 数组; 说明: 使用的设备列表，每个设备包含名称和参数。限制: 每个设备对象必须包含 "name" 字段，"parameters" 为可选，根据设备需要填充。
10. output 类型: 字符串; 说明: 动作的输出标识符，用于后续步骤的输入。限制: 标识符应唯一且有意义。
11. dependencies 类型: 数组; 说明: 依赖的前一步骤的 "step_id" 列表。限制: 每个依赖项必须是有效的 "step_id"。
12. step_output 类型: 字符串; 说明: 步骤的输出标识符，用于后续步骤的输入。限制: 标识符应唯一且有意义。
### JSON结构主要字段说明
1. task_id 类型: 字符串 说明: 任务的唯一标识符，用于区分不同的任务 限制: 必须唯一，不能重复
2. materials 类型: 数组 说明: 使用的材料列表，每个材料包含ID、名称、数量和单位
3. containers 类型: 数组 说明: 使用的容器列表，每个容器包含ID、类型、容量、单位和可选的附加参数
限制: a.数组中的每个对象必须包含以下字段：a.1 "type": 容器类型（如烧杯、锥形瓶、离心管等） a.2 "capacity": 容器的容量 a.3 "unit": 容量的单位 a.4 "additional_parameters" 为可选字段，可包含材质、耐温范围等信息
4. equipments 类型: 数组 说明: 使用的设备列表，每个设备包含ID、名称和可选的参数 限制: 数组中的每个对象必须包含 "name" 字段 "parameters" 为可选，可根据设备实际需求进行配置（如搅拌速度、超声功率、温度范围等）
5. workflow 类型: 数组 说明: 包含所有步骤的列表 限制:每个步骤都是一个对象 顺序重要（一般按步骤顺序依次执行）
6. step_id 类型: 整数 说明: 步骤的唯一标识符，用于区分不同的步骤 限制: 必须唯一，不能重复
7. actions 类型: 数组 说明: 包含该步骤中所有动作的列表 限制: a.每个动作都是一个对象 b.动作在数组中的顺序通常会影响执行顺序
8. action_id 类型: 字符串 说明: 动作的唯一标识符，用于区分同一步骤内的不同动作 限制: 在同一步骤内必须唯一
9. action_type 类型: 字符串 说明: 动作的类型，但此处特别强调仅限于机械臂可执行的动作 限制:
a.必须是以下预定义类型之一（对应机械臂操作）：
a.1 "pick_container" （拿容器）
a.2 "place_container" （放容器）
a.3 "pick_container_with_material" （拿容器接材料/把材料加到容器里时的动作）
a.4 "place_container_into_equipment" （将容器放进某设备）
a.5 "remove_container_from_equipment" （从设备中取出容器）
b. 诸如“搅拌”、“超声”、“离心”等动作不在此列，它们属于设备自身的潜在动作，不在机械臂的动作范围内
10. dependencies 类型: 数组 说明: 依赖的前一步骤的 step_id 列表 限制: 每个依赖项必须是有效的 step_id 当本步骤需要等待前面若干步骤完成后再执行时，可通过此字段进行控制
11. step_output 类型: 字符串 说明: 步骤的输出标识符，用于后续步骤的输入或引用 限制: 标识符应唯一且有意义 可用来表示该步骤总体产出或容器中的新溶液名等
"""
def send_instruction_to_robot_platform():
# def send_instruction_to_robot_platform():
"""从S3获取最新的json文件并返回其URL链接
Returns:
@@ -242,8 +278,8 @@ def upload_to_s3(json_data: str):
return url.replace("http://100.85.52.31:9000" or "", "https://s3-api.siat-mic.com")
except Exception as e:
print(e)
return e
# print(e)
return f"Error: {str(e)}, Request human/user intervention."
install_boto3()
# 去掉可能存在的 ```json 和 ``` 标记
@@ -251,21 +287,22 @@ def upload_to_s3(json_data: str):
try:
# 尝试解析清理后的JSON数据
data = json.loads(json_data_cleaned)
# 取得task id
task_id = data['task_id']
# print("解析后的JSON数据:", data)
with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_file:
try:
json.dump(data, temp_file, indent=4, ensure_ascii=False)
temp_file.flush() # 确保数据写入文件
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
file_name = f"robot_expriment_results_{timestamp}.json"
file_name = f"robotExprimentScheme_{task_id}.json"
url = handle_minio_upload(temp_file.name, file_name)
print(f"文件上传成功, 唯一链接为{url}, 请将该URL记下来并传递给机器人平台。")
return f"JSON Scheme has been uploaded to S3 storage. The unique URL is: {url}, please pass it to the robot platform."
except Exception as e:
print(f"写入临时文件或上传文件时出错: {e}")
raise # 重新抛出异常以便上层调用者处理
# print(f"写入临时文件或上传文件时出错: {e}")
return f"Error: {str(e)}, Request human/user intervention."
except json.JSONDecodeError as e:
print(f"JSON解析错误: {e}")
raise # 重新抛出异常以便上层调用者处理
# print(f"JSON解析错误: {e}")
return f"Error: {str(e)}, Request human/user intervention."
def get_latest_exp_log():
def get_uv_latest_file():
@@ -316,7 +353,74 @@ def get_latest_exp_log():
def default_func():
return "Approved. Proceed as planned!"
def sendScheme2RobotPlatform():
def generate_task_id():
import datetime
# 获取当前时间
now = datetime.datetime.now()
# 格式化时间为字符串
formatted_time = now.strftime("%Y%m%d%H%M%S")
# 生成任务ID
task_id = f"task_{formatted_time}"
return task_id
def sendScheme2RobotPlatform(task_id: str, scheme_url: str):
# 首先检查task_id是否和scheme_url是否匹配
if task_id not in scheme_url:
return {"status": "error", "message": "task_id and scheme_url do not match, Request human/user intervention."}
# 读取scheme_url的内容
import requests
try:
response = requests.get(scheme_url)
response.raise_for_status()
scheme_content = response.text
# 读取scheme_content的内容为JSON
import json
scheme_data = json.loads(scheme_content)
# print(scheme_data)
except requests.exceptions.RequestException as e:
return {"status": "error", "message": f"Error reading scheme_url: {e}"}
def mol2mg(formula: str, source_unit: str, target_unit: str, value: float):
"""
将mol转换为mg
Args:
formula: 化学式，如CsPb
source_unit: 源单位 (mol或mmol)
target_unit: 目标单位 (mg)
value: 要转换的值
"""
import requests
from periodictable import formula as chem_formula
# 检查单位是否有效
if source_unit.lower() not in ['mol', 'mmol'] or target_unit.lower() != 'mg':
return {"status": "error", "message": "Invalid units. Only mol/mmol to mg conversion supported"}
try:
# 解析化学式并计算摩尔质量
compound = chem_formula(formula)
molar_mass = compound.mass # 获取化合物摩尔质量 (g/mol)
# 转换计算
if source_unit.lower() == 'mol':
mg_value = value * molar_mass * 1000 # mol -> g -> mg
elif source_unit.lower() == 'mmol':
mg_value = value * molar_mass # mmol -> g -> mg
return {
"status": "success",
"formula": formula,
"value": round(mg_value, 4), # 保留4位小数
"unit": "mg"
}
except ValueError as e:
return {"status": "error", "message": f"Invalid chemical formula: {formula}. Error: {str(e)}"}
import requests
url = "http://100.122.132.69:50000/sendScheme2RobotPlatform"
data = {"status": "ok"}
@@ -329,4 +433,4 @@ def sendScheme2RobotPlatform():
return None
if __name__ == "__main__":
print(sendScheme2RobotPlatform())
print(sendScheme2RobotPlatform())