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

后端添加cache；前端改动agents

Browse Source
This commit is contained in:
Yutang Li
2024-12-30 19:45:48 +08:00
parent 34e25633ba
commit 99301d7b3d
19 changed files with 90 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
backend/.coding/functions.py
View File

@@ -1,116 +1,61 @@
import os
import pandas
import glob
import requests
def get_max_uv_wavelength_from_txt(latest_file_path: str):
import pandas as pd
import os
# 文件检查
if not os.path.isfile(latest_file_path):
res = "ERROR: 指定的文件不存在"
return res
def retrieval_from_knowledge_base(
query: str,
topk: int
) -> str:
"""
Retrieval for knowledge from the knowledge base based on the specified query and returns the topk results.
Parameters:
query (str): The query for knowledge retrieval.
topk (int): The number of top results to return, default is 3.
Returns:
str: The result of the knowledge retrieval in JSON format.
"""
url = 'http://127.0.0.1:7080/v1/chat-messages'
headers = {
'Authorization': f'Bearer app-uJgo3TQKcS1O9PMCDHko71Fp',
'Content-Type': 'application/json'
}
data = {
"inputs": {"topK": topk},
"query": query,
"response_mode": "blocking",
"user": "tangger",
"files": []
}
# 打开并读取最新文件
with open(latest_file_path, 'r') as file:
lines = file.readlines()
response = requests.post(url, headers=headers, json=data)
# 找到数据开始的行号
data_start_index = -1
for i, line in enumerate(lines):
if "Wavelength Scan Data Record" in line:
data_start_index = i + 2 # 数据从该行的下两行开始
break
if response.status_code == 524:
print("Server is not responding. Please try again later. Maybe GPU was down in the container.")
return None
if data_start_index == -1:
res = "ERROR: 无法找到数据记录部分"
return res
try:
result = response.json()
except ValueError:
return [{"error": "Response is not in JSON format"}]
# 解析数据并构建表格
data = []
for line in lines[data_start_index:]:
parts = line.split()
if len(parts) == 7: # 保证每行有7列数据
no, wavelength, abs_value, trans, energy, energy_100, energy_0 = parts
try:
data.append({
'No': int(no),
'Wavelength(nm)': float(wavelength),
'Abs': float(abs_value),
'Trans(%T)': float(trans),
'Energy': float(energy),
'Energy(100%T)': float(energy_100),
'Energy(0%T)': float(energy_0)
})
except ValueError:
print(f"跳过无法解析的行: {line}")
if not data:
res = "ERROR: 未解析到任何有效数据"
return res
# 构建DataFrame
df = pd.DataFrame(data)
# 找到Abs值最大的行
max_abs_row = df.loc[df['Abs'].idxmax()]
# 获取最大Abs值对应的波长
max_abs_wavelength = max_abs_row['Wavelength(nm)']
res = f"本次实验的UV波长为: {max_abs_wavelength} nm"
print(res)
return res
def get_max_pl_peak_from_txt(latest_file_path: str):
import pandas as pd
import os
# 文件检查
if not os.path.isfile(latest_file_path):
res = "ERROR: 指定的文件不存在"
return res
# 打开并读取最新文件
with open(latest_file_path, 'r') as file:
lines = file.readlines()
# 找到 'Data Points' 开始的行号
data_start_index = -1
for i, line in enumerate(lines):
if "Data Points" in line:
data_start_index = i + 1 # 数据从该行的下一行开始
break
if data_start_index == -1:
res = "ERROR: 无法找到数据记录部分"
return res
# 解析nm和Data数据
data = []
for line in lines[data_start_index:]:
parts = line.split()
if len(parts) == 2: # 每行应该有2列数据，nm 和 Data
try:
nm = float(parts[0])
data_value = float(parts[1])
data.append({'nm': nm, 'Data': data_value})
except ValueError:
print(f"跳过无法解析的行: {line}")
if not data:
res = "ERROR: 未解析到任何有效数据"
return res
# 构建DataFrame
df = pd.DataFrame(data)
# 找到Data值最大的行
max_data_row = df.loc[df['Data'].idxmax()]
# 获取最大Data值对应的nm
max_data_nm = max_data_row['nm']
res = f"本次实验的PL峰位为: {max_data_nm} nm"
print(res)
return res
useful_results = []
try:
answer = eval(result.get("answer", "[]"))
for item in answer:
metadata = item.get("metadata", {})
useful_info = {
"id": metadata.get("document_id"),
"title": item.get("title"),
"content": item.get("content"),
"metadata": None,
"embedding": None,
"score": metadata.get("score")
}
useful_results.append(useful_info)
except Exception as e:
return [{"error": f"Error processing result: {e}", "status": "TERMINATE"}]
if useful_results == []:
useful_results = "NULL"
return str(useful_results)

18
backend/.coding/tmp_code_13e7e54147f77b0bfaecbf7c920ef7dd.py
View File

@@ -1,18 +0,0 @@
import urllib.parse
import requests
def web_searcher(query: str):
from urllib.parse import quote
encoded_query = quote(query)
url = 'https://s.jina.ai/' + encoded_query
headers = {
'Authorization': 'Bearer jina_8fc99db105ed48d7ab6a76ecd117e8e9GkNoWHY3tbmIMYSwQ6GAgzmr6c7z',
'X-Return-Format': 'markdown'
}
response = requests.get(url, headers=headers)
# print(response.text)
return str(response.text)

