self-constructing_graph/command/generate_doc_abstract.py

#通过分析文章，生成分析结果
import asyncio
import os
import time
from test.deepseek_chat import chat_with_deepseek

def load_prompt(filename):
    with open(filename, "r", encoding="utf-8") as f:
        return "".join(f.readlines())
    
async def chat(prompt: str) -> str:
    message = [{'role':'user', 'content': prompt}]
    call_deepseek = chat_with_deepseek(message)
    output = ""
    async for chunk in call_deepseek:
        output = output + chunk
        print(chunk, end="")
    print("\n")
    return output
        
if __name__ == "__main__":
    #
    #
    prompt_template = load_prompt("kb/prompt_4_abstract.txt")
    path = "./docs"    
    for root, dirs, files in os.walk(path):
        for file in files:
            file_path = os.path.join(root, file)
            print(">>> process", file_path)
            text = load_prompt(file_path)
            prompt = prompt_template.format(text=text)
            count = 0
            while count < 3:
                try:
                    coro =  chat(prompt)
                    output = asyncio.run(coro)   
                    if os.path.exists(f"./doc_abstract/{file}"):
                        print("abstract file already exists, skip")
                    else:
                        with open(f"./doc_abstract/{file}", "w", encoding="utf-8") as f:
                            f.write(output)             
                    count = 3
                except Exception as e:
                    print(e)
                    print(">>> process", file_path, "failed, retry", count)
                    count = count + 1
            time.sleep(3)