Python开发环境 python 版本用到3.12.1 版本没有强制要求
支持Function Call 工具的大模型(可以是自己部署的大模型或者第三方模型)。使用第三方模型需要自己申请并获取其API-KEY，代码中还用到了tavily搜索，这个也需要自己去申请。

# 约束constraints = ["仅使用下面列出的动作","你只能主动行动，在计划行动时需要考虑到这一点","你无法与物理对象交互，如果对于完成任务或目标是绝对必要的，则必须要求用户为你完成，如果用户拒绝，并且没有其它方法实现目标，则直接终止，避免浪费时间和精力"]
# 资源resources = ["提供搜索和信息搜集的互联网接入","读取和写入文件的能力","你是一个大语言模型，接受了大量的文本训练，包括大量的事实知识，利用这些知识来避免不必要的信息收集"]
# 最佳实践说明best_practices = ["不断地回顾和分析你的行为，确保发挥出你最大的能力","不断地进行建设性的自我批评","反思过去的决策和策略，完善你的方案","每个动作执行部分代价，所以要聪明高效，目的是用最少的步骤完成任务"]
prompt_template = """你是一个问答专家，你必须始终独立做出决策，无需寻求用户的帮助，发挥你作为LLM的优势，追求简单的策略，不要涉及法律问题任务：{query}限制条件说明：{constraints}动作说明：这是你唯一可以使用的作用，你的任何操作都必须通过以下操作实现:{actions}资源说明: {resources}最佳实践的说明：{best_practices}

agent_scratch:{agent_scratch}你应该只以json格式响应，响应格式如下：{response_format_prompt}确保响应结果可以由python json.loads解析"""
response_format_prompt = """{"action":{"name": "action name","args": {"answer": "任务的最终结果"}},"thoughts": {"plan": "简短的描述短期和长期的计划列表","criticism": "建设性的自我批评","speak": "当前步骤，返回给用户的总结","reasoning": "推理"},"observation": "观察当前任务的整体进度"}"""

"""1. 写文件2. 读文件3. 追加4. 网络搜索"""
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("搜索结果", 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": "完成用户目标","args": [{"name": "answer","type": "string","description": "最后的目标结果"}]}]
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：环境变量的设置工具的引入prompt模板模型的初始化"""
# 初始化模型mp = ModelProvider()
# 解析大模型的响应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 promptexcept Exception as err:print("parse thoughts err: {}".format(err))return "".format(err)
def agent_execute(query, max_request_time=10):cur_request_time = 0# 大模型记忆 包括短期记忆和长期记忆chat_history = []# agent 反思 规划等agent_scratch = ''while cur_request_time < max_request_time:cur_request_time += 1"""如果返回结果达到预期，则直接返回"""# 提示词模板"""prompt包含的功能：1. 任务描述2. 工具描述3. 用户的输入user_msg4. assistant_msg5. 限制6. 给出更好实践的描述"""prompt = gen_prompt(query, agent_scratch)
start_time = time.time()print("--------------------------{}, 开始调用大模型LLM------------".format(cur_request_time), flush=True)
# 调用大模型 直接返回json格式数据"""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("--------------------------{}, 调用大模型結束，耗时：{}---------------".format(cur_request_time, end_time-start_time), flush=True)
if not response or not isinstance(response, dict):print("调用大模型错误, 即将重试: ", response)continue
"""大模型返回格式约定response: {"action":{"name": "action name","args": {"args name": "args values"}},"thoughts": {"text": "thought","plan": "plan","criticism": "criticism","speak": "当前步骤，返回给用户的总结","reasoning": ""}}"""

action_info = response.get("action")action_name = action_info.get("name")action_args = action_info.get("args")print("当前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 到函数的映射, map -> { action_name: func }"""# todo: tools_map 的实现# tools_map = {}func = tools_map.get(action_name)call_func_result = func(**action_args)
#{ action_name: func }except Exception as err:print("调用工具异常: ", err)call_func_result = "{}".format(err)agent_scratch = agent_scratch + "\n:observation:{}\n execute action result: {}".format(observation, call_func_result)
# 由于大模型没有记忆功能，因此需要把大模型之前的输入和输出加入到history中，这也是Agent四大块中的记忆模块 由于目前的业务并不复杂，因此不需要外部记忆模块 如果业务比较复杂可能还需要外部模块进行存储assistant_msg = parse_thoughts(response)
chat_history.append([user_prompt, assistant_msg])
if cur_request_time == max_request_time:print("很遗憾，本次任务失败")else:print("恭喜你，任务完成")
def main(): # 支持用户多次交互 最大规划次数设置为10次 当超过10次时 则说明任务失败 用户可以根据自己的情况进行调整max_request_time = 10while True:query = input("请输入你的目标：")if query == "exit":return
agent_execute(query, max_request_time=max_request_time)

if __name__ == "__main__":main()