Python開(kāi)發(fā)環(huán)境 python 版本用到3.12.1 版本沒(méi)有強(qiáng)制要求


支持Function Call 工具的大模型(可以是自己部署的大模型或者第三方模型)。
使用第三方模型需要自己申請(qǐng)并獲取其API-KEY，代碼中還用到了tavily搜索，這個(gè)也需要自己去申請(qǐng)。

# 約束
constraints = [
    "僅使用下面列出的動(dòng)作",
    "你只能主動(dòng)行動(dòng)，在計(jì)劃行動(dòng)時(shí)需要考慮到這一點(diǎn)",
    "你無(wú)法與物理對(duì)象交互，如果對(duì)于完成任務(wù)或目標(biāo)是絕對(duì)必要的，則必須要求用戶為你完成，如果用戶拒絕，并且沒(méi)有其它方法實(shí)現(xiàn)目標(biāo)，則直接終止，避免浪費(fèi)時(shí)間和精力"
]


# 資源
resources = [
    "提供搜索和信息搜集的互聯(lián)網(wǎng)接入",
    "讀取和寫(xiě)入文件的能力",
    "你是一個(gè)大語(yǔ)言模型，接受了大量的文本訓(xùn)練，包括大量的事實(shí)知識(shí)，利用這些知識(shí)來(lái)避免不必要的信息收集"
]


# 最佳實(shí)踐說(shuō)明
best_practices = [
    "不斷地回顧和分析你的行為，確保發(fā)揮出你最大的能力",
    "不斷地進(jìn)行建設(shè)性的自我批評(píng)",
    "反思過(guò)去的決策和策略，完善你的方案",
    "每個(gè)動(dòng)作執(zhí)行部分代價(jià)，所以要聰明高效，目的是用最少的步驟完成任務(wù)"
]


prompt_template = """
    你是一個(gè)問(wèn)答專家，你必須始終獨(dú)立做出決策，無(wú)需尋求用戶的幫助，發(fā)揮你作為L(zhǎng)LM的優(yōu)勢(shì)，追求簡(jiǎn)單的策略，不要涉及法律問(wèn)題
    
任務(wù)：
{query}
    
    
    
限制條件說(shuō)明：
{constraints}
    
動(dòng)作說(shuō)明：這是你唯一可以使用的作用，你的任何操作都必須通過(guò)以下操作實(shí)現(xiàn):
{actions}
    
資源說(shuō)明: 
{resources}
    
最佳實(shí)踐的說(shuō)明：
{best_practices}




agent_scratch:
{agent_scratch}
    
你應(yīng)該只以json格式響應(yīng)，響應(yīng)格式如下：
{response_format_prompt}
確保響應(yīng)結(jié)果可以由python json.loads解析
    
"""


response_format_prompt = """
{
    "action":{
        "name": "action name",
        "args": {
            "answer": "任務(wù)的最終結(jié)果"
        }
    },
    "thoughts": {
        "plan": "簡(jiǎn)短的描述短期和長(zhǎng)期的計(jì)劃列表",
        "criticism": "建設(shè)性的自我批評(píng)",
        "speak": "當(dāng)前步驟，返回給用戶的總結(jié)",
        "reasoning": "推理"
    },
    "observation": "觀察當(dāng)前任務(wù)的整體進(jìn)度"
}
"""

"""
    1. 寫(xiě)文件
    2. 讀文件
    3. 追加
    4. 網(wǎng)絡(luò)搜索
    
"""


def get_workdir_root():
    workdir_root = os.environ.get("WORKDIR_ROOT", './data/llm_result')
    return workdir_root


WORKDIR_ROOT = get_workdir_root()


def read_file(filename):


    if not os.path.exists(filename):
        return f"{filename} not exist, please check file exist before read"


    with open(filename, "r") as f:
        return "\n".join(f.readline())




def append_to_file(filename, content):
    filename = os.path.join(WORKDIR_ROOT, filename)


    if not os.path.exists(filename):
        return f"{filename} not exist, please check file exist before read"


    with open(filename, 'a') as f:
        f.write(content)


    return "append content to file success"


def write_to_file(filename, content):
    filename = os.path.join(WORKDIR_ROOT, filename)


    if not os.path.exists(WORKDIR_ROOT):
        os.makedirs(WORKDIR_ROOT)


    with open(filename, 'w') as f:
        f.write(content)


    return "write content to file success"


def search(query):
    tavily = TavilySearchResults(max_results=5)


    try:
        ret = tavily.invoke(input=query)


        """
            ret:
            [{
                "content": "",
                "url": ""
            }]
        """
        print("搜索結(jié)果", ret)
        content_list = [obj["content"] for obj in ret]
        return "\n".join(content_list)
    except Exception as err:
        return "search err: {}".format(err)


tools_info = [
    {
        "name": "read_file",
        "description": "read file from agent generate, should write file before read.",
        "args": [
            {
                "name": "filename",
                "type": "string",
                "description": "read file name"
            }
        ]
    },
    {
        "name": "append_to_file",
        "description": "append llm content to file, should write file before read.",
        "args": [
            {
                "name": "filename",
                "type": "string",
                "description": "file name"
            },
            {
                "name": "filename",
                "type": "string",
                "description": "append to file content"
            }
        ]
    },
{
        "name": "write_to_file",
        "description": "write llm content to file",
        "args": [
            {
                "name": "filename",
                "type": "string",
                "description": "file name"
            },
            {
                "name": "filename",
                "type": "string",
                "description": "write to file content"
            }
        ]
    },
    {
        "name": "search",
        "description": "this is a search engine, you can gain additional knowledge though this search engine when you are unsure of what large model return",
        "args": [
            {
                "name": "query",
                "type": "string",
                "description": "search query to lookup"
            }
        ]
    },
    {
        "name": "finish",
        "description": "完成用戶目標(biāo)",
        "args": [{
            "name": "answer",
            "type": "string",
            "description": "最后的目標(biāo)結(jié)果"
        }]
    }
]


