liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-07-17 17:34:35 +08:00
Code Activity

update knowledge module;

Browse Source 
* Knowledge.storage to storages list;
* optimize Knowledge & Storage save and load method;
* optimize Knowledge query prompt;
This commit is contained in:
Cadenza-Li
2023-07-13 17:20:22 +08:00
parent d7ab6935dd
commit 37d83fd747
3 changed files with 150 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
qlib/finco/knowledge.py
View File

@@ -3,12 +3,13 @@ from jinja2 import Template
from typing import List, Union from typing import List, Union
import pickle import pickle
import yaml import yaml
import inspect
from qlib.workflow import R from qlib.workflow import R
from qlib.finco.log import FinCoLog from qlib.finco.log import FinCoLog
from qlib.finco.llm import APIBackend from qlib.finco.llm import APIBackend
from qlib.finco.utils import similarity from qlib.finco.utils import similarity, random_string
logger = FinCoLog()
class Storage: class Storage:
@@ -17,9 +18,9 @@ class Storage:
""" """
def __init__(self, path: Union[str, Path]): def __init__(self, path: Union[str, Path], name: str = None):
self.path = path if isinstance(path, Path) else Path(path) self.path = path if isinstance(path, Path) else Path(path)
self.logger = FinCoLog() self.name = name if name else self.path.name
self.source = None self.source = None
# todo: get document by key # todo: get document by key
@@ -54,6 +55,7 @@ class PickleStorage(Storage):
def save(self, **kwargs): def save(self, **kwargs):
"""use pickle as the default save method""" """use pickle as the default save method"""
Path.mkdir(self.path.parent, exist_ok=True)
with open(self.path, "wb") as f: with open(self.path, "wb") as f:
pickle.dump(self, f) pickle.dump(self, f)
@@ -63,9 +65,11 @@ class YamlStorage(Storage):
This class is responsible for storage and loading of Knowledge related data in yaml format. This class is responsible for storage and loading of Knowledge related data in yaml format.
""" """
DEFAULT_NAME = "storage.yml"
def __init__(self, path: Union[str, Path]): def __init__(self, path: Union[str, Path]):
super().__init__(path) super().__init__(path)
assert self.path.name, "Yaml storage should specify file name."
self.load() self.load()
def load(self): def load(self):
@@ -73,10 +77,11 @@ class YamlStorage(Storage):
try: try:
self.documents = yaml.load(open(self.path, "r"), Loader=yaml.FullLoader) self.documents = yaml.load(open(self.path, "r"), Loader=yaml.FullLoader)
except FileNotFoundError: except FileNotFoundError:
self.logger.warning(f"YamlStorage: file {self.path} doesn't exist.") logger.warning(f"YamlStorage: file {self.path} doesn't exist.")
def save(self, **kwargs): def save(self, **kwargs):
"""use pickle as the default save method""" """use pickle as the default save method"""
Path.mkdir(self.path.parent, exist_ok=True)
with open(self.path, 'w') as f: with open(self.path, 'w') as f:
yaml.dump(self.documents, f) yaml.dump(self.documents, f)
@@ -109,11 +114,21 @@ class Knowledge:
Use to handle knowledge in finCo such as experiment and outside domain information Use to handle knowledge in finCo such as experiment and outside domain information
""" """
def __init__(self, storage: Storage): def __init__(self, storages: Union[List[Storage], Storage], name: str = None):
self.logger = FinCoLog() self.name = name if name else random_string()
self.storage = storage self.workdir = Path.cwd().joinpath("knowledge")
self.storages = [storages] if isinstance(storages, Storage) else storages
self.knowledge = [] self.knowledge = []
def get_storage(self, name: str):
"""
return first storage matched given name, else return None
"""
for storage in self.storages:
if storage.name == name:
return storage
return None
def summarize(self, **kwargs): def summarize(self, **kwargs):
""" """
summarize storage data to knowledge, default knowledge is storage.documents summarize storage data to knowledge, default knowledge is storage.documents
@@ -124,20 +139,24 @@ class Knowledge:
Return Return
------ ------
""" """
for storage in self.storages:
self.knowledge.extend(storage.documents)
self.knowledge = self.storage.documents @classmethod
def load(cls, path: Union[str, Path]):
def load(self, **kwargs):
""" """
Load knowledge in memory Load knowledge in memory
use pickle as the default file type
Parameters Parameters
---------- ----------
Return Return
------ ------
""" """
raise NotImplementedError(f"Please implement the `load` method.") """"""
path = path if isinstance(path, Path) else Path(path)
with open(path, "rb") as f:
return pickle.load(f)
def brief(self, **kwargs): def brief(self, **kwargs):
""" """
@@ -153,7 +172,10 @@ class Knowledge:
def save(self, **kwargs): def save(self, **kwargs):
"""save knowledge persistently""" """save knowledge persistently"""
self.storage.save(**kwargs) # todo: storages save index only
Path.mkdir(self.workdir.joinpath(self.name), exist_ok=True)
with open(self.workdir.joinpath(self.name).joinpath("knowledge.pkl"), "wb") as f:
pickle.dump(self, f)
class ExperimentKnowledge(Knowledge): class ExperimentKnowledge(Knowledge):
@@ -161,9 +183,9 @@ class ExperimentKnowledge(Knowledge):
Handle knowledge from experiments Handle knowledge from experiments
""" """
def __init__(self, storage: ExperimentStorage): def __init__(self, storages: Union[List[ExperimentStorage], ExperimentStorage]):
super().__init__(storage=storage) super().__init__(storages=storages)
self.storage = storage self.storage = storages
def brief(self): def brief(self):
docs = [] docs = []
@@ -179,13 +201,15 @@ class PracticeKnowledge(Knowledge):
some template sentence for now some template sentence for now
""" """
def __init__(self, storage: YamlStorage): def __init__(self, storages: Union[List[YamlStorage], YamlStorage]):
super().__init__(storage=storage) super().__init__(storages=storages, name="practice")
self.summarize() self.summarize()
def add(self, docs: List): def add(self, docs: List):
self.storage.add(docs) storage = YamlStorage(path=self.workdir.joinpath(self.name).joinpath(YamlStorage.DEFAULT_NAME))
storage.add(documents=docs)
self.storages.append(storage)
self.summarize() self.summarize()
self.save() self.save()
@@ -196,15 +220,17 @@ class FinanceKnowledge(Knowledge):
Knowledge from articles Knowledge from articles
""" """
def __init__(self, storage: YamlStorage): def __init__(self, storages: Union[List[YamlStorage], YamlStorage]):
super().__init__(storage=storage) super().__init__(storages=storages, name="finance")
if len(self.storage.documents) == 0:
docs = self.read_files_in_directory(self.storage.path) docs = self.read_files_in_directory(self.workdir.joinpath(self.name))
self.add(docs) self.add(docs)
self.summarize() self.summarize()
def add(self, docs: List): def add(self, docs: List):
self.storage.add(docs) storage = YamlStorage(path=self.workdir.joinpath(self.name).joinpath(YamlStorage.DEFAULT_NAME))
storage.add(documents=docs)
self.storages.append(storage)
self.summarize() self.summarize()
self.save() self.save()
@@ -228,12 +254,14 @@ class ExecuteKnowledge(Knowledge):
Config and associate execution result(pass or error message). We can regard the example in prompt as pass execution Config and associate execution result(pass or error message). We can regard the example in prompt as pass execution
""" """
def __init__(self, storage: YamlStorage): def __init__(self, storages: Union[List[YamlStorage], YamlStorage]):
super().__init__(storage=storage) super().__init__(storages=storages, name="execute")
self.summarize() self.summarize()
def add(self, docs: List): def add(self, docs: List):
self.storage.add(docs) storage = YamlStorage(path=self.workdir.joinpath(YamlStorage.DEFAULT_NAME))
storage.add(documents=docs)
self.storages.append(storage)
self.summarize() self.summarize()
self.save() self.save()
@@ -244,34 +272,65 @@ class InfrastructureKnowledge(Knowledge):
Knowledge from sentences, docstring, and code Knowledge from sentences, docstring, and code
""" """
def __init__(self, storage: YamlStorage): def __init__(self, storages: Union[List[YamlStorage], YamlStorage]):
super().__init__(storage=storage) super().__init__(storages=storages, name="infrastructure")
if len(self.storage.documents) == 0: storage = self.get_storage(YamlStorage.DEFAULT_NAME)
# todo: change the path to qlib root path if len(storage.documents) == 0:
docs = self.get_functions_and_docstrings(Path.cwd().parent) docs = self.get_functions_and_docstrings(Path(__file__).parent.parent.parent)
self.add(docs) self.add(docs)
def add(self, docs: List): def add(self, docs: List):
self.storage.add(docs) storage = YamlStorage(path=self.workdir.joinpath(self.name).joinpath(YamlStorage.DEFAULT_NAME))
storage.add(documents=docs)
self.storages.append(storage)
self.summarize() self.summarize()
self.save() self.save()
@staticmethod def get_functions_and_docstrings(self, directory):
def get_functions_and_docstrings(directory):
""" """
get all method and docstring in .py files under directory get all method and docstring in .py files under directory
""" """
functions = [] functions = []
for file_path in Path(directory).rglob("*.py"): for py_file_path in Path(directory).rglob('*.py'):
with file_path.open("r", encoding="utf-8") as f: for _functions in self.get_functions_with_docstrings(py_file_path):
lines = f.readlines() functions.append(_functions)
for line in lines:
if line.startswith("def "): return functions
function_name = line.split("(")[0][4:].strip()
function_docstring = inspect.getdoc(eval(function_name)) @staticmethod
functions.append({"function_name": function_name, "docstring": function_docstring}) def get_functions_with_docstrings(file_path):
"""
Extract method name and docstring using string matching method
"""
with open(file_path, "r", encoding="utf8") as f:
lines = f.readlines()
functions = []
current_func = None
docstring = None
for line in lines:
if line.strip().startswith("def ") or line.strip().startswith("class "):
func = line.strip().split(' ')[1].split('(')[0]
if func.startswith("__"):
continue
if current_func is not None:
docstring = docstring.replace('"""', "") if docstring else docstring
functions.append({"function": current_func, "docstring": docstring})
current_func = f"{file_path.name.split('.')[0]}.{func}"
docstring = None
elif current_func is not None and docstring is None and line.strip().startswith('"""'):
docstring = line
elif current_func is not None and docstring is not None:
docstring += line.strip()
if line.strip().endswith('"""'):
docstring = docstring.replace('"""', "") if docstring else docstring
functions.append({"function": current_func, "docstring": docstring})
current_func = None
docstring = None
return functions return functions
@@ -303,18 +362,20 @@ class KnowledgeBase:
KT_EXECUTE = "execute" KT_EXECUTE = "execute"
KT_PRACTICE = "practice" KT_PRACTICE = "practice"
KT_FINANCE = "finance" KT_FINANCE = "finance"
KT_INFRASTRUCTURE = "infrastructure"
def __init__(self, init_path=None): def __init__(self, workdir=None):
self.logger = FinCoLog() self.logger = FinCoLog()
self.init_path = Path(init_path) if init_path else Path.cwd() self.workdir = Path(workdir) if workdir else Path.cwd()
if not self.init_path.exists(): if not self.workdir.exists():
self.logger.warning(f"{self.init_path} not exist, create empty directory.") self.logger.warning(f"{self.workdir} not exist, create empty directory.")
Path.mkdir(self.init_path) Path.mkdir(self.workdir)
self.practice_knowledge = self.load_practice_knowledge(self.init_path) self.practice_knowledge = self.load_practice_knowledge(self.workdir)
self.execute_knowledge = self.load_execute_knowledge(self.init_path) self.execute_knowledge = self.load_execute_knowledge(self.workdir)
self.finance_knowledge = self.load_finance_knowledge(self.init_path) self.finance_knowledge = self.load_finance_knowledge(self.workdir)
self.infrastructure_knowledge = self.load_infrastructure_knowledge(self.workdir)
def load_experiment_knowledge(self, path) -> List: def load_experiment_knowledge(self, path) -> List:
# similar to practice knowledge, not use for now # similar to practice knowledge, not use for now
@@ -326,33 +387,43 @@ class KnowledgeBase:
# todo: check the influence of set uri # todo: check the influence of set uri
R.set_uri(path.as_uri()) R.set_uri(path.as_uri())
for exp_name in R.list_experiments(): for exp_name in R.list_experiments():
knowledge.append(ExperimentKnowledge(storage=ExperimentStorage(exp_name=exp_name))) knowledge.append(ExperimentKnowledge(storages=ExperimentStorage(exp_name=exp_name)))
self.logger.plain_info(f"Load knowledge from: {path} finished.") self.logger.plain_info(f"Load knowledge from: {path} finished.")
return knowledge return knowledge
def load_practice_knowledge(self, path: Path) -> PracticeKnowledge: def load_practice_knowledge(self, path: Path) -> PracticeKnowledge:
self.practice_knowledge = PracticeKnowledge(YamlStorage(path.joinpath("practice_knowledge.yaml"))) self.practice_knowledge = PracticeKnowledge(
YamlStorage(path.joinpath(f"{self.KT_PRACTICE}/{YamlStorage.DEFAULT_NAME}")))
return self.practice_knowledge return self.practice_knowledge
def load_execute_knowledge(self, path: Path) -> ExecuteKnowledge: def load_execute_knowledge(self, path: Path) -> ExecuteKnowledge:
self.execute_knowledge = ExecuteKnowledge(YamlStorage(path.joinpath("execute_knowledge.yaml"))) self.execute_knowledge = ExecuteKnowledge(
YamlStorage(path.joinpath(f"{self.KT_EXECUTE}/{YamlStorage.DEFAULT_NAME}")))
return self.execute_knowledge return self.execute_knowledge
def load_finance_knowledge(self, path: Path) -> FinanceKnowledge: def load_finance_knowledge(self, path: Path) -> FinanceKnowledge:
self.finance_knowledge = FinanceKnowledge(YamlStorage(path.joinpath("finance_knowledge.yaml"))) self.finance_knowledge = FinanceKnowledge(
YamlStorage(path.joinpath(f"{self.KT_FINANCE}/{YamlStorage.DEFAULT_NAME}")))
return self.finance_knowledge return self.finance_knowledge
def knowledge(self, knowledge_type: str = None): def load_infrastructure_knowledge(self, path: Path) -> InfrastructureKnowledge:
self.infrastructure_knowledge = InfrastructureKnowledge(
YamlStorage(path.joinpath(f"{self.KT_INFRASTRUCTURE}/{YamlStorage.DEFAULT_NAME}")))
return self.infrastructure_knowledge
def get_knowledge(self, knowledge_type: str = None):
if knowledge_type == self.KT_EXECUTE: if knowledge_type == self.KT_EXECUTE:
knowledge = self.execute_knowledge knowledge = self.execute_knowledge.knowledge
elif knowledge_type == self.KT_PRACTICE: elif knowledge_type == self.KT_PRACTICE:
knowledge = self.practice_knowledge knowledge = self.practice_knowledge.knowledge
elif knowledge_type == self.KT_FINANCE: elif knowledge_type == self.KT_FINANCE:
knowledge = self.finance_knowledge knowledge = self.finance_knowledge.knowledge
elif knowledge_type == self.KT_INFRASTRUCTURE:
knowledge = self.infrastructure_knowledge.knowledge
else: else:
knowledge = self.execute_knowledge.knowledge + self.practice_knowledge.knowledge \ knowledge = self.execute_knowledge.knowledge + self.practice_knowledge.knowledge \
+ self.finance_knowledge.knowledge + self.finance_knowledge.knowledge + self.infrastructure_knowledge.knowledge
return knowledge return knowledge
def query(self, knowledge_type: str = None, content: str = None, n: int = 5): def query(self, knowledge_type: str = None, content: str = None, n: int = 5):
@@ -366,7 +437,7 @@ class KnowledgeBase:
# todo: replace list with persistent storage strategy such as ES/pinecone to enable # todo: replace list with persistent storage strategy such as ES/pinecone to enable
# literal search/semantic search # literal search/semantic search
knowledge = self.knowledge(knowledge_type=knowledge_type) knowledge = self.get_knowledge(knowledge_type=knowledge_type)
scores = [] scores = []
for k in knowledge: for k in knowledge:
scores.append(similarity(str(k), content)) scores.append(similarity(str(k), content))
@@ -374,10 +445,12 @@ class KnowledgeBase:
similar_n_indexes = sorted_indexes[:n] similar_n_indexes = sorted_indexes[:n]
similar_n_docs = [knowledge[i] for i in similar_n_indexes] similar_n_docs = [knowledge[i] for i in similar_n_indexes]
prompt = Template("""summarize this information: '{{docs}}'""") prompt = Template(
prompt_workflow_selection = prompt.render(docs=similar_n_docs) """find the most relevant doc with this query: '{{content}}' from docs='{{docs}}'.
Just return the most relevant item I provided, no more explain. For example: {'function': 'config.resolve_path', 'docstring': None}""")
prompt_workflow_selection = prompt.render(content=content, docs=similar_n_docs)
response = APIBackend().build_messages_and_create_chat_completion( response = APIBackend().build_messages_and_create_chat_completion(
user_prompt=prompt_workflow_selection user_prompt=prompt_workflow_selection, system_prompt="You are an excellent assistant."
) )
return response return response

