更新最新代码
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+114
-46
@@ -1,21 +1,39 @@
|
||||
# src/code_executor.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from langchain_core.output_parsers import StrOutputParser
|
||||
from langchain_experimental.utilities import PythonREPL
|
||||
from langchain_core.tools import Tool
|
||||
from langchain_experimental.tools import PythonREPLTool
|
||||
from project import ProjectBuilder, ProjectToolkit
|
||||
from src.project import ProjectBuilder, ProjectToolkit
|
||||
import sys
|
||||
import io
|
||||
import traceback
|
||||
|
||||
logger = logging.getLogger("BoweiAgent.CodeExecutor")
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
|
||||
class CodeExecutor:
|
||||
def __init__(self, prompts, llm_client, max_retries=3):
|
||||
def __init__(self, prompts, llm_client, max_retries):
|
||||
self.llm_client = llm_client
|
||||
self.prompts = prompts
|
||||
self.max_retries = max_retries
|
||||
self.max_retries = max_retries if max_retries >= 1 else 1
|
||||
self.output_parser = StrOutputParser()
|
||||
|
||||
def generate_code(self, user_request: str, context: str = "", bowei_api_docs: str = "") -> str:
|
||||
@@ -27,7 +45,13 @@ class CodeExecutor:
|
||||
response = self.llm_client.invoke(prompt.to_messages())
|
||||
code = self.output_parser.parse(response)
|
||||
logger.debug(f"生成的代码内容：\n{code}")
|
||||
return code
|
||||
|
||||
return {
|
||||
"code": 20000,
|
||||
"message": 'ok',
|
||||
"status": True,
|
||||
"data": code.content
|
||||
}
|
||||
|
||||
def fix_code(self, code: str, error: str) -> str:
|
||||
logger.warning(f"代码执行出错，开始修复。错误信息：{error}")
|
||||
@@ -35,11 +59,19 @@ class CodeExecutor:
|
||||
response = self.llm_client.invoke(prompt.to_messages())
|
||||
fixed_code = self.output_parser.parse(response)
|
||||
logger.debug(f"修复后的代码内容：\n{fixed_code}")
|
||||
return fixed_code
|
||||
|
||||
def execute_code(self, code_str):
|
||||
return {
|
||||
"code": 20000,
|
||||
"message": 'ok',
|
||||
"status": True,
|
||||
"data": fixed_code.content
|
||||
}
|
||||
|
||||
def execute_code(self, code_str) -> dict:
|
||||
"""封装代码执行逻辑"""
|
||||
logger.debug(f"开始执行代码: {code_str}")
|
||||
logger.debug(f"开始执行代码:\n {code_str}")
|
||||
|
||||
old_stdout = None
|
||||
try:
|
||||
namespace = {
|
||||
"project": __import__("project"),
|
||||
@@ -56,65 +88,101 @@ class CodeExecutor:
|
||||
if "project_get_calculate_function" not in namespace:
|
||||
raise ValueError("代码中未定义project_get_calculate_function函数")
|
||||
|
||||
result_tuple = namespace["project_get_calculate_function"]()
|
||||
result_dict = namespace["project_get_calculate_function"]()
|
||||
|
||||
sys.stdout = old_stdout
|
||||
output = redirected_output.getvalue().strip()
|
||||
|
||||
if not isinstance(result_tuple, tuple) or len(result_tuple) != 4:
|
||||
raise ValueError("函数应返回包含4个元素的元组(status, data, error, helper_info)")
|
||||
if not isinstance(result_dict, dict) or len(result_dict) != 4:
|
||||
raise ValueError("函数应返回包含4个元素的字典(status, data, error, helper_info)")
|
||||
|
||||
status, data, error, helper_info = result_tuple
|
||||
|
||||
logger.info(f"执行结果: status={status}, data={data}, error={error}")
|
||||
|
||||
return {
|
||||
"status": status,
|
||||
"data": data,
|
||||
"error": error,
|
||||
"helper_info": helper_info,
|
||||
"output": output,
|
||||
}
|
||||
logger.debug(f"执行结果: {result_dict}")
|
||||
|
||||
return result_dict
|
||||
except Exception as e:
|
||||
# 确保恢复stdout
|
||||
sys.stdout = old_stdout
|
||||
logger.error(f"执行代码时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
return {
|
||||
"status": "error",
|
||||
"error": str(e),
|
||||
"helper_info": [],
|
||||
"traceback": traceback.format_exc(),
|
||||
"code": 50000,
|
||||
"message": str(e),
|
||||
"status": False,
|
||||
"data": []
|
||||
}
|
||||
|
||||
def generate_and_run_code(self, user_request: str, context: str = "", bowei_api_docs: str = "") -> str:
|
||||
code = self.generate_code(user_request, context, bowei_api_docs)
|
||||
logger.info("开始执行生成的代码")
|
||||
|
||||
def generate_and_run_code(self, user_request: str, context: str = "", bowei_api_docs: str = "") -> dict:
|
||||
code = ""
|
||||
pre_code = ""
|
||||
error_msg = ""
|
||||
prev_happend_error = False
|
||||
|
||||
result = self.generate_code(user_request, context, bowei_api_docs)
|
||||
if isinstance(result, dict) and result.get('status', False):
|
||||
code = result['data']
|
||||
elif isinstance(result, dict):
|
||||
error_msg = result.get("message", "返回结果中没有错误信息")
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"code": 50000,
|
||||
"message": error_msg,
|
||||
"status": False,
|
||||
"data": code
|
||||
}
|
||||
else:
|
||||
error_msg = "返回结果格式未知"
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"code": 50000,
|
||||
"message": error_msg,
|
||||
"status": False,
|
||||
"data": code
|
||||
}
|
||||
|
||||
logger.info("开始执行生成的代码")
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
if prev_happend_error:
|
||||
logger.error(f"代码执行失败，尝试第 {attempt+1} 次修复。错误信息：{error_msg}")
|
||||
code = self.fix_code(pre_code, error_msg)
|
||||
|
||||
import re
|
||||
|
||||
pre_code = re.sub(r"^```python\s*|\s*```$", "", code.content, flags=re.MULTILINE)
|
||||
result = self.execute_code(pre_code)
|
||||
if result["status"] == "success":
|
||||
logger.info(f"代码执行成功，结果： {result['data']}")
|
||||
return result["data"]
|
||||
else:
|
||||
error_msg = result.get("error", "未知错误")
|
||||
prev_happend_error = True
|
||||
result = self.fix_code(pre_code, error_msg)
|
||||
if isinstance(result, dict) and result.get('status', False):
|
||||
code = result['data']
|
||||
elif isinstance(result, dict):
|
||||
error_msg = result.get("message", "返回结果中没有错误信息")
|
||||
raise ValueError(error_msg)
|
||||
else:
|
||||
error_msg = "返回结果格式未知"
|
||||
raise ValueError(error_msg)
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
error_msg = f"生成修复代码时发生异常: {str(e)}"
|
||||
prev_happend_error = True
|
||||
|
||||
logger.error(f"代码执行失败，超过最大重试次数 {self.max_retries}")
|
||||
return f"代码执行失败，超过最大重试次数 {self.max_retries}。\n最后一次错误信息：\n{error_msg}"
|
||||
try:
|
||||
import re
|
||||
pre_code = re.sub(r"^```python\s*|\s*```$", "", code, flags=re.MULTILINE)
|
||||
except Exception as e:
|
||||
error_msg = f"解析生成代码格式时发生异常: {str(e)}"
|
||||
prev_happend_error = True
|
||||
|
||||
try:
|
||||
result = self.execute_code(pre_code)
|
||||
if isinstance(result, dict) and result.get('status', False):
|
||||
logger.info(f"代码执行成功，结果： {result['data']}")
|
||||
return result
|
||||
elif isinstance(result, dict):
|
||||
error_msg = result.get("message", "返回结果中没有错误信息")
|
||||
else:
|
||||
error_msg = "返回结果格式未知"
|
||||
prev_happend_error = True
|
||||
except Exception as e:
|
||||
error_msg = f"执行代码时发生异常: {str(e)}"
|
||||
prev_happend_error = True
|
||||
|
||||
error_msg = f"代码执行失败，最后一次错误信息： {error_msg}"
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"code": 50000,
|
||||
"message": error_msg,
|
||||
"status": False,
|
||||
"data": []
|
||||
}
|
||||
|
||||
@@ -1,4 +1,24 @@
|
||||
# src/config.py
|
||||
|
||||
import yaml
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class Config:
|
||||
def __init__(self, path="config.yaml"):
|
||||
@@ -35,3 +55,7 @@ class Config:
|
||||
@property
|
||||
def langsmith(self):
|
||||
return self._config.get("langsmith", {})
|
||||
|
||||
@property
|
||||
def max_retries(self):
|
||||
return self._config.get("max_retries", 3)
|
||||
|
||||
+53
-1
@@ -1,10 +1,26 @@
|
||||
# src/dialog_manager.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from langchain.schema import SystemMessage, HumanMessage
|
||||
import asyncio
|
||||
|
||||
logger = logging.getLogger("BoweiAgent.DialogManager")
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class QuestionProcessor:
|
||||
def __init__(self, llm_client, business_structure, prompts):
|
||||
@@ -112,6 +128,18 @@ class DialogManager:
|
||||
rewritten_list.append((rewritten, doc.page_content))
|
||||
return rewritten_list
|
||||
|
||||
def understand_user_question(self, user_input: str):
|
||||
"""
|
||||
同步调用 understand_user_question_stream 方法
|
||||
:param user_input: 用户输入的问题
|
||||
:return: 改写后的问题列表
|
||||
"""
|
||||
try:
|
||||
return asyncio.run(self.understand_user_question_stream(user_input))
|
||||
except Exception as e:
|
||||
logger.error(f"执行 understand_user_question 时发生错误: {str(e)}")
|
||||
return []
|
||||
|
||||
async def run_async(self, pre_input: str = None):
|
||||
logger.info("启动对话管理器，等待用户输入")
|
||||
print("欢迎使用博微造价工程数据访问系统，输入 exit 退出。")
|
||||
@@ -174,3 +202,27 @@ class DialogManager:
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
def generated_code(self, selected_rewritten, selected_knowledge):
|
||||
"""
|
||||
生成代码
|
||||
:param selected_rewritten: 选中的重写后的请求
|
||||
:param selected_knowledge: 选中的知识
|
||||
:return: 生成的代码
|
||||
"""
|
||||
result = self.code_executor.generate_code(
|
||||
selected_rewritten,
|
||||
context=selected_knowledge,
|
||||
bowei_api_docs=self.bowei_api_docs
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
def execute_code(self, code):
|
||||
"""
|
||||
执行代码并返回结果
|
||||
:param code: 代码
|
||||
:return: 代码执行结果
|
||||
"""
|
||||
result = self.code_executor.execute_code(code)
|
||||
return result
|
||||
@@ -1,3 +1,24 @@
|
||||
# src/document_loader.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
def load_file(path: str) -> str:
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
|
||||
@@ -1,6 +1,25 @@
|
||||
# src/embedding_client.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from langchain_openai import OpenAIEmbeddings
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class EmbeddingClient:
|
||||
def __init__(self, embedding_config: dict):
|
||||
api_key = embedding_config.get("api_key")
|
||||
|
||||
+26
-1
@@ -1,8 +1,27 @@
|
||||
# src/llm_client.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
import os
|
||||
import getpass
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langchain_core.rate_limiters import InMemoryRateLimiter
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class LLMClient:
|
||||
def __init__(self, openai_config: dict):
|
||||
@@ -22,8 +41,14 @@ class LLMClient:
|
||||
|
||||
model_name = openai_config.get("model_name", "gpt-4o-mini")
|
||||
|
||||
# 初始化限速器：每秒最多0.1次请求（即每10秒1次）
|
||||
rate_limiter = InMemoryRateLimiter(
|
||||
requests_per_second=13,
|
||||
check_every_n_seconds=10,
|
||||
max_bucket_size=10,
|
||||
)
|
||||
# 开启流式
|
||||
self.llm = ChatOpenAI(model_name=model_name, temperature=0, streaming=True)
|
||||
self.llm = ChatOpenAI(model_name=model_name, temperature=0, streaming=True, rate_limiter=rate_limiter)
|
||||
|
||||
def invoke(self, messages):
|
||||
# 同步调用，返回完整响应
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
# src/multi_llm_client.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
import os
|
||||
import getpass
|
||||
import itertools
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langchain_core.rate_limiters import InMemoryRateLimiter
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class MultiAPIKeyChatOpenAI:
|
||||
def __init__(self, openai_config: dict):
|
||||
api_keys = openai_config.get("api_keys", [])
|
||||
if not api_keys:
|
||||
api_key = openai_config.get("api_key")
|
||||
if not api_key:
|
||||
api_key = os.environ.get("OPENAI_API_KEY")
|
||||
api_keys = [api_key]
|
||||
|
||||
api_base = openai_config.get("api_base")
|
||||
api_type = openai_config.get("api_type", "openai")
|
||||
model_name = openai_config.get("model_name", "gpt-4o-mini")
|
||||
|
||||
# 设置环境变量（可选，根据需要）
|
||||
if api_base:
|
||||
os.environ["OPENAI_API_BASE"] = api_base
|
||||
os.environ["OPENAI_API_TYPE"] = api_type
|
||||
|
||||
# 初始化限速器
|
||||
rate_limiter = InMemoryRateLimiter(
|
||||
requests_per_second=13,
|
||||
check_every_n_seconds=10,
|
||||
max_bucket_size=10,
|
||||
)
|
||||
|
||||
# 创建多个 ChatOpenAI 实例，每个使用不同的 API Key
|
||||
self.llms = [
|
||||
ChatOpenAI(
|
||||
api_key=key,
|
||||
model_name=model_name,
|
||||
temperature=0,
|
||||
streaming=True,
|
||||
rate_limiter=rate_limiter,
|
||||
)
|
||||
for key in api_keys
|
||||
]
|
||||
|
||||
# 轮询器，用于循环调用不同的 llm 实例
|
||||
self._llm_cycle = itertools.cycle(self.llms)
|
||||
|
||||
def invoke(self, messages):
|
||||
llm = next(self._llm_cycle)
|
||||
return llm.invoke(messages)
|
||||
|
||||
def stream(self, messages):
|
||||
llm = next(self._llm_cycle)
|
||||
return llm.astream(messages)
|
||||
@@ -1,11 +1,26 @@
|
||||
# src/neo4j_raw_retriever.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import List
|
||||
from langchain.schema import Document
|
||||
from neo4j import GraphDatabase
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger("BoweiAgent.Neo4jRawRetriever")
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class Neo4jRawRetriever:
|
||||
def __init__(self, neo4j_conf: dict):
|
||||
|
||||
@@ -1,6 +1,26 @@
|
||||
# src/neo4j_retriever.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from langchain_neo4j import Neo4jVector
|
||||
from langchain_openai import OpenAIEmbeddings
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class Neo4jKnowledgeRetriever:
|
||||
def __init__(self, neo4j_conf: dict, embedding_client):
|
||||
neo4j_uri = neo4j_conf.get("uri")
|
||||
|
||||
+728
@@ -0,0 +1,728 @@
|
||||
"""
|
||||
软件知识图谱类定义
|
||||
根据Ontology_Layer.txt文件中的知识图谱信息创建
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
import json
|
||||
|
||||
|
||||
class ProjectToolkit(ABC):
|
||||
"""
|
||||
项目类(抽象基类)
|
||||
描述: 代表整个项目结构的顶层容器
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.project_division_set = ProjectDivisionItem() # 项目划分集对象
|
||||
|
||||
# 项目划分查询方法
|
||||
|
||||
@abstractmethod
|
||||
def get_division_by_name(self, name):
|
||||
"""
|
||||
通过名称获取项目划分对象
|
||||
|
||||
Args:
|
||||
name_part (str): 项目划分节点名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (List[Dict[str, Any]]): 成功时返回的数据列表，失败时为空列表
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_division_item_by_path(self, path):
|
||||
"""
|
||||
通过路径获取项目划分对象
|
||||
|
||||
Args:
|
||||
path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Dict[str, Any]): 成功时返回的项目划分对象数据，失败时为空字典
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_division_node_by_parent_and_name(self, parent_path, partial_name):
|
||||
"""
|
||||
通过父节点路径和模糊节点名称获取项目划分对象，包括子节点
|
||||
|
||||
Args:
|
||||
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
||||
partial_name (str): 目标节点的模糊或不完整名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (List[Dict[str, Any]]): 成功时返回的数据列表，失败时为空列表
|
||||
"""
|
||||
pass
|
||||
|
||||
# 工程量查询方法
|
||||
@abstractmethod
|
||||
def get_quantities_by_paths(self, paths_str):
|
||||
"""
|
||||
获取指定项目路径下的工程量对象
|
||||
|
||||
Args:
|
||||
paths_str (str): 以'/'分隔的多级节点路径
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Dict[str, Any]): 成功时返回的节点数据字典，失败时为空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_quantities_node_by_parent_and_code(self, parent_path, quantity_type=None, code=None):
|
||||
"""
|
||||
通过父节点路径和编码获取工程量对象（定额、主材或设备），包括子节点
|
||||
|
||||
Args:
|
||||
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
||||
quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None表示所有类型)
|
||||
code (str): 工程量编码，以'/'分隔的多个编码
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[List[Dict[str, Any]], Dict]): 成功时返回的数据列表或字典，失败时为空列表 []
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_quantities_node_by_parent_and_name(self, parent_path, partial_name, quantity_type=None):
|
||||
"""
|
||||
通过父节点路径、模糊节点名称和类型获取工程量对象（主材或者设备），包括子节点
|
||||
|
||||
Args:
|
||||
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
||||
partial_name (str): 目标节点的模糊或不完整名称
|
||||
quantity_type (str): 工程量类型('定额'、'主材'、'设备')
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[List[Dict[str, Any]], Dict]): 成功时返回的数据列表，失败时为空列表 []
|
||||
"""
|
||||
pass
|
||||
|
||||
# 材机查询方法
|
||||
@abstractmethod
|
||||
def get_material_equipment_by_path(self, paths_str):
|
||||
"""
|
||||
通过路径获取材机对象（MaterialOrEquipment 类型节点）
|
||||
|
||||
Args:
|
||||
path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List]): 成功时返回的节点属性数据字典，失败时为空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_material_equipment_by_parent_and_code(self, parent_path, code):
|
||||
"""
|
||||
通过父节点路径和编码获取材机对象
|
||||
|
||||
Args:
|
||||
parent_path (str): 父节点的部分路径，以'/'分隔的多级节点路径
|
||||
code (str): 目标节点的编码属性值
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List]): 成功时返回的节点属性数据字典，失败时为空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 取费查询方法
|
||||
@abstractmethod
|
||||
def get_fee_template_by_path(self, paths_str):
|
||||
"""
|
||||
通过路径获取取费表模板节点
|
||||
|
||||
Args:
|
||||
path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
|
||||
例如：工程数据/取费表模板/余物清理/线路取费表（余物清理）/直接费
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List]): 成功时返回的节点属性数据字典，失败时为空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_fee_template_by_parent_and_name(self, parent_path, partial_name):
|
||||
"""
|
||||
通过父节点路径和节点名称获取取费表模板节点
|
||||
|
||||
Args:
|
||||
parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
|
||||
例如：工程数据/取费表模板/余物清理/线路取费表（余物清理）
|
||||
node_name (str): 目标节点的名称，例如：直接费
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List]): 成功时返回的节点属性数据字典，失败时为空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 费用表查询方法
|
||||
@abstractmethod
|
||||
def get_fee_schedule_on_auxiliary_expense_table(self, table_name, fee_name, fee: str):
|
||||
"""
|
||||
在辅助费用表中查找费用
|
||||
|
||||
Args:
|
||||
table_name (str): 费用表名称
|
||||
fee_name (str): 要查找的费用名称（可能在多级子节点中）
|
||||
fee_attribute (str): 费用值属性名
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性值（可能是 str/int/float 等），失败时返回 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_fee_schedule_on_other_expense_table(self, table_name, fee_name, fee):
|
||||
"""
|
||||
在其它费用表中查找费用
|
||||
|
||||
Args:
|
||||
table_name (str): 费用表名称
|
||||
fee_name (str): 要查找的费用名称（可能在多级子节点中）
|
||||
fee_attribute (str): 费用值属性名
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性值（可能是 str/int/float 等），失败时返回 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_fee_schedule_on_land_acquisition_fee_table_table(self, table_name, fee_name, fee):
|
||||
"""
|
||||
在其中：场地征用费用表中查找费用
|
||||
|
||||
Args:
|
||||
table_name (str): 费用表名称
|
||||
fee_name (str): 要查找的费用名称（可能在多级子节点中）
|
||||
fee_attribute (str): 费用值属性名
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性值（可能是 str/int/float 等），失败时返回 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_fee_schedule_on_installation_price_difference_table(self, table_name, fee_name, fee):
|
||||
"""
|
||||
在安装价差费用表中查找费用
|
||||
|
||||
Args:
|
||||
table_name (str): 费用表名称
|
||||
fee_name (str): 要查找的费用名称（可能在多级子节点中）
|
||||
fee_attribute (str): 费用值属性名
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性值（可能是 str/int/float 等），失败时返回 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_fee_schedule_on_Engineering_Cost_table(self, table_name, fee_name, fee):
|
||||
"""
|
||||
在工程费用表中查找费用
|
||||
|
||||
Args:
|
||||
table_name (str): 费用表名称
|
||||
fee_name (str): 要查找的费用名称（可能在多级子节点中）
|
||||
fee_attribute (str): 费用值属性名
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性值（可能是 str/int/float 等），失败时返回 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 工程属性查询方法
|
||||
@abstractmethod
|
||||
def get_project_property(self, property_name):
|
||||
"""
|
||||
通过属性名称获取工程属性
|
||||
|
||||
Args:
|
||||
property_name (str): 属性名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 项目划分查找取费表
|
||||
@abstractmethod
|
||||
def get_fee_table_by_project_division(self, project_division_path, fee_name):
|
||||
"""
|
||||
通过项目划分路径和取费名称查找费用
|
||||
|
||||
Args:
|
||||
project_division_path (str): 项目划分节点的路径
|
||||
fee_name (str): 取费名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 清单查找取费表
|
||||
@abstractmethod
|
||||
def get_fee_table_by_list(self, parent_path, list_code, list_unit, fee_name):
|
||||
"""
|
||||
通过清单名称和取费名称查找费用
|
||||
|
||||
Args:
|
||||
parent_path (str): 父级路径
|
||||
list_code (str): 清单编号
|
||||
list_unit (str): 清单单位
|
||||
fee_name (str): 取费名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 定额节点查找合价
|
||||
@abstractmethod
|
||||
def get_fee_table_by_quoto_code(self, parent_path, quantity_type, code, fee_name):
|
||||
"""
|
||||
通过父级路径、工程量类型、定额编码和取费名称查找费用
|
||||
|
||||
Args:
|
||||
parent_path (str): 父级路径
|
||||
quantity_type (str): 工程量类型
|
||||
code (str): 定额编码
|
||||
fee_name (str): 取费名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 工程量节点查找合价
|
||||
@abstractmethod
|
||||
def get_fee_table_by_quantities_name(self, parent_path, quantity_type, quantity_name, fee_name):
|
||||
"""
|
||||
通过父级路径、工程量类型、工程量名称和取费名称查找费用
|
||||
|
||||
Args:
|
||||
parent_path (str): 父级路径
|
||||
quantity_type (str): 工程量类型
|
||||
quantity_name (str): 工程量名称
|
||||
fee_name (str): 取费名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 材机节点查找市场价
|
||||
@abstractmethod
|
||||
def get_fee_by_material_equipment_code(self, parent_path, material_equipment_code, fee_name):
|
||||
"""
|
||||
通过父级路径、材机编码和取费名称查找费用
|
||||
|
||||
Args:
|
||||
parent_path (str): 父级路径
|
||||
material_equipment_code (str): 材机编码
|
||||
fee_name (str): 取费名称
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
# 查找清单节点
|
||||
@abstractmethod
|
||||
def get_fee_table_by_list_name(self, parent_path, list_code, unit):
|
||||
"""
|
||||
通过父级路径、清单编号和单位查找清单
|
||||
|
||||
Args:
|
||||
parent_path (str): 父级路径
|
||||
list_code (str): 清单编号
|
||||
unit (str): 单位
|
||||
|
||||
Returns:
|
||||
dict: 返回字典，字段包括：
|
||||
- code (int): 状态码，固定为 200（成功）或 201（失败）
|
||||
- message (str): 成功时为 "Ok"，失败时包含错误信息和辅助信息
|
||||
- status (bool): 成功为 True，失败为 False
|
||||
- data (Union[Dict, List, Any]): 成功时返回属性数据字典，失败时返回空字典 {}
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class ProjectDivisionSet:
|
||||
"""
|
||||
项目划分项
|
||||
描述: 代表项目结构中的层级条目，具有自身的属性，并且可以包含ProjectDivisionTree（项目划分树）
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
|
||||
|
||||
class ProjectDivisionTree:
|
||||
"""
|
||||
项目划分项
|
||||
描述: 代表项目结构中的层级条目，具有自身的属性，并且可以包含ProjectDivisionItem（项目划分项）
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
|
||||
|
||||
class ProjectDivisionItem:
|
||||
"""
|
||||
项目划分项
|
||||
描述: 代表项目结构中的层级条目，具有自身的属性，并且可以包含子项目划分项或详细工作项
|
||||
JSON对应: ProjectDivisionSet数组中的对象，或ProjectDivisionItem的children数组中type为"项目划分"的对象
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.GUID = None # xsd:string
|
||||
self.id = None # xsd:string
|
||||
self.name = None # xsd:string
|
||||
self.代码 = None # xsd:string (可选)
|
||||
self.费率 = None # xsd:string
|
||||
self.单位 = None # xsd:string (可选)
|
||||
self.取费表id = None # xsd:string
|
||||
self.颜色标记 = None # xsd:string
|
||||
self.取费表 = None # xsd:string
|
||||
self.合价含税 = None # xsd:string (可选)
|
||||
self.type = None # xsd:string
|
||||
self.专业类型 = None # xsd:string
|
||||
self.资源库列表 = None # xsd:list (可选)
|
||||
self.notCheck = None # xsd:string (可选)
|
||||
|
||||
|
||||
class ProjectQuantity:
|
||||
"""
|
||||
工程量
|
||||
描述: 代表项目划分项(ProjectDivisionItem)下的具体工作单元或物料项，是定额、主材、设备的父类型
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.id = None # xsd:string
|
||||
self.类型 = None # xsd:string ("0"为定额,"1"为主材,"5"为设备)
|
||||
self.name = None # xsd:string
|
||||
self.编码 = None # xsd:string
|
||||
self.单位 = None # xsd:string
|
||||
self.数量 = None # xsd:string
|
||||
self.资源库名称 = None # xsd:string
|
||||
self.投标数量 = None # xsd:string (可选)
|
||||
self.投标单价 = None # xsd:string (可选)
|
||||
self.特征段 = None # xsd:string (可选)
|
||||
self.关联父级量 = None # xsd:string (可选)
|
||||
self.颜色标记 = None # xsd:string (可选)
|
||||
self.单价不含税 = None
|
||||
self.cost_set = None # xsd:CostSet
|
||||
|
||||
|
||||
class Ration(ProjectQuantity):
|
||||
"""
|
||||
定额
|
||||
描述: 代表一种标准的工程量条目，通常包含详细的工、料、机消耗标准
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.计算式 = None # xsd:string
|
||||
self.中标计算式 = None # xsd:string
|
||||
self.人工费 = None # xsd:string
|
||||
self.机械费 = None # xsd:string
|
||||
self.甲供材料费不含税 = None # xsd:string
|
||||
self.材料费 = None # xsd:string
|
||||
self.定额系数 = None # xsd:string
|
||||
self.人工系数 = None # xsd:string
|
||||
self.材料系数 = None # xsd:string
|
||||
self.机械系数 = None # xsd:string
|
||||
self.定额范围 = None # xsd:string
|
||||
self.定额章节名称 = None # xsd:string
|
||||
self.费用类型 = None # xsd:string
|
||||
self.甲供材料费含税 = None # xsd:string
|
||||
self.投标合价 = None # xsd:string
|
||||
self.其中甲供材料费 = None # xsd:string
|
||||
self.合价不含税 = None # xsd:string
|
||||
self.基价 = None # xsd:string
|
||||
self.所属定额库 = None # xsd:string
|
||||
|
||||
def __str__(self):
|
||||
"""返回定额的字符串表示"""
|
||||
attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
|
||||
return json.dumps(attrs, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
class Material(ProjectQuantity):
|
||||
"""
|
||||
主材
|
||||
描述: 代表工程中使用的主要材料
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.规格型号 = None # xsd:string
|
||||
self.损耗率 = None # xsd:string
|
||||
self.供货方 = None # xsd:string
|
||||
self.集中配送 = None # xsd:string (可选)
|
||||
self.单重 = None # xsd:string (可选)
|
||||
self.市场价不含税 = None # xsd:string
|
||||
self.市场价含税 = None # xsd:string
|
||||
self.单价含税 = None # xsd:string
|
||||
self.结算市场价不含税 = None # xsd:string (可选)
|
||||
self.结算市场价含税 = None # xsd:string (可选)
|
||||
self.基准价不含税 = None # xsd:string (可选)
|
||||
self.基准价含税 = None # xsd:string (可选)
|
||||
self.费用类型 = None # xsd:string
|
||||
self.增值税率 = None # xsd:string (可选)
|
||||
self.合价含税 = None # xsd:string
|
||||
self.合价不含税 = None # xsd:string
|
||||
self.线重 = None # xsd:string (可选)
|
||||
self.制造长度 = None # xsd:string (可选)
|
||||
self.截面积 = None # xsd:string (可选)
|
||||
|
||||
def __str__(self):
|
||||
"""返回材料的字符串表示"""
|
||||
attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
|
||||
return json.dumps(attrs, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
class Equipment(ProjectQuantity):
|
||||
"""
|
||||
设备
|
||||
描述: 代表工程中安装或使用的设备
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.规格型号 = None # xsd:string
|
||||
self.供货方 = None # xsd:string
|
||||
self.运杂费率 = None # xsd:string (可选)
|
||||
self.单价含税 = None # xsd:string
|
||||
self.设备类型 = None # xsd:string (可选)
|
||||
self.增值税率 = None # xsd:string (可选)
|
||||
self.合价含税 = None # xsd:string
|
||||
self.合价不含税 = None # xsd:string
|
||||
|
||||
|
||||
class MaterialOrEquipment:
|
||||
"""
|
||||
材机
|
||||
描述: 代表DetailedWorkItem中所列出的具体材料、人工或机械设备及其详细信息
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.id = None # xsd:string
|
||||
self.编码 = None # xsd:string
|
||||
self.名称 = None # xsd:string
|
||||
self.单位 = None # xsd:string
|
||||
self.类型 = None # xsd:string
|
||||
self.供货方 = None # xsd:string
|
||||
self.预算价不含税 = None # xsd:string
|
||||
self.市场价不含税 = None # xsd:string
|
||||
self.预算价含税 = None # xsd:string
|
||||
self.市场价含税 = None # xsd:string
|
||||
self.结算预算价不含税 = None # xsd:string
|
||||
self.结算市场价不含税 = None # xsd:string
|
||||
self.结算预算价含税 = None # xsd:string
|
||||
self.结算市场价含税 = None # xsd:string
|
||||
self.暂估价 = None # xsd:string
|
||||
self.拆分 = None # xsd:string
|
||||
self.全口径市场价不含税 = None # xsd:string
|
||||
self.全口径市场价含税 = None # xsd:string
|
||||
self.商品砼 = None # xsd:string
|
||||
self.数量 = None # xsd:string
|
||||
self.是否未计价 = None # xsd:string
|
||||
|
||||
|
||||
class FeeTableTemplateItem:
|
||||
"""
|
||||
取费表模板项
|
||||
描述:
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
self.OutlayID = None # xsd:string
|
||||
self.type = None # xsd:string
|
||||
self.profession = None # xsd:string
|
||||
|
||||
|
||||
class FeeCollection:
|
||||
"""
|
||||
取费
|
||||
描述:
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
self.serialNumber = None # xsd:string
|
||||
self.code = None # xsd:string
|
||||
self.base = None # xsd:string
|
||||
self.rate = None # xsd:string
|
||||
|
||||
|
||||
class FeeScheduleItem:
|
||||
"""
|
||||
费用表项
|
||||
描述:
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
|
||||
|
||||
class Fee:
|
||||
"""
|
||||
取费
|
||||
描述:
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.name = None # xsd:string
|
||||
self.serialNumber = None # xsd:string
|
||||
self.code = None # xsd:string
|
||||
self.rate = None # xsd:string
|
||||
self.amount = None # xsd:string
|
||||
self.输出 = None # xsd:string (可选)
|
||||
self.取费基数 = None # xsd:string (可选)
|
||||
self.编制依据 = None # xsd:string (可选)
|
||||
self.编码 = None # xsd:string (可选)
|
||||
self.备注 = None # xsd:string (可选)
|
||||
self.表一显示 = None # xsd:string (可选)
|
||||
self.报表输出 = None # xsd:string (可选)
|
||||
self.建安合计费 = None # xsd:string (可选)
|
||||
self.合计费 = None # xsd:string (可选)
|
||||
self.占总计 = None # xsd:string (可选)
|
||||
self.建筑费 = None # xsd:string (可选)
|
||||
self.安装费 = None # xsd:string (可选)
|
||||
self.设备费 = None # xsd:string (可选)
|
||||
self.其他费 = None # xsd:string (可选)
|
||||
self.施工费 = None # xsd:string (可选)
|
||||
self.单位投资 = None # xsd:string (可选)
|
||||
|
||||
class ProjectBuilder:
|
||||
# 存储注册的工具类
|
||||
_registry = None
|
||||
_config = {}
|
||||
|
||||
"""项目工具工厂类"""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def register(cls, toolkit_class: type, config: dict):
|
||||
"""
|
||||
注册工具类到工厂
|
||||
|
||||
参数:
|
||||
toolkit_class: 继承自ProjectToolkit的具体工具类
|
||||
"""
|
||||
if not issubclass(toolkit_class, ProjectToolkit):
|
||||
raise TypeError(f"{toolkit_class.__name__} 必须继承自 ProjectToolkit")
|
||||
|
||||
cls._config = config
|
||||
cls._registry = toolkit_class
|
||||
|
||||
@classmethod
|
||||
def build(cls) -> ProjectToolkit:
|
||||
"""
|
||||
创建工具实例
|
||||
|
||||
参数:
|
||||
|
||||
返回:
|
||||
实例化的工具对象
|
||||
"""
|
||||
if cls._registry is None:
|
||||
raise KeyError(f"未注册的类，请先注册类")
|
||||
|
||||
|
||||
return cls._registry(cls._config)
|
||||
File diff suppressed because it is too large
Load Diff
+197
-5
@@ -1,5 +1,8 @@
|
||||
# src/prompt_manager.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass
|
||||
from langchain.prompts import ChatPromptTemplate
|
||||
from langchain.schema import SystemMessage, HumanMessage
|
||||
@@ -60,14 +63,23 @@ class PromptManager:
|
||||
# 输出格式（必须严格遵循）
|
||||
def project_get_calculate_function():
|
||||
project = ProjectBuilder.build()
|
||||
status, data, error, helper_info = project.[SELECTED_METHOD]([PARAMETERS])
|
||||
return status, data, error, helper_info
|
||||
result_dict = project.[SELECTED_METHOD]([PARAMETERS])
|
||||
status = result_dict.get('status', False)
|
||||
message = result_dict.get('message', '')
|
||||
code = result_dict.get('data', '')
|
||||
data = result_dict.get('data', [])
|
||||
if status:
|
||||
return result_dict
|
||||
else:
|
||||
return result_dict
|
||||
|
||||
# 执行规则
|
||||
- 参数必须从用户问题或上下文信息中提取
|
||||
- 输出代码中必须以def project_get_calculate_function() -> Tuple[bool, Any, Optional[str], Dict[str, Any]]函数作为入口函数
|
||||
- 输出代码中必须以def project_get_calculate_function() -> dict函数作为入口函数
|
||||
- 必须确保生成的代码可以直接执行，代码要注意进行各类错误检查，出错采用抛出异常方式，说明详细信息
|
||||
- 禁止添加任何注释或解释
|
||||
- ProjectToolkit 类中涉及项目划分的函数已考虑在其及其子孙项目划分下查找，所以无需生成递归子项目划分的代码
|
||||
- 如果文本中包含范围编码格式则需要进行编码展开，如'YX2-1~7'展开为‘YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7’
|
||||
"""
|
||||
)
|
||||
|
||||
@@ -89,6 +101,7 @@ def project_get_calculate_function():
|
||||
4. 修复后的代码应该完整，可以直接执行，并且能够返回查询结果
|
||||
|
||||
注意：
|
||||
- 如果文本中包含范围编码格式则需要进行编码展开，如'YX2-1~7'展开为‘YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7’
|
||||
- 必须只输出最终的Python代码，不要添加任何解释、注释、推理过程或自然语言描述。
|
||||
- 不要以“以下是修正后的代码”、“修改如下”等语句开头。
|
||||
- 不要输出任何其他无关的内容。
|
||||
@@ -131,8 +144,187 @@ def project_get_calculate_function():
|
||||
{user_input}
|
||||
|
||||
改写规则：
|
||||
识别目标层级: 从用户输入中解析层级路径，例如字符串“安装/架空输电线路本体工程”映射为：根节点(name: '安装') -[:CHILD_OF]-> 子节点(name: '架空输电线路本体工程')。知识图谱节点标签包括ProjectDivisionItem、ProjectQuantity、Fee、FeeCollection等。
|
||||
识别目标对象：仅从业务结构中识别用户问题中需要获取的核心对象类型（如 ProjectDivisionItem、ProjectQuantity、Fee、FeeCollection 等）。对象类型必须精确映射到结构中的节点标签（例如，使用 ProjectDivisionItem 而非“项目划分项”）。
|
||||
识别目标层级: 从用户输入中解析层级路径，例如字符串“安装/架空输电线路本体工程”映射为：根节点(name: '安装') -[:CHILD_OF]-> 子节点(name: '架空输电线路本体工程')。
|
||||
识别目标对象：仅从业务结构中识别如下类型：ProjectDivisionItem、ProjectAttributeSet、FeeSchedule、FeeItem的核心对象。对象类型必须精确映射到结构中的节点标签（例如，使用 ProjectDivisionItem 而非“项目划分项”）。
|
||||
提取查询条件：从用户输入中解析关键条件（如名称、量、类型、值、层级等），条件应基于对象属性（如 name、quantity、type），注意区分查找子串和相等的区别。如果条件涉及层级路径（如路径中包含特定部分），需提取路径部分作为条件（例如，用户提到“架空输电线路”时，解析为路径匹配）。
|
||||
如果用户指定层级（如“叶节点”），需在条件中体现（例如，添加 WHERE item.isLeaf = true）。
|
||||
忽略任何计算、转换或后处理要求（如“乘以1000”），只关注获取原始数据对象或属性。
|
||||
构建Cypher查询：生成一个Cypher查询语句，格式为：
|
||||
|
||||
MATCH 子句：匹配目标对象节点，并应用条件。如果涉及层级路径，使用路径匹配（如 MATCH path = (item1:ProjectDivisionItem)-[:CHILD_OF*]->(item2:ProjectDivisionItem) WHERE item1.name = '安装' AND item2.name = '架空输电线路本体工程'）。变量仅用于目标节点和必要路径节点。
|
||||
WHERE 子句：包含提取的条件（使用变量或具体值）。
|
||||
RETURN 子句：必须返回对象（如 RETURN item），不能包含对象属性、函数（如乘法、SUM）。
|
||||
LIMIT 子句： 最多返回5条。
|
||||
使用业务术语在节点标签和属性中（例如，ProjectDivisionItem 而不是“项目”）。
|
||||
查询应简洁，只获取数据，不执行计算。
|
||||
输出格式：直接输出最终Cypher查询语句，不添加解释或额外文本。
|
||||
|
||||
**示例**:
|
||||
用户问题：查找一下名称中包含工程的项目划分
|
||||
Cypher查询语句：MATCH (item:ProjectDivisionItem)\nWHERE item.name CONTAINS '工程'\nRETURN item\nLIMIT 5
|
||||
""")
|
||||
|
||||
return CodeExecutorPrompts(
|
||||
understand_prompt=understand_prompt,
|
||||
code_gen_prompt=code_gen_prompt,
|
||||
code_fix_prompt=code_fix_prompt,
|
||||
rewrite_prompt_template=rewrite_prompt_template,
|
||||
cypher_conversion_prompt=cypher_conversion_prompt,
|
||||
)
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
@dataclass
|
||||
class CodeExecutorPrompts:
|
||||
understand_prompt: ChatPromptTemplate
|
||||
code_gen_prompt: ChatPromptTemplate
|
||||
code_fix_prompt: ChatPromptTemplate
|
||||
rewrite_prompt_template: ChatPromptTemplate
|
||||
cypher_conversion_prompt: ChatPromptTemplate # 新增Cypher转换提示模板
|
||||
|
||||
|
||||
class PromptManager:
|
||||
def __init__(self):
|
||||
self.prompts = self._init_prompts()
|
||||
|
||||
def _init_prompts(self) -> CodeExecutorPrompts:
|
||||
understand_prompt = ChatPromptTemplate.from_template(
|
||||
"""
|
||||
你是一名电力造价业务专家，请基于以下示意工程文件业务结构，将用户自然语言问题改写成专业查询语句：
|
||||
|
||||
**示意工程文件业务结构**：
|
||||
{business_structure}
|
||||
|
||||
**改写规则**：
|
||||
1. **定位目标对象**：仅从示意工程文件业务结构中识别核心对象（如 `ProjectDivisionTree`→项目划分树、`FeeScheduleItem`→费用表）。
|
||||
2. **提取条件**：从用户输入中解析关键条件（如名称、量、类型），用【】标注变量。
|
||||
3. **构建专业语句**：格式为：`在[目标对象]中查找【条件】的项。`
|
||||
- 使用业务术语（如“项目划分项”而非“项目”）。
|
||||
- 条件需明确属性（如【名称】、【量】、【类型】）。
|
||||
4. **精确映射结构**：若用户查询层级（如“叶节点”），需在条件中体现。
|
||||
|
||||
**用户输入**：{user_input}
|
||||
**改写输出**：（仅输出改写后的语句）
|
||||
"""
|
||||
)
|
||||
|
||||
code_gen_prompt = ChatPromptTemplate.from_template(
|
||||
"""
|
||||
你是一个专业的Python工程师。我会给你一个用户问题，你需要将其转换为对应的Python代码
|
||||
|
||||
用户问题:
|
||||
{user_request}
|
||||
|
||||
上下文信息:
|
||||
{context}
|
||||
|
||||
工程数据访问库:
|
||||
{bowei_api_docs}
|
||||
|
||||
# 工作流程
|
||||
1. 从用户问题中提取关键信息（节点路径、节点类型、节点名称等）
|
||||
2. 根据"用户问题"和"上下文信息"选择最匹配的"工程数据访问库"中的函数和对象属性
|
||||
3. 生成可直接执行的完全满足用户输入问题要求功能效果的Python函数代码
|
||||
|
||||
# 输出格式（必须严格遵循）
|
||||
def project_get_calculate_function():
|
||||
project = ProjectBuilder.build()
|
||||
result_dict = project.[SELECTED_METHOD]([PARAMETERS])
|
||||
status = result_dict.get('status', False)
|
||||
message = result_dict.get('message', '')
|
||||
code = result_dict.get('data', '')
|
||||
data = result_dict.get('data', [])
|
||||
if status:
|
||||
return result_dict
|
||||
else:
|
||||
return result_dict
|
||||
|
||||
# 执行规则
|
||||
- 参数必须从用户问题或上下文信息中提取
|
||||
- 输出代码中必须以def project_get_calculate_function() -> dict函数作为入口函数
|
||||
- 必须确保生成的代码可以直接执行，代码要注意进行各类错误检查，出错采用抛出异常方式，说明详细信息
|
||||
- 禁止添加任何注释或解释
|
||||
- ProjectToolkit 类中涉及项目划分的函数已考虑在其及其子孙项目划分下查找，所以无需生成递归子项目划分的代码
|
||||
- 如果文本中包含范围编码格式则需要进行编码展开，如'YX2-1~7'展开为‘YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7’
|
||||
"""
|
||||
)
|
||||
|
||||
code_fix_prompt = ChatPromptTemplate.from_template(
|
||||
"""
|
||||
|
||||
你是一个专业的Python工程师。我会给你一段错误python代码和错误信息，你需要帮我修复这段出错的代码
|
||||
|
||||
已执行代码:
|
||||
{code}
|
||||
|
||||
代码执行报错信息:
|
||||
{error}
|
||||
|
||||
你的任务是：
|
||||
1. 根据"已执行代码"和"代码执行报错信息"来对“已执行代码”和函数调用参数进行修改，修复执行错误
|
||||
2. 如果错误信息中是代码的逻辑出现错误，那么就需要对代码本身整体结构进行修改
|
||||
3. 如果是代码中参数出现问题了，那么就需要结合错误信息中的帮助信息（helper_info）来对代码总的参数进行修改
|
||||
4. 修复后的代码应该完整，可以直接执行，并且能够返回查询结果
|
||||
|
||||
注意：
|
||||
- 如果文本中包含范围编码格式则需要进行编码展开，如'YX2-1~7'展开为‘YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7’
|
||||
- 必须只输出最终的Python代码，不要添加任何解释、注释、推理过程或自然语言描述。
|
||||
- 不要以“以下是修正后的代码”、“修改如下”等语句开头。
|
||||
- 不要输出任何其他无关的内容。
|
||||
- 输出格式必须完全符合指定的函数模板。
|
||||
- 如果无法根据已有信息进行修改，请原样返回原始代码。
|
||||
- 禁止在代码前加上```python字样
|
||||
- 禁止在代码后加上```字样
|
||||
|
||||
请输出你修补后的代码：
|
||||
""")
|
||||
|
||||
rewrite_prompt_template = ChatPromptTemplate.from_template(
|
||||
"""
|
||||
您是一个AI查询改写助手。基于给定的原始查询和上下文知识，生成一个精确的改写查询。步骤：
|
||||
1. 从上下文知识的`labels`提取对象类型，翻译为中文。
|
||||
2. 从`properties`选择对象标识：优先用`path`值，若无则用`name`值。
|
||||
3. 智能映射原始查询的属性名称：
|
||||
- 如果属性名称是上下文属性的缩写、省略或同义词，映射到实际属性名称（如“人工费”可能映射到“费率”或“合价含税”）。
|
||||
- 如果无法映射，保留原始名称。
|
||||
4. 保留原始查询的额外操作（如计算指令）。
|
||||
5. 输出格式：“获取[对象标识][对象类型]的[属性]属性，[额外操作]”。
|
||||
|
||||
示例参考：
|
||||
- 输入：原始问题="查找名称中包含“工程”的项目划分项，并返回其人工费乘以1000的值。", 上下文知识=...
|
||||
- 输出="获取[安装/架空输电线路本体工程/基础工程/基础工程材料工地运输]项目划分项的人工费，并乘以1000的值"
|
||||
|
||||
现在，处理以下输入：
|
||||
- 原始问题：{user_input}
|
||||
- 上下文知识：{context}
|
||||
""")
|
||||
|
||||
cypher_conversion_prompt = ChatPromptTemplate.from_template(
|
||||
"""
|
||||
你是一名电力造价业务专家，负责将用户自然语言问题中需要访问的对象识别出来，并生成针对该对象的NEO4J知识图谱的Cypher查询语句，获取该对象的全部信息。知识图谱基于从文件“获取[安装/架空输电线路本体工程]项目划分项的单位.”中读取的层级关联结构构建。该文件包含用反斜杠分割的多个字符串（如“安装/架空输电线路本体工程”），每个字符串表示一个完整的层级路径，路径部分用斜杠分隔，对应于知识图谱中ProjectDivisionItem节点的层级关系。路径映射规则：每个路径部分（如“安装”）是一个ProjectDivisionItem节点，父子关系通过关系类型`:CHILD_OF`连接，形成从根节点到叶节点的层级结构。
|
||||
|
||||
示例业务结构：
|
||||
{business_structure}
|
||||
|
||||
用户问题：
|
||||
{user_input}
|
||||
|
||||
改写规则：
|
||||
识别目标层级: 从用户输入中解析层级路径，例如字符串“安装/架空输电线路本体工程”映射为：根节点(name: '安装') -[:CHILD_OF]-> 子节点(name: '架空输电线路本体工程')。
|
||||
识别目标对象：仅从业务结构中识别如下类型：ProjectDivisionItem、ProjectAttributeSet、FeeSchedule、FeeItem的核心对象。对象类型必须精确映射到结构中的节点标签（例如，使用 ProjectDivisionItem 而非“项目划分项”）。
|
||||
提取查询条件：从用户输入中解析关键条件（如名称、量、类型、值、层级等），条件应基于对象属性（如 name、quantity、type），注意区分查找子串和相等的区别。如果条件涉及层级路径（如路径中包含特定部分），需提取路径部分作为条件（例如，用户提到“架空输电线路”时，解析为路径匹配）。
|
||||
如果用户指定层级（如“叶节点”），需在条件中体现（例如，添加 WHERE item.isLeaf = true）。
|
||||
忽略任何计算、转换或后处理要求（如“乘以1000”），只关注获取原始数据对象或属性。
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
# src/user_interaction.py
|
||||
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from langchain_core.output_parsers import JsonOutputParser
|
||||
from langchain_core.prompts import PromptTemplate
|
||||
from langchain.prompts import ChatPromptTemplate
|
||||
from langchain_openai import ChatOpenAI
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import List, Dict, Any
|
||||
import asyncio
|
||||
|
||||
current_file = os.path.splitext(os.path.basename(__file__))[0]
|
||||
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
log_filename = f"{current_file}_{now_str}.log"
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
],
|
||||
)
|
||||
|
||||
logger = logging.getLogger(current_file)
|
||||
|
||||
class BusinessObject(BaseModel):
|
||||
name: str = Field(description="对象的名称")
|
||||
constraints: Dict[str, Any] = Field(description="对象约束条件")
|
||||
|
||||
class UserInteraction:
|
||||
def __init__(self, llm: ChatOpenAI, business_structure: str):
|
||||
self.llm = llm
|
||||
self.business_structure = business_structure
|
||||
self.prompt_template = ChatPromptTemplate.from_template("""
|
||||
请基于提供的"示例业务结构"，分析用户输入问题。严格识别用户输入中提到的所有业务对象（对象必须来源于业务结构，不能是属性）。输出一个JSON数组，每个元素是一个对象，包含键"name"和"constraints"：
|
||||
- "name": 对象的名称，优先使用用户输入中指定的属性值（如项目划分名称、费用项名称等）；如果未指定具体名称，则使用对象类型（如"ProjectDivisionItem"）。
|
||||
- "constraints": 一个对象，包含：
|
||||
- "type": 业务结构中的对象类型（如ProjectDivisionItem、ProjectQuantity）。
|
||||
- 其他属性：用户输入中明确指定的属性条件（如"path"、"name"、"类型"、"编码"等），使用业务结构中的属性名（英文或中文，需一致）。
|
||||
注意：
|
||||
- 对象只能来源于业务结构（如EngineeringData的子对象），勿将属性（如"数量"、"编码"）识别为对象。
|
||||
- 勿执行查询、过滤或汇总操作；只识别对象及其约束。
|
||||
- 对于集合性描述（如"所有【定额】"），输出一个对象代表该集合，在"constraints"中指定条件。
|
||||
- 输出仅基于用户输入，不推断额外信息。
|
||||
|
||||
示例业务结构（供参考）：
|
||||
{business_structure}
|
||||
|
||||
示例输入：从项目划分【架空输电线路本体工程/辅助工程】的费用集中查找名称为【'合计'】费用
|
||||
示例输出：
|
||||
[
|
||||
{{
|
||||
"name": "辅助工程",
|
||||
"constraints": {{
|
||||
"type": "ProjectDivisionItem",
|
||||
"path": "架空输电线路本体工程/辅助工程",
|
||||
"name": "辅助工程"
|
||||
}}
|
||||
}},
|
||||
{{
|
||||
"name": "合计",
|
||||
"constraints": {{
|
||||
"parent": "辅助工程",
|
||||
"type": "CostItem",
|
||||
"name": "合计"
|
||||
}}
|
||||
}}
|
||||
]
|
||||
|
||||
用户输入：{query}
|
||||
""")
|
||||
self.parser = JsonOutputParser(pydantic_object=List[BusinessObject])
|
||||
|
||||
def understand(self, user_input: str) -> List[BusinessObject]:
|
||||
prompt_text = self.prompt_template.format_prompt(query=user_input, business_structure = self.business_structure)
|
||||
messages = prompt_text.to_messages()
|
||||
response = self.llm.invoke(messages)
|
||||
return self.parser.parse(response.text())
|
||||
|
||||
async def aunderstand(self, user_input: str) -> List[BusinessObject]:
|
||||
prompt_text = self.prompt_template.format_prompt(query=user_input)
|
||||
messages = prompt_text.to_messages()
|
||||
response = await self.llm.ainvoke(messages)
|
||||
return self.parser.parse(response.text())
|
||||
Reference in New Issue
Block a user