66
backend/.coding/tmp_code_2ad9eda61379dc378928cf732508c890.py
View File

@@ -1,66 +0,0 @@
import requests
def retrieval_from_knowledge_base(
query: str,
topk: int
) -> str:
"""
Retrieval for knowledge from the knowledge base based on the specified query and returns the topk results.
Parameters:
query (str): The query for knowledge retrieval.
topk (int): The number of top results to return, default is 3.
Returns:
str: The result of the knowledge retrieval in JSON format.
"""
# url = 'https://dify.siat-mic.com/v1/chat-messages'
# headers = {
# 'Authorization': f'Bearer app-uJgo3TQKcS1O9PMCDHko71Fp',
# 'Content-Type': 'application/json'
# }
url = f'{DIFY_BASE_URL}/chat-messages'
headers = {
'Authorization': f'{DIFY_API_KEY}',
'Content-Type': 'application/json'
}
data = {
"inputs": {"topK": topk},
"query": query,
"response_mode": "blocking",
"user": "tangger",
"files": []
}
response = requests.post(url, headers=headers, json=data)
if response.status_code == 524:
print("Server is not responding. Please try again later. Maybe GPU was down in the container.")
return None
try:
result = response.json()
except ValueError:
return [{"error": "Response is not in JSON format"}]
useful_results = []
try:
answer = eval(result.get("answer", "[]"))
for item in answer:
metadata = item.get("metadata", {})
useful_info = {
"id": metadata.get("document_id"),
"title": item.get("title"),
"content": item.get("content"),
"metadata": None,
"embedding": None,
"score": metadata.get("score")
}
useful_results.append(useful_info)
except Exception as e:
return [{"error": f"Error processing result: {e}", "status": "TERMINATE"}]
if useful_results == []:
useful_results = "NULL"
return str(useful_results)

61
backend/.coding/tmp_code_2c3931dbbd9beafb9ec2777cebf789fe.py
View File

@@ -1,61 +0,0 @@
import requests
def retrieval_from_knowledge_base(
query: str,
topk: int
) -> str:
"""
Retrieval for knowledge from the knowledge base based on the specified query and returns the topk results.
Parameters:
query (str): The query for knowledge retrieval.
topk (int): The number of top results to return, default is 3.
Returns:
str: The result of the knowledge retrieval in JSON format.
"""
url = 'https://dify.siat-mic.com/v1/chat-messages'
headers = {
'Authorization': f'Bearer app-uJgo3TQKcS1O9PMCDHko71Fp',
'Content-Type': 'application/json'
}
data = {
"inputs": {"topK": topk},
"query": query,
"response_mode": "blocking",
"user": "tangger",
"files": []
}
response = requests.post(url, headers=headers, json=data)
if response.status_code == 524:
print("Server is not responding. Please try again later. Maybe GPU was down in the container.")
return None
try:
result = response.json()
except ValueError:
return [{"error": "Response is not in JSON format"}]
useful_results = []
try:
answer = eval(result.get("answer", "[]"))
for item in answer:
metadata = item.get("metadata", {})
useful_info = {
"id": metadata.get("document_id"),
"title": item.get("title"),
"content": item.get("content"),
"metadata": None,
"embedding": None,
"score": metadata.get("score")
}
useful_results.append(useful_info)
except Exception as e:
return [{"error": f"Error processing result: {e}", "status": "TERMINATE"}]
if useful_results == []:
useful_results = "NULL"
return str(useful_results)

18
backend/.coding/tmp_code_3890c47cfc390b061ca90ccfcabe61d1.py
View File

@@ -1,18 +0,0 @@
import requests
import urllib.parse
def web_searcher(query: str):
from urllib.parse import quote
encoded_query = quote(query)
url = 'https://s.jina.ai/' + encoded_query
headers = {
'Authorization': 'Bearer jina_8fc99db105ed48d7ab6a76ecd117e8e9GkNoWHY3tbmIMYSwQ6GAgzmr6c7z',
'X-Return-Format': 'markdown'
}
response = requests.get(url, headers=headers)
# print(response.text)
return str(response.text)

14
backend/.coding/tmp_code_54d6b7617799bd780ee3d5e3e5edd994.py
View File

@@ -1,14 +0,0 @@
from functions import retrieval_from_knowledge_base
# Define the query for the task
query = "如何在常温条件下制备CsPbBr3纳米立方体"
# Specify the number of relevant top results to return
top_k_results = 3
# Retrieve information from the knowledge base
results = retrieval_from_knowledge_base(query, top_k_results)
# Display the retrieved results
print("Relevant Information:")
print(results)

28
backend/.coding/tmp_code_ba7b17ddef8656ae723efc3dbaf752c1.py
View File

@@ -1,28 +0,0 @@
from functions import retrieval_from_knowledge_base
def retrieve_nanocube_preparation(query, topk=3):
"""
Retrieves information on the preparation of CsPbBr3 nanocubes at room temperature.
Parameters:
query (str): The query string about the preparation process.
topk (int): The number of top results to return.
Returns:
str: The relevant information in JSON format.
"""
try:
# Execute retrieval function with the query and topk value
result = retrieval_from_knowledge_base(query, topk)
return result
except Exception as e:
return f"An error occurred while retrieving information: {str(e)}"
# Define the query
query = "Room-temperature synthesis of CsPbBr3 nanocubes"
# Perform retrieval
retrieval_result = retrieve_nanocube_preparation(query)
# Print the result
print(retrieval_result)