大模型实战-动手实现单agent

文章目录

  • 入口cli_main.py
  • 工具tools.py
  • prompt prompt_cn.py
  • LLM 推理 model_provider.py
  • 致谢

image.png

agent 的核心思想:不断调用 LLM(多轮对话),让 LLM 按照指定的格式(例如 json)进行回复,提取 LLM 回复的字段信息执行相应的 action(工具),并把 LLM 每次执行的结果(observation)加入到 LLM 的对话历史中拼接到 prompt 里,作为新一轮的输入。在工具中预设 finsh 工具,告诉模型应该什么时候停止,并获取答案。

入口cli_main.py

# -*- encoding: utf-8 -*-
"""
@author: acedar  
@time: 2024/5/12 10:25
@file: cli_main.py 
"""

import time
from tools import tools_map
from prompt_cn import gen_prompt, user_prompt
from model_provider import ModelProvider
from dotenv import load_dotenv

load_dotenv()

# agent入口

"""
todo:
    1. 环境变量的设置
    2. 工具的引入
    3. prompt模板
    4. 模型的初始化
"""

mp = ModelProvider()


def parse_thoughts(response):
    """
        response:
        {
            "action": {
                "name": "action name",
                "args": {
                    "args name": "args value"
                }
            },
            "thoughts":
            {
                "text": "thought",
                "plan": "plan",
                "criticism": "criticism",
                "speak": "当前步骤,返回给用户的总结",
                "reasoning": ""
            }
        }
    """
    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}\nreasoning:{reasoning}\ncriticism: {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
    chat_history = []
    agent_scratch = ''  # agent思考的内容

    while cur_request_time < max_request_time:
        cur_request_time += 1

        """
        如果返回结果达到预期,则直接返回
        """
        """
        prompt包含的功能:
            1. 任务描述
            2. 工具描述
            3. 用户的输入user_msg
            4. assistant_msg
            5. 限制
            6. 给出更好实践的描述
            
        """
        prompt = gen_prompt(query, agent_scratch)
        start_time = time.time()
        print("*************** {}. 开始调用大模型llm.......".format(cur_request_time), flush=True)
        # call llm
        """
        sys_prompt: 
        user_msg, assistant, history
        """
        if cur_request_time < 3:
            print("prompt:", prompt)
        response = mp.chat(prompt, 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

        """
        规定的LLM返回格式
        response:
        {
            "action": {
                "name": "action name", 对应工具名
                "args": {
                    "args name": "args value" 对应工具参数
                }
            },
            "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)

        # 如果action_name=finish就代表任务完成,action_args.get("answer")返回给用户答案
        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}
            """
            # tools_map的实现
            func = tools_map.get(action_name)
            call_func_result = func(**action_args)

        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)

        assistant_msg = parse_thoughts(response)
        chat_history.append([user_prompt, assistant_msg])
    if cur_request_time == max_request_time:
        print("很遗憾,本次任务失败")
    else:
        print("恭喜你,任务完成")


def main():
    # 需求: 支持用户的多次交互
    max_request_time = 30
    while True:
        query = input("请输入您的目标:")
        if query == "exit":
            return
        agent_execute(query, max_request_time=max_request_time)


if __name__ == "__main__":
    main()

工具tools.py

  • 需要注意的是不符合工具执行条件的信息、报错信息需要一并传给 LLM
  • 在这里获取Tavily AI 的 API:https://docs.tavily.com/docs/gpt-researcher/getting-started,并加入到系统环境变量中
export TAVILY_API_KEY={Your Tavily API Key here}
# -*- encoding: utf-8 -*-
"""
@author: acedar  
@time: 2024/5/12 11:07
@file: tools.py 
"""

import os
import json
from langchain_community.tools.tavily_search import TavilySearchResults

"""
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):
    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, 'r', encoding='utf-8') as f:
        return "\n".join(f.readlines())


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', encoding='utf-8') 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', encoding='utf-8') 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": "content",
            "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": "content",
            "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 look up"
        }]
    },
    {
        "name": "finish",
        "description": "return finish when you get exactly the right answer",
        "args": [{
            "name": "answer",
            "type": "string",
            "description": "the final answer"
        }]
    }
]


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

prompt prompt_cn.py

  • prompt 十分重要,非常影响效果
# -*- encoding: utf-8 -*-
"""
@author: acedar  
@time: 2024/5/12 11:40
@file: prompt.py 
"""

from tools import gen_tools_desc

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": "观察当前任务的整体进度"
}
"""


# todo: query, agent_scratch, actions
action_prompt = gen_tools_desc()
constraints_prompt = "\n".join(
    [f"{idx+1}. {con}" for idx, con in enumerate(constraints)])
resources_prompt = "\n".join(
    [f"{idx+1}. {con}" for idx, con in enumerate(resources)])
best_practices_prompt = "\n".join(
    [f"{idx+1}. {con}" for idx, con in enumerate(best_practices)])


def gen_prompt(query, agent_scratch):
    prompt = prompt_template.format(
        query=query,
        constraints=constraints_prompt,
        actions=action_prompt,
        resources=resources_prompt,
        best_practices=best_practices_prompt,
        agent_scratch=agent_scratch,
        response_format_prompt=response_format_prompt
    )
    return prompt


user_prompt = "根据给定的目标和迄今为止取得的进展,确定下一个要执行的action,并使用前面指定的JSON模式进行响应:"


if __name__ == '__main__':
    prompt = gen_prompt("query", "agent_scratch")
    print(prompt)
'''
    你是一个问答专家,你必须始终独立做出决策,无需寻求用户的帮助,发挥你作为LLM的优势,追求简答的策略,不要涉及法律问题。
    
任务:
query

限制条件说明:
1. 仅使用下面列出的动作
2. 你只能主动行动,在计划行动时需要考虑到这一点
3. 你无法与物理对象交互,如果对于完成任务或目标是绝对必要的,则必须要求用户为你完成,如果用户拒绝,并且没有其他方法实现目标,则直接终止,避免浪费时间和精力。

动作说明: 这是你唯一可以使用的动作,你的任何操作都必须通过以下操作实现:
1. read_file: read file from agent generate, should write file before read, args: [{"name": "filename", "description": "read file name", "type": "string"}]
2. append_to_file: append llm content to file, should write file before read, args: [{"name": "filename", "description": "file name", "type": "string"}, {"name": "content", "description": "append to file content", "type": "string"}]
3. write_to_file: write llm content to file, args: [{"name": "filename", "description": "file name", "type": "string"}, {"name": "content", "description": "write to file content", "type": "string"}]
4. search: 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", "description": "search query to look up", "type": "string"}]
5. finish: return finish when you get exactly the right answer, args: [{"name": "answer", "description": "the final answer", "type": "string"}]

资源说明:
1. 提供搜索和信息收集的互联网接入
2. 读取和写入文件的能力
3. 你是一个大语言模型,接受了大量文本的训练,包括大量的事实知识,利用这些知识来避免不必要的信息收集

最佳实践的说明:
1. 不断地回顾和分析你的行为,确保发挥出你最大的能力
2. 不断地进行建设性的自我批评
3. 反思过去的决策和策略,完善你的方案
4. 每个动作执行都有代价,所以要聪明高效,目的是用最少的步骤完成任务
5. 利用你的信息收集能力来寻找你不知道的信息

agent_scratch:agent_scratch

你应该只以json格式响应,响应格式如下:

{
    "action": {
        "name": "action name",
        "args": {
             "answer": "任务的最终结果"
        }
    },
    "thoughts":
    {
        "plan": "简短的描述短期和长期的计划列表",
        "criticism": "建设性的自我批评",
        "speak": "当前步骤,返回给用户的总结",
        "reasoning": "推理"
    },
    "observation": "观察当前任务的整体进度"
}

确保响应结果可以由python json.loads解析
'''

LLM 推理 model_provider.py

  • 这里以阿里云百炼为例https://bailian.console.aliyun.com/?spm=5176.29228872.J_TC9GqcHi2edq9zUs9ZsDQ.1.74cd38b1e2yJWL#/model-market
  • 需要填入自己 key
# -*- encoding: utf-8 -*-
"""
@author: acedar  
@time: 2024/5/12 12:30
@file: model_provider.py 
"""

import os
import json
import dashscope
from dashscope.api_entities.dashscope_response import Message
from prompt_cn import user_prompt


class ModelProvider(object):
    def __init__(self):
        self.api_key = os.environ.get(
            "API_KEY", '')
        self.model_name = os.environ.get(
            "MODEL_NAME", default='qwen-max')
        self._client = dashscope.Generation()
        print("model_name:", self.model_name)
        self.max_retry_time = 3

    def chat(self, prompt, chat_history):
        cur_retry_time = 0
        while cur_retry_time < self.max_retry_time:
            cur_retry_time += 1
            try:
                messages = [Message(role='system', content=prompt)]
                for his in chat_history:
                    messages.append(Message(role='user', content=his[0]))
                    messages.append(Message(role='assistant', content=his[1]))
                messages.append(Message(role='user', content=user_prompt))
                response = self._client.call(
                    model=self.model_name,
                    api_key=self.api_key,
                    messages=messages
                )
                """
                {
                    "status_code": 200,
                     "request_id": "c965bd27-c89c-9b5c-924d-2f1688e8041e", 
                     "code": "", 
                     "message": "", 
                     "output": {
                        "text": null, "finish_reason": null,
                         "choices": [{
                            "finish_reason": "null", "message": 
                            {"role": "assistant", "content": "当然可以,这里有一个简单又美味"}
                        }]
                    }, 
                    "usage": {
                        "input_tokens": 31, 
                        "output_tokens": 8, 
                        "total_tokens": 39, 
                        "plugins": {}
                    }
                }
                """
                print("response:", response)

                content = json.loads(response['output']['text'])
                return content
            except Exception as err:
                print("调用大模型出错:{}".format(err))
            return {}

致谢

https://gitee.com/open-llm/llm-agent
https://www.bilibili.com/video/BV1Sz421m7Rr/

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/654117.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【LaTex】11 ACM参考文献顺序引用 - 解决 ACM-Reference-Format 顺序不符合论文实际引用顺序的问题

【LaTex】11 ACM参考文献顺序引用 写在最前面解决 ACM-Reference-Format 顺序不符合论文实际引用顺序的问题问题描述问题原因如何解决问题解决方案1&#xff08;更简单&#xff09;解决方案2&#xff08;更自由&#xff09; 小结 &#x1f308;你好呀&#xff01;我是 是Yu欸 …

巧用java8的stream流的.collect(Collectors.toMap(arg1,arg2))

最近公司接手了一个低代码二次开发平台的需求&#xff0c;需要连接多张表的数据然后展示到界面上。 按照java的sql思路&#xff0c;我们直接通过left join去关联表就行了&#xff0c;但是该低代码平台有对sql连表查询有限制&#xff0c;就是有些表它是存在一个domainKey的&…

牛客题霸-SQL大厂面试真题(一)

本文基于前段时间学习总结的 MySQL 相关的查询语法&#xff0c;在牛客网找了相应的 MySQL 题目进行练习&#xff0c;以便加强对于 MySQL 查询语法的理解和应用。 由于涉及到的数据库表较多&#xff0c;因此本文不再展示&#xff0c;只提供 MySQL 代码与示例输出。 以下内容是…

Java | Leetcode Java题解之第101题对称二叉树

题目&#xff1a; 题解&#xff1a; class Solution {public boolean isSymmetric(TreeNode root) {return check(root, root);}public boolean check(TreeNode u, TreeNode v) {Queue<TreeNode> q new LinkedList<TreeNode>();q.offer(u);q.offer(v);while (!q.…

大数据框架总结(全)

☔️ 大数据框架总结&#xff08;全&#xff09; 关注“大数据领航员”&#xff0c;在公众号号中回复关键字【大数据面试资料】&#xff0c;即可可获取2024最新大数据面试资料的pdf文件 一. Hadoop HDFS读流程和写流程 HDFS写数据流程 &#xff08;1&#xff09;客户端通过…

【ARM+Codesys案例】T3/RK3568/树莓派+Codesys锂电池测试设备控制解决方案

锂电池诞生于上世纪60年代&#xff0c;90年代开始由日本索尼公司实现商业化。锂离子电池凭借快速充放电、长循环寿命、无记忆效应等众多优点&#xff0c;成为当今数码产品及电动汽车大规模应用的第一选择。与镍氢电池、铅酸电池相比&#xff0c;锂电池可以存储更多电能。现在&a…

ECOLOGY9解决明细表数据按需显示或隐藏

ECOLOGY9:由于领导需要查看完整的明细表数据&#xff0c;没有启用分页功能&#xff0c;导致大领导查看时又出现几百行的数据&#xff0c;影响使用体验。需求&#xff1a;只看正常表单数据&#xff0c;明细表数据&#xff0c;默认不显示&#xff0c;需要查看时再打开。 实现&am…

抖店怎么选品?抖店爆款选品思路技巧,新手直接用!

大家好&#xff0c;我是电商花花。 抖店选品永远是我们做抖店&#xff0c;做电商的核心&#xff0c;店铺想要出单&#xff0c;想要赚钱&#xff0c;我们就一定要学会怎么选品&#xff0c;怎么筛选商品。 而我们绝大多数新手并没有办法保证持续选爆款的能力&#xff0c;如果店…

Linux:confluence8.5.9的部署(下载+安装+破ji)离线部署全流程

0.环境 Confluence也是比较吃运存了&#xff0c;我建议运行运存给到4g或者4g以上就可以了&#xff0c;核数可以给到1核或以上 我部署在centos7.9操作系统上&#xff0c;ip地址为&#xff1a;192.168.6.1&#xff0c;yum仓库使用的是自己的镜像文件&#xff0c;本章所有使用到…

iptables防火墙【☆】

一、防火墙的基础 防火墙分为硬件防火墙和软件防火墙&#xff0c;硬件防火墙有专门的设备&#xff1a;如国产华为、绿盟等&#xff0c;作为网关层做地址转换等安全防护。很多公司基本都会使用硬件防火墙做第一道防御&#xff0c;在核心业务上再添加软件防火墙提高安全性能…

逆向基础:软件手动脱壳技术入门

这里整合了一下之前自己学习软件手工脱壳的一些笔记和脱文&#xff0c;希望能给新学软件逆向和脱壳的童鞋们一点帮助。 1 一些概念 1.1 加壳 加壳的全称应该是可执行程序资源压缩&#xff0c;是保护文件的常用手段。加壳过的程序可以直接运行&#xff0c;但是不能查看源代码…

MySQL详细安装、配置过程,多图,详解

本文适合centos7环境下安装mysql&#xff0c;在安装和卸载过程中&#xff0c;都在root用户下完成。文章目录 清理环境获取mysql官方yum源安装mysql yum源安装mysql服务安装报错解决办法验证是否安装完成启动mysql服务登录服务方法一&#xff1a;方法二&#xff1a;方法三&#…

huawei 华为 交换机 配置 Dot1q 终结子接口实现跨设备 VLAN 间通信示例

组网需求 如图7-8所示&#xff0c;SwitchA和SwitchB分别下挂VLAN 10和VLAN 20的二层网络&#xff0c;SwitchA和SwitchB之间通过三层网络互通&#xff0c;三层网络采用OSPF协议。要求两个二层网络的PC实现二层隔离三层互通。 图 7-8 配置 Dot1q 终结子接口实现跨设备 VLAN 间通信…

Vue3实战笔记(37)—粒子特效登录页面

文章目录 前言一、粒子特效登录页总结 前言 上头了&#xff0c;再来一个粒子特效登录页面。 一、粒子特效登录页 登录页&#xff1a; <template><div><vue-particles id"tsparticles" particles-loaded"particlesLoaded" :options"…

Spring系列-02-Bean类型, 作用域, 实例化, 生命周期

Bean类型, 作用域, 实例化, 生命周期 Bean类型 在 SpringFramework 中,对于 Bean 的类型,一般有两种设计: 普通BeanFactoryBean 普通Bean 三种方式 Component注解配置类Beanxml <bean> Component public class Child {}Bean public Child child() {return new Ch…

Postgresql源码(134)优化器针对volatile函数的排序优化分析

相关 《Postgresql源码&#xff08;133&#xff09;优化器动态规划生成连接路径的实例分析》 上一篇对路径的生成进行了分析&#xff0c;通过make_one_rel最终拿到了一个带着路径的RelOptInfo。本篇针对带volatile函数的排序场景继续分析subquery_planner的后续流程。 subquer…

国内信创web中间件生态

国内信创web中间件生态 东方通 官网https://www.tongtech.com/pctype/25.html 宝蓝德 官网https://www.bessystem.com/product/0ad9b8c4d6af462b8d15723a5f25a87d/info?p101 金蝶天燕 官网 https://www.apusic.com/list-117.html 中创 官网http://www.inforbus.com…

小程序使用vant组件库

一:下载组件库 在小程序内npm下载的包 vant组件库官网:快速上手 - Vant Weapp (youzan.github.io) 1)首先有有package.json文件,没有的话则先初始化 即使通过package.json去下载包,也需要有,可以观察下载的包. 2)下载包 3)构建npm包 下载包之后存储在node_modules内,但是我们…

详谈 Java中的list.forEach()和list.stream().forEach() 异同点

涉及的文章链接&#xff1a;ArrayList 循环Remove遇到的坑 一、想总结本篇博客的原因 在日常开发中&#xff0c;需要对集合数据进行或多或少的赋值修改&#xff0c;那么循环赋值或者做一些处理就是最常见的一种操作了&#xff0c;但是用习惯了stream流&#xff0c;所以在循环的…

HQChart使用教程99-K线窗口设置上下间距

HQChart使用教程99-K线窗口设置上下预留间距 指标窗口布局说明设置预留间距数据结构通过Setoption设置通过ChangeIndex设置 HQChart代码地址 指标窗口布局说明 顶部预留间距(3)和底部预留间距(5) 这个部分是算在Y轴坐标上的 设置预留间距 数据结构 HorizontalReserved&#…