tools_map = {
    "read_file": read_file,
    "append_to_file": append_to_file,
    "write_to_file": write_to_file,
    "search": search
}


def gen_tools_desc():
    tools_desc = []
    for idx, t in enumerate(tools_info):
        args_desc = []
        for info in t['args']:
            args_desc.append({
                "name": info['name'],
                "description": info["description"],
                "type": info['type']
            })
        args_desc = json.dumps(args_desc, ensure_ascii=False)
        tool_desc = f"{idx + 1}. {t['name']}: {t['description']}, args: {args_desc}"
        tools_desc.append(tool_desc)


    tools_prompt = "\n".join(tools_desc)
    return tools_prompt

"""
todo：
環(huán)境變量的設(shè)置
工具的引入
prompt模板
模型的初始化
"""


# 初始化模型
mp = ModelProvider()


# 解析大模型的響應(yīng)
def parse_thoughts(response):
    try:
        thoughts = response.get("thoughts")
        observation = response.get("observation")


        plan = thoughts.get("plan")
        reasoning = thoughts.get("reasoning")
        criticism = thoughts.get("criticism")
        prompt = f"plan: {plan}\n reasoning: {reasoning}\n criticism: {criticism}\nobservation: {observation}"
        print("thoughts: ", prompt)
        return prompt
    except Exception as err:
        print("parse thoughts err: {}".format(err))
        return "".format(err)


def agent_execute(query, max_request_time=10):
    cur_request_time = 0
    # 大模型記憶 包括短期記憶和長(zhǎng)期記憶
    chat_history = []
    # agent 反思 規(guī)劃等
    agent_scratch = ''
    while cur_request_time < max_request_time:
        cur_request_time += 1
        """
            如果返回結(jié)果達(dá)到預(yù)期，則直接返回
        """
        # 提示詞模板
        """
            prompt包含的功能：
            1. 任務(wù)描述
            2. 工具描述
            3. 用戶的輸入user_msg
            4. assistant_msg
            5. 限制
            6. 給出更好實(shí)踐的描述
        """
        prompt = gen_prompt(query, agent_scratch)


        start_time = time.time()
        print("--------------------------{}, 開(kāi)始調(diào)用大模型LLM------------".format(cur_request_time), flush=True)


        # 調(diào)用大模型 直接返回json格式數(shù)據(jù)
        """
            sys_prompt:
            user_msg, assistant, history
        """


        if cur_request_time < 3:
            print("prompt: ", prompt)


        # response = call_llm()


        response = mp.chat(prompt, chat_history=chat_history)
        end_time = time.time()
        print("--------------------------{}, 調(diào)用大模型結(jié)束，耗時(shí)：{}---------------".format(cur_request_time, end_time-start_time), flush=True)


        if not response or not isinstance(response, dict):
            print("調(diào)用大模型錯(cuò)誤, 即將重試: ", response)
            continue


        """
        大模型返回格式約定
            response: 
            {
                "action":{
                    "name": "action name",
                    "args": {
                        "args name": "args values"
                    }
                },
                "thoughts": {
                    "text": "thought",
                    "plan": "plan",
                    "criticism": "criticism",
                    "speak": "當(dāng)前步驟，返回給用戶的總結(jié)",
                    "reasoning": ""
                }
            }
        """




        action_info = response.get("action")
        action_name = action_info.get("name")
        action_args = action_info.get("args")
        print("當(dāng)前action name: ", action_name, action_args)


        if action_name == "finish":
            final_answer = action_args.get("answer")
            print("final_answer: ", final_answer)
            break


        observation = response.get("observation")


        try:
            """
                action_name 到函數(shù)的映射, map -> { action_name: func }
                
            """
            # todo: tools_map 的實(shí)現(xiàn)
            # tools_map = {}
            func = tools_map.get(action_name)
            call_func_result = func(**action_args)


        #  { action_name: func }
        except Exception as err:
            print("調(diào)用工具異常: ", err)
            call_func_result = "{}".format(err)
        agent_scratch = agent_scratch + "\n:observation:{}\n execute action result: {}".format(observation, call_func_result)


        # 由于大模型沒(méi)有記憶功能，因此需要把大模型之前的輸入和輸出加入到history中，這也是Agent四大塊中的記憶模塊 由于目前的業(yè)務(wù)并不復(fù)雜，因此不需要外部記憶模塊 如果業(yè)務(wù)比較復(fù)雜可能還需要外部模塊進(jìn)行存儲(chǔ)
        assistant_msg = parse_thoughts(response)


        chat_history.append([user_prompt, assistant_msg])


    if cur_request_time == max_request_time:
        print("很遺憾，本次任務(wù)失敗")
    else:
        print("恭喜你，任務(wù)完成")


def main():
     # 支持用戶多次交互 最大規(guī)劃次數(shù)設(shè)置為10次 當(dāng)超過(guò)10次時(shí) 則說(shuō)明任務(wù)失敗 用戶可以根據(jù)自己的情況進(jìn)行調(diào)整
    max_request_time = 10
    while True:
        query = input("請(qǐng)輸入你的目標(biāo)：")
        if query == "exit":
            return


        agent_execute(query, max_request_time=max_request_time)




if __name__ == "__main__":
    main()