|
|
# agent/question/question_utils.py
|
|
|
import json
|
|
|
from config import *
|
|
|
from agent.llm_client import create_chat_completion
|
|
|
from agent.sql.exec import exec_sql_s
|
|
|
from agent.utils import find_json, dict_to_sentence
|
|
|
|
|
|
def preprocessing_question(question):
|
|
|
"""
|
|
|
SQL问题预处理
|
|
|
"""
|
|
|
if "年末" in question:
|
|
|
question = question.replace("年末","年12月31日")
|
|
|
if "年底" in question:
|
|
|
question = question.replace("年底","年12月31日")
|
|
|
return question
|
|
|
|
|
|
def add_question_mark(text):
|
|
|
# 检查文本最后是否缺少问号
|
|
|
if not text.endswith('?') and not text.endswith('？'):
|
|
|
text += '？'
|
|
|
|
|
|
# 检查文本中右括号“）”的对应左括号“（”的处理
|
|
|
if len(text) >= 2 and text[-2] == '）':
|
|
|
left_bracket_index = text.rfind('（')
|
|
|
if left_bracket_index != -1:
|
|
|
if left_bracket_index == 0 or text[left_bracket_index - 1] not in ['?', '？']:
|
|
|
text = text[:left_bracket_index] + '？' + text[left_bracket_index:]
|
|
|
|
|
|
return text
|
|
|
|
|
|
def process_company_name(value):
|
|
|
"""
|
|
|
Given a company name (or related keyword), search in three tables:
|
|
|
ConstantDB.SecuMain, ConstantDB.HK_SecuMain, ConstantDB.US_SecuMain.
|
|
|
|
|
|
Attempts to match various company-related fields (e.g., ChiName, EngName, etc.)
|
|
|
and returns all matching results along with the table where they were found.
|
|
|
|
|
|
Parameters:
|
|
|
value (str): The company name or related string to match.
|
|
|
|
|
|
Returns:
|
|
|
list: A list of tuples (result, table) where result is the matched data and table is the table name.
|
|
|
If no matches found, prints a message and returns an empty list.
|
|
|
"""
|
|
|
res_lst = []
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
if len(QUESTION_TYPE_LIST) == 0:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
elif len(QUESTION_TYPE_LIST) >= 2:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
elif "港股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.HK_SecuMain']
|
|
|
elif "美股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.US_SecuMain']
|
|
|
elif "A股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.SecuMain']
|
|
|
else:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
# print(f"--------------tables:{tables}")
|
|
|
columns_to_match = ['CompanyCode', 'SecuCode', 'ChiName', 'ChiNameAbbr',
|
|
|
'EngName', 'EngNameAbbr', 'SecuAbbr', 'ChiSpelling']
|
|
|
columns_to_select = ['InnerCode', 'CompanyCode', 'SecuCode', 'ChiName', 'ChiNameAbbr',
|
|
|
'EngName', 'EngNameAbbr', 'SecuAbbr', 'ChiSpelling']
|
|
|
|
|
|
# Escape single quotes to prevent SQL injection
|
|
|
value = value.replace("'", "''")
|
|
|
|
|
|
for table in tables:
|
|
|
# For the US table, remove columns that may not be available
|
|
|
local_match_cols = columns_to_match.copy()
|
|
|
local_select_cols = columns_to_select.copy()
|
|
|
if 'US' in table:
|
|
|
if 'ChiNameAbbr' in local_match_cols:
|
|
|
local_match_cols.remove('ChiNameAbbr')
|
|
|
if 'ChiNameAbbr' in local_select_cols:
|
|
|
local_select_cols.remove('ChiNameAbbr')
|
|
|
if 'EngNameAbbr' in local_match_cols:
|
|
|
local_match_cols.remove('EngNameAbbr')
|
|
|
if 'EngNameAbbr' in local_select_cols:
|
|
|
local_select_cols.remove('EngNameAbbr')
|
|
|
|
|
|
# Build the WHERE clause with OR conditions for each column
|
|
|
match_conditions = [f"{col} = '{value}'" for col in local_match_cols]
|
|
|
where_clause = ' OR '.join(match_conditions)
|
|
|
|
|
|
sql = f"""
|
|
|
SELECT {', '.join(local_select_cols)}
|
|
|
FROM {table}
|
|
|
WHERE {where_clause}
|
|
|
"""
|
|
|
result = exec_sql_s(sql)
|
|
|
if result:
|
|
|
res_lst.append((result, table))
|
|
|
# else:
|
|
|
# # The 'else' clause in a for loop runs only if no 'break' was encountered.
|
|
|
# # Here it just prints if no results were found.
|
|
|
# if not res_lst:
|
|
|
# if DEBUG_VER == 3:
|
|
|
# print(f"未在任何表中找到上市公司名称为 {value} 的信息。")
|
|
|
else:
|
|
|
# If no result, modify query for fuzzy matching
|
|
|
fuzzy_match_conditions = [f"{col} LIKE '%{value}%' " for col in local_match_cols]
|
|
|
fuzzy_where_clause = ' OR '.join(fuzzy_match_conditions)
|
|
|
|
|
|
# Query with fuzzy match
|
|
|
sql_fuzzy = f"""
|
|
|
SELECT {', '.join(local_select_cols)}
|
|
|
FROM {table}
|
|
|
WHERE {fuzzy_where_clause}
|
|
|
"""
|
|
|
fuzzy_result = exec_sql_s(sql_fuzzy)
|
|
|
if fuzzy_result:
|
|
|
res_lst.append((fuzzy_result, table))
|
|
|
|
|
|
# If no results found after both exact and fuzzy matches
|
|
|
if not res_lst:
|
|
|
if DEBUG_VER == 3:
|
|
|
print(f"未在任何表中找到上市公司名称为 {value} 的信息。")
|
|
|
|
|
|
return res_lst
|
|
|
|
|
|
def process_code(value):
|
|
|
"""
|
|
|
Given a code (e.g., a stock code), search the three tables and return matches.
|
|
|
|
|
|
Parameters:
|
|
|
value (str): The code to search for.
|
|
|
|
|
|
Returns:
|
|
|
list: A list of tuples (result, table) if found, else empty.
|
|
|
"""
|
|
|
res_lst = []
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
if len(QUESTION_TYPE_LIST) == 0:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
elif len(QUESTION_TYPE_LIST) >= 2:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
elif "港股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.HK_SecuMain']
|
|
|
elif "美股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.US_SecuMain']
|
|
|
elif "A股" in QUESTION_TYPE_LIST:
|
|
|
tables = ['ConstantDB.SecuMain']
|
|
|
else:
|
|
|
tables = ['ConstantDB.SecuMain', 'ConstantDB.HK_SecuMain', 'ConstantDB.US_SecuMain']
|
|
|
# print(f"--------------tables:{tables}")
|
|
|
columns_to_select = ['InnerCode', 'CompanyCode', 'SecuCode', 'ChiName', 'ChiNameAbbr',
|
|
|
'EngName', 'EngNameAbbr', 'SecuAbbr', 'ChiSpelling']
|
|
|
|
|
|
value = value.replace("'", "''") # Escape single quotes
|
|
|
|
|
|
for table in tables:
|
|
|
local_select_cols = columns_to_select.copy()
|
|
|
if 'US' in table:
|
|
|
if 'ChiNameAbbr' in local_select_cols:
|
|
|
local_select_cols.remove('ChiNameAbbr')
|
|
|
if 'EngNameAbbr' in local_select_cols:
|
|
|
local_select_cols.remove('EngNameAbbr')
|
|
|
|
|
|
sql = f"""
|
|
|
SELECT {', '.join(local_select_cols)}
|
|
|
FROM {table}
|
|
|
WHERE SecuCode = '{value}'
|
|
|
"""
|
|
|
result = exec_sql_s(sql)
|
|
|
if result:
|
|
|
res_lst.append((result, table))
|
|
|
else:
|
|
|
if not res_lst:
|
|
|
if DEBUG_VER == 3:
|
|
|
print(f"未在任何表中找到代码为 {value} 的信息。")
|
|
|
return res_lst
|
|
|
|
|
|
|
|
|
def process_jj(value):
|
|
|
sql = f"SELECT ChiName, CompanyCode FROM InstitutionDB.LC_InstiArchive WHERE ChiName LIKE '%{value}%'"
|
|
|
res = exec_sql_s(sql)
|
|
|
if res:
|
|
|
return [(res,'InstitutionDB.LC_InstiArchive')]
|
|
|
else:
|
|
|
res_lst = process_company_name(value)
|
|
|
if res_lst:
|
|
|
return res_lst
|
|
|
return []
|
|
|
|
|
|
|
|
|
def process_items(item_list):
|
|
|
"""
|
|
|
Given a list of items (dictionaries) from JSON extraction, attempt to process each based on its key:
|
|
|
- If key is '基金名称' or '上市公司名称', use process_company_name.
|
|
|
- If key is '代码', use process_code.
|
|
|
- Otherwise, print an unrecognized key message.
|
|
|
|
|
|
Parameters:
|
|
|
item_list (list): A list of dictionaries like [{"上市公司名称": "XX公司"}, {"代码":"600872"}].
|
|
|
|
|
|
Returns:
|
|
|
tuple: (res, tables)
|
|
|
res (str): A formatted string showing what was found.
|
|
|
tables (list): A list of table names where matches were found.
|
|
|
"""
|
|
|
res_list = []
|
|
|
try:
|
|
|
for item in item_list:
|
|
|
key, value = list(item.items())[0]
|
|
|
if key in ["基金名称", "上市公司名称"]:
|
|
|
res_list.extend(process_company_name(value))
|
|
|
elif key == "代码":
|
|
|
res_list.extend(process_code(value))
|
|
|
elif key == "基金公司简称":
|
|
|
res_list.extend(process_jj(value))
|
|
|
else:
|
|
|
if DEBUG_VER == 3:
|
|
|
print(f"无法识别的键：{key}")
|
|
|
pass
|
|
|
except Exception as e:
|
|
|
if DEBUG_VER == 3:
|
|
|
print(f"process_items 发生错误: {e}")
|
|
|
pass
|
|
|
# Filter out empty results
|
|
|
res_list = [i for i in res_list if i]
|
|
|
res = ''
|
|
|
tables = []
|
|
|
for result_data, table_name in res_list:
|
|
|
tables.append(table_name)
|
|
|
res += f"预处理程序通过表格：{table_name} 查询到以下内容：\n {json.dumps(result_data, ensure_ascii=False, indent=1)} \n"
|
|
|
|
|
|
return res, tables
|
|
|
|
|
|
|
|
|
def process_question(question):
|
|
|
"""
|
|
|
Given a question, run it through a prompt to perform Named Entity Recognition (NER),
|
|
|
extract entities (上市公司名称, 代码, 基金名称), parse the assistant's JSON response,
|
|
|
and process the items to retrieve relevant information from the database.
|
|
|
|
|
|
Parameters:
|
|
|
question (str): The user question.
|
|
|
|
|
|
Returns:
|
|
|
tuple: (res, tables) where
|
|
|
res (str) - Processed result details as a string.
|
|
|
tables (list) - List of tables involved in the final result.
|
|
|
"""
|
|
|
prompt = '''
|
|
|
你将会进行命名实体识别任务，并输出实体json。你只需要识别以下4种实体：
|
|
|
|
|
|
-上市公司名称
|
|
|
-代码
|
|
|
-基金名称
|
|
|
-基金公司简称
|
|
|
|
|
|
其中，上市公司名称可以是全称，简称，拼音缩写，代码包含股票代码和基金代码，基金名称包含债券型基金，
|
|
|
以下是几个示例：
|
|
|
user:唐山港集团股份有限公司是什么时间上市的（回答XXXX-XX-XX）
|
|
|
当年一共上市了多少家企业？
|
|
|
这些企业有多少是在北京注册的？
|
|
|
assistant:```json
|
|
|
[{"上市公司名称":"唐山港集团股份有限公司"}]
|
|
|
```
|
|
|
user:JD的职工总数有多少人？
|
|
|
该公司披露的硕士或研究生学历（及以上）的有多少人？
|
|
|
20201月1日至年底退休了多少人？
|
|
|
assistant:```json
|
|
|
[{"上市公司名称":"JD"}]
|
|
|
```
|
|
|
user:600872的全称、A股简称、法人、法律顾问、会计师事务所及董秘是？
|
|
|
该公司实控人是否发生改变？如果发生变化，什么时候变成了谁？是哪国人？是否有永久境外居留权？（回答时间用XXXX-XX-XX）
|
|
|
assistant:```json
|
|
|
[{"代码":"600872"}]
|
|
|
```
|
|
|
user:华夏鼎康债券A在2019年的分红次数是多少？每次分红的派现比例是多少？
|
|
|
基于上述分红数据，在2019年最后一次分红时，如果一位投资者持有1000份该基金，税后可以获得多少分红收益？
|
|
|
assistant:```json
|
|
|
[{"基金名称":"华夏鼎康债券A"}]
|
|
|
```
|
|
|
user:实体识别任务：```易方达基金管理有限公司在19年成立了多少支基金？
|
|
|
哪支基金的规模最大？
|
|
|
这支基金20年最后一次分红派现比例多少钱？```
|
|
|
assistant:```json
|
|
|
[{"基金公司简称":"易方达"}]
|
|
|
```
|
|
|
user:化工纳入过多少个子类概念？
|
|
|
assistant:```json
|
|
|
[]
|
|
|
```
|
|
|
'''
|
|
|
messages = [{'role': 'system', 'content': prompt}, {'role': 'user', 'content': question}]
|
|
|
aa = create_chat_completion(messages)
|
|
|
bb = find_json(aa.choices[0].message.content)
|
|
|
return process_items(bb)
|
|
|
|
|
|
def question_rew(context_text, original_question):
|
|
|
"""
|
|
|
Rewrite the given question to be clearer and more specific based on the provided context,
|
|
|
without altering the original meaning or omitting any information.
|
|
|
|
|
|
Parameters:
|
|
|
context_text (str): The context text that the question is based on.
|
|
|
original_question (str): The question to be rewritten.
|
|
|
|
|
|
Returns:
|
|
|
str: The rewritten question.
|
|
|
"""
|
|
|
prompt = (
|
|
|
f"根据这些内容：'{context_text}',帮我重写当前问题：'{original_question}' ,让问题清晰明确，"
|
|
|
"不改变原意，代词转成具体人事物，不要遗漏信息，只返回问题。"
|
|
|
"如果当前问题中有时间代词（如“当年、当天”等）或指物代词（如“该公司”、“它”等），检查前面问题和回答（一般是上一个，就近原则）中是否明确了时间或主体等，并将这些信息补充到当前问题中。"
|
|
|
"如果当前问题无法从前面问题和回答中找到代词（如时间）所指的具体信息，则表示当前代词，如时间指全部时间。"
|
|
|
"问题可能需要时间，也可能不需要时间，如果不需要则在后面追加一个不带时间的小问题（不需要换行等，只需要接在原问题后面）。"
|
|
|
"让我们一步一步思考"
|
|
|
"以下是几个示例：\n"
|
|
|
"user:根据这些内容：'第1轮问题：最新更新的2021年度报告中，机构持有无限售流通A股数量合计最多的公司简称是？ 第1轮回答：公司简称 帝尔激光',帮我重写当前问题：'在这份报告中，该公司机构持有无限售流通A股比例合计是多少，保留2位小数？'"
|
|
|
"assistant:最新更新的2021年度报告中,公司简称 帝尔激光 持有无限售流通A股比例合计是多少，保留2位小数？"
|
|
|
"user:根据这些内容：'第1轮问题：TK他是否已经成立了？（是或者否） 第1轮回答：是',帮我重写当前问题：'这家公司17年最高收盘价是多少？'"
|
|
|
"assistant: 2017年，TK这家公司的最高收盘价是多少？"
|
|
|
"user:根据这些内容：'第1轮问题：TK他是否已经成立了？（是或者否） 第1轮回答：是 \n第2轮问题：2017年，TK这家公司的最高收盘价是多少？ 第2轮回答：2017年TK最高收盘价 5.79',帮我重写当前问题：'当天有多少家公司成立了？'"
|
|
|
"assistant: 2017年TK最高收盘价 10.79 是什么时候？当天有多少家公司成立了？"
|
|
|
"user:根据这些内容：'第1轮问题：航锦科技股份有限公司是否变更过公司名称？ 第1轮回答：否（没有）\n第2轮问题：？航锦科技股份有限公司涉及回购的最大的一笔金额是多少？第2轮回答：43951008.0',帮我重写当前问题：该年度前十大股东的持股比例变成了多少？"
|
|
|
"assistant:？航锦科技股份有限公司涉及回购的最大的一笔金额 43951008.0是哪一年？该年度前十大股东的持股比例变成了多少？"
|
|
|
)
|
|
|
|
|
|
messages = [{"role": "user", "content": prompt}]
|
|
|
response = create_chat_completion(messages)
|
|
|
return response.choices[0].message.content
|
|
|
|
|
|
def process_dict(d):
|
|
|
"""
|
|
|
Recursively process a nested dictionary to produce a comma-separated description.
|
|
|
For nested dictionaries, it processes them recursively and returns a descriptive string.
|
|
|
|
|
|
For example:
|
|
|
{
|
|
|
"company": {
|
|
|
"name": "ABC Corp",
|
|
|
"location": "New York"
|
|
|
},
|
|
|
"year": 2021
|
|
|
}
|
|
|
might be processed into a string like:
|
|
|
"company company 是 name 是 ABC Corp, location 是 New York, year 2021"
|
|
|
|
|
|
Parameters:
|
|
|
d (dict): A dictionary or another object to describe.
|
|
|
|
|
|
Returns:
|
|
|
str: A descriptive string.
|
|
|
"""
|
|
|
def recursive_process(sub_dict):
|
|
|
sentences = []
|
|
|
for key, value in sub_dict.items():
|
|
|
if isinstance(value, dict):
|
|
|
# Process nested dictionary and wrap result in dict_to_sentence for formatting
|
|
|
nested_result = recursive_process(value)
|
|
|
sentences.append(dict_to_sentence({key: nested_result}))
|
|
|
else:
|
|
|
# Non-dict values are directly appended
|
|
|
sentences.append(f"{key} {value}")
|
|
|
return ", ".join(sentences)
|
|
|
|
|
|
if not isinstance(d, dict):
|
|
|
# If it's not a dictionary, just return its string representation
|
|
|
return str(d)
|
|
|
|
|
|
return recursive_process(d) |