7
qlib/finco/utils.py
View File

@@ -1,4 +1,6 @@
import json import json
import string
import random
from fuzzywuzzy import fuzz from fuzzywuzzy import fuzz
@@ -36,3 +38,8 @@ def similarity(text1, text2):
# Maybe we can use other similarity algorithm such as tfidf # Maybe we can use other similarity algorithm such as tfidf
return fuzz.ratio(text1, text2) return fuzz.ratio(text1, text2)
def random_string(length=10):
letters = string.ascii_letters + string.digits
return ''.join(random.choice(letters) for i in range(length))

6
qlib/finco/workflow.py
View File

@@ -176,7 +176,9 @@ class LearnManager:
self.topics = [Topic(name=topic, describe=self.wm.prompt_template.get(f"Topic_{topic}")) for topic in self.topics = [Topic(name=topic, describe=self.wm.prompt_template.get(f"Topic_{topic}")) for topic in
self.__DEFAULT_TOPICS] self.__DEFAULT_TOPICS]
self.knowledge_base = KnowledgeBase(init_path=Path.cwd().joinpath('knowledge')) self.knowledge_base = KnowledgeBase(workdir=Path.cwd().joinpath('knowledge'))
self.knowledge_base.execute_knowledge.add([])
self.knowledge_base.query(knowledge_type="infrastructure", content="resolve_path")
def run(self, prompt): def run(self, prompt):
# todo: add early stop condition # todo: add early stop condition
@@ -202,7 +204,7 @@ class LearnManager:
user_prompt = self.wm.context.get_context("user_prompt") user_prompt = self.wm.context.get_context("user_prompt")
summary = self.wm.context.get_context("summary") summary = self.wm.context.get_context("summary")
[topic.summarize(self.knowledge_base.knowledge()) for topic in self.topics] [topic.summarize(self.knowledge_base.get_knowledge()) for topic in self.topics]
knowledge_of_topics = [{topic.name: topic.knowledge} for topic in self.topics] knowledge_of_topics = [{topic.name: topic.knowledge} for topic in self.topics]
for task in task_finished: for task in task_finished: