首先，需要明确大模型的概念，通过百度的文心一言的结果，我们得到如下概念：
大模型的特点
大模型（Large Models） 在人工智能和机器学习领域，特别是深度学习领域，指的是那些具有大量参数、复杂结构和高度计算需求的模型。这些模型通常需要在大规模数据集上进行长时间的训练，以捕捉数据中的复杂模式和关系。大模型的出现和发展，主要得益于计算能力的提升、算法的优化以及大规模数据集的可用性。
大模型的特点可以概括为以下几个方面：
- 参数数量多：大模型通常包含数以亿计甚至更多的参数，这些参数在训练过程中通过优化算法进行调整，以最小化预测误差。
- 计算复杂度高：由于参数数量庞大，大模型在训练和推理过程中的计算复杂度也相应增加。这要求使用高性能的计算资源，如GPU、TPU等，来加速计算过程。
- 数据需求大：为了训练出有效的大模型，通常需要大量的训练数据。这些数据不仅要求数量充足，还要求质量高、覆盖全面，以充分反映数据的真实分布和特性。
- 泛化能力强：大模型通过在大规模数据集上进行训练，能够学习到数据中的通用特征和规律，从而在未见过的数据上也能表现出良好的性能。这种泛化能力是衡量模型好坏的重要指标之一。
- 应用前景广阔：大模型在多个领域都有广泛的应用前景，如自然语言处理、计算机视觉、语音识别等。它们可以用于解决复杂的任务和问题，如机器翻译、图像识别、语音识别等。
大模型的发展也带来了一些挑战和争议。一方面，大模型需要消耗大量的计算资源和时间成本，这限制了其在实际应用中的普及和推广。另一方面，大模型也面临着可解释性、隐私保护等问题，需要研究者们不断探索和解决。
总之，大模型是人工智能和机器学习领域的重要研究方向之一，它们通过利用大规模数据集和强大的计算能力，实现了对复杂任务的精准处理和高效解决。随着技术的不断发展，大模型的应用前景将会更加广阔。
因此，从上面的特点来看，博主研究的DETR类目标检测算法也属于大模型的范畴。
LLM学习路线
这里，我们侧重的是语言大模型的学习，其又名Large Language Model（LLM），其学习路线如下：
这里我们
Transformer结构
下面便是在自然语言处理领域中Transformer的结构，可以看到，其主要部分为Embedding、自（交叉）注意力以及前馈神经网络。
由于博主先前对DETR系列目标检测方法有过研究，因此对Transformer中的自（交叉）注意力计算，前馈神经网络等较为了解，因此，这里我们只对Embedding模块进行介绍：
在自然语言处理中，如何把词转换为向量是一个十分重要的问题：
自然语言单词是离散信号，比如“我”、“ 爱”、“人工智能”。如何把每个离散的单词转换为一个向量？通常情况下，我们可以维护一个如下图所示的查询表。表中每一行都存储了一个特定词语的向量值，每一列的第一个元素都代表着这个词本身，以便于我们进行词和向量的映射（如“我”对应的向量值为 [0.3，0.5，0.7，0.9，-0.2，0.03] ）。给定任何一个或者一组单词，我们都可以通过查询这个excel，实现把单词转换为向量的目的，这个查询（查找one hot 编码）和替换过程称之为Embedding Lookup。
详细的替换过程如下：
经过One-Hot Encoding后，句子“我，爱，人工，智能”就被转换成为了一个形状为 4×5000的张量，记为V。在这个张量里共有4行、5000列，从上到下，每一行分别代表了“我”、“爱”、“人工”、“智能”四个单词的One-Hot Encoding。最后，我们把这个张量
V和另外一个稠密张量W（这个相当于语义关系，是我们要训练的）相乘，其中W张量的形状为5000 × 128（5000表示词表大小，128表示每个词的向量大小）。经过张量乘法，我们就得到了一个4×128的张量，从而完成了把单词表示成向量的目的。
按照上述理解，这个W便是权重参数，即语义关系，即我们要训练的，如下图所示：
那么是如何学的呢，过程如下：
第一步，Self-supervisedPre-training,利用互联网上的序列数据（天然的带有标准答案的数据），利用前面的序列去预测接下来的Token，如果预测错误，则更新参数，通用大模型本质就是在做next token prediction
第二步，Supervised Fine-Tuning，这是进行专有任务训练，要让模型听懂你的问题，这个阶段中我们使用的是人工标注的数据，这个数据包含正确答案，同时也会针对一些错误的，不理想的情况，我们会设计对应的回答。
当然虽然我们给设计了标准答案，但在实际应用中我们希望其能够真正的像人一样，具备灵活性，这也就引出了我们的第三步：
第三步，learning from human feedback，在这个阶段，他不会规定固定的输出值，而是会告诉模型这样的输出结果是好的，那样的输出结果是不好的。即给出人类的偏好
对大模型的思考
根据上面的了解，我们似乎可以认为大模型是依靠大参数量来提升性能，那么我们将来的方向是将模型训练的无限大吗，并非如此，在大模型领域，有一个类似于摩尔定律的规律，即知识密度，根据近五年的发展，从ChatGPT3的上千亿参数，到如今ChatGLM模型的62亿参数，尽管参数量下降许多，但其性能却得到提升，即每八个月的知识密度会提升一倍。
此外，要想让大模型能够有一技之长，即要担任一个角色，那么我们给提供的外部知识库才是最重要的，这也是一个企业的竞争力体现，如客服回答记录等。
Agent（智能体）
大模型 Agent，作为一种人工智能体，是具备环境感知能力、自主理解、决策制定及执行行动能力的智能实体。简而言之，它是构建于大模型之上的计算机程序，能够模拟独立思考过程，灵活调用各类工具，逐步达成预设目标的智能存在。
上面说了这么多，其实就是一句话，Agent是能够完成特定任务的应用程序，这里我们举一个简单的例子即可理解Agent到底是什么了。
比如，我是一个商店的管理者，我想采购一批商品，这是如果我想ChatGPT大模型发出这样的提问：
可以看到，虽然ChatGPT给出了结果，但都是一些模式化的步骤，到最后还是需要我们自己去判断决策。
那么Agent是如何做的呢？
我们可以看到当我们发出购入商品的promot时，大模型其实给出了一个个步骤，这里面的步骤即可看作一个个小目标，我们将其看作是一个思考的过程，即Chain of thoughts，这体现的便是Agent的规划能力（Planning）在购入商品时，我们需要分析出某个商品的销售情况，本店商品的库存等等，这些就需要我们从外部去获取，如执行查询销售的API，查询库存的SQL语句之类的，该模块即工具（Tools），而这些工具需要被执行（Action），最终Agent会对执行结果进行约束封装，即将其设计为固定的Promot形式，从而再次输入到大模型中，进而再次让大模型去规划，这就是Reflection（反馈，反射）与Self-critics（自我批判），同时在这个过程中可能需要多轮对话，这就需要Agent具有存储记忆的能力（Memory）来存储对话信息。
Agent案例部署
在这里我们通过部署一个Agent的方式来理解Agent的用法，我们选择使用阿里云的通义千问的模型
下面我们来看看如何实现利用这个大模型来实现Agent
首先是Agent入口，由于我们需要多次与大模型的交互，所以使用循环来不断提问，即多次发生query
,同时我们不希望模型对于一个问题重复思考太多次，因此设置max_request_time来进行限制，agent_execute即具体允许智能体的过程。
def main():
"""支持用户的多次需要输入和交互"""
max_request_time = 10
while True:
query = input("请输入您的目标:")
if query == "exit":
return
agent_execute(query, max_request_time=max_request_time)
if __name__ == '__main__':
# input = "请为我制定一个理财计划"
main()
agent_execute方法代码如下：
def agent_execute(query, max_request_time):
cur_request_time = 0
chat_history = []
agent_scratch = ""
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('********* {}.开始调用大模型.....'.format(cur_request_time))
end_time = time.time()
"""
# call_llm
1、sys_prompt
2、user_prompt
3、history
"""
response = mp.chat(prompt, chat_history)
print('结束调用{}次,花费时间:{}'.format(cur_request_time, end_time-start_time))
# 大模型输出结果的处理
if not response or not isinstance(response, dict):
print("call llm exception, response is :{}".format(response))
continue
"""
response:
{
"action": {
"name": "action_name",
"args": {
"args name": "args value"
}
},
"thoughts":{
"text": "thought",
"plan": "plan",
"criticism": "criticism",
"speak": "当前步骤，返回给用户的总结",
"reasoning": ""
}
}
"""
# 这里统一叫tools #
action_info = response.get("action")
action_name = action_info.get("name")
action_args = action_info.get("args")
print("当前action_name:{}||action_入参:{}".format(action_name, action_args))
# 其他输出信息
thoughts = response.get("thoughts")
plan = thoughts.get("plan")
reasoning = thoughts.get("reasoning")
criticism = thoughts.get("criticism")
observation = thoughts.get("speak")
print("observation:{}".format(observation))
print("plan:{}".format(plan))
print("reasoning:{}".format(reasoning))
print("criticism:{}".format(criticism))
if action_name == "finish":
# 最终将结果返回给用户
final_answer = action_args.get("answer")
print("final_answer:{}".format(final_answer))
break
# speak
observation = response.get("observation")
try:
"""action-name到函数的映射 map -> {"action_name":func}"""
# tools_map = {}
# 获得函数然后直接调用,获得函数的结果
func = tools_map.get(action_name)
call_function_result = func(**action_args)
except Exception as e:
print("调用工具异常:{}".format(e))
call_function_result = "{}".format(e)
agent_scratch = agent_scratch + "\n: observation:{}\n execute action result: {}".format(observation,
call_function_result)
# 从response 中拿出来想要用的信息
assistant_msg = parse_thoughts(response)
chat_history.append([user_prompt, assistant_msg])
if cur_request_time == max_request_time:
print("本次任务执行失败!")
else:
print("本次任务成功！")
首次传入query时：
随后通过prompt = gen_prompt(query, agent_scratch)来对query进行封装，构成promot，过程如下：
promot工程
这里会对我们的提问内容进行封装，同时告诉模型能够做的操作，返回的数据格式等
from tools import gen_tools_desc
"""
prompt包含的功能:
1、任务的描述
2、工具的描述
3、用户的输入user_msg:
4、assistant_msg:
5、结果的限制
6、给出更好实践的描述
"""
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": {
"args name": "args value"
}
},
"thoughts":{
"plan": "简单的描述短期和长期的计划列表",
"criticism": "建设性的自我批评",
"speak": "当前步骤，返回给用户的总结",
"reasoning": "推理"
},
"observation": "观察当前任务的整体进度"
}
"""
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):
"""
:param query:
:param agent_scratch:
:return:
"""
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模式进行响应："
经过构建后的promot如下，看到这里我想大家已经明白了这个Agent到底是如何做的了，因为我们使用的模型是智能的，它能够理解我们的意思，而我们虽然只发出了一个提问，但在promot时我们需要完成大量对promot的封装，这里面包含允许大模型所调用的工具，返回结果的格式等，这个过程也就是promot工程。
你是一个问答专家，你必须始终独立做出决策，无需寻求用户的帮助，发挥你作为LLM的优势，追求简答的策略，不要涉及法律的问题。
目标:
请问如何做一份土豆炖牛腩
限制条件说明:
1.仅使用下面列出的动作
2.你只能主动行动，在计划行动时需要考虑这一点
3.你无法与物理对象交互,如果对于完成任务或目标是绝对必要，则必须要求用户为你完成，如果用户拒绝，并且没有办法实现目标，则直接终止，避免浪费时间和精力。
动作说明:这是你唯一可使用的动作，你的任何操作都必须通过以下操作实现：
1.read_file:read file form 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.finish:完成用户目标, args: [{"name": "answer", "description": "最后的目标结果", "type": "string"}]
5.search:this is a search engine, you can gain additional knowledge though this search engine when you are unsure of large model return, args: [{"name": "query", "description": "search query to look up", "type": "string"}]
资源说明:
1.提供搜索和信息收集的互联网接入
2.读取和写入文件的能力
3.你是一个大语言模型，接受了大量文本的训练，包括大量的事实知识，利用这些知识避免不必要的信息收集
最佳实践的说明:
1.不断地回顾和分析你的行为，确保发挥你最大的能力
2.不断地进行建设性的自我批评
3.反思你过去的决策和策略，完善你的方案
4.每个动作执行都有代价，所以要聪明高效，目的是用最少的步骤完成任务
5.利用你的信息收集能力来寻找你不知道的信息
agent_scratch:
你应该以json格式响应,响应格式如下:
{
"action": {
"name": "action name",
"args": {
"args name": "args value"
}
},
"thoughts":{
"plan": "简单的描述短期和长期的计划列表",
"criticism": "建设性的自我批评",
"speak": "当前步骤，返回给用户的总结",
"reasoning": "推理"
},
"observation": "观察当前任务的整体进度"
}
确保响应结果可以由python json.loads()成功加载。
待完成promot后，便是将其输入到大模型中了，这里我们使用的是通义千问模型，调用如下：
response = mp.chat(prompt, chat_history)
接入大模型
在模型市场，我们可以选择我们需要的大模型，同时他们为我们提供了接入的API调用方式
from openai import OpenAI
import os
client = OpenAI(
api_key="",# 如果您没有配置环境变量，请在此处用您的API Key进行替换
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # 填写DashScope服务的base_url
)
completion = client.chat.completions.create(
model="qwen-turbo",
messages=[
{'role': 'system', 'content': 'You are a helpful assistant.'},
{'role': 'user', 'content': '你是谁？'}],
)
print(completion.model_dump_json())
也可以使用dashscope中封装好的大模型，这里我们使用的便是这种方法，这个代码只是一个测试
import random
from http import HTTPStatus
from dashscope import Generation
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("DASH_SCOPE_API_KEY")
def call_stream_with_messages():
messages = [
{'role': 'user', 'content': '用萝卜、土豆、茄子做饭，给我个菜谱'}]
responses = Generation.call(
'qwen1.5-110b-chat',
messages=messages,
seed=random.randint(1, 10000), # set the random seed, optional, default to 1234 if not set
result_format='message', # set the result to be "message" format.
stream=True,
output_in_full=True, # get streaming output incrementally.\
api_key=api_key,
)
full_content = ''
for response in responses:
if response.status_code == HTTPStatus.OK:
full_content += response.output.choices[0]['message']['content']
print(response)
else:
print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
response.request_id, response.status_code,
response.code, response.message
))
print('Full content: \n' + full_content)
if __name__ == '__main__':
call_stream_with_messages()
真正完成mp.chat(prompt, chat_history)中的chat方法的代码如下：与上面一样我们需要获取调用模型的API_KEY，模型名称等，同时还会将历史信息传入，这里还对历史信息的角色进行了区分
import os, json
import dashscope
from prompt import user_prompt
from dashscope.api_entities.dashscope_response import Message
class ModelProvider(object):
def __init__(self):
self.api_key = os.environ.get('DASH_SCOPE_API_KEY')
self.model_name = os.environ.get('MODEL_NAME')
self._client = dashscope.Generation()
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="system", content=his[1]))
# 最后1条信息是用户的输入
messages.append(Message(role="user", content=user_prompt))
response = self._client.call(
model=self.model_name,
api_key=self.api_key,
messages=messages
)
# print("response:{}".format(response))
content = json.loads(response["output"]["text"])
return content
except Exception as e:
print("call llm exception:{}".format(e))
return {}
得到的response结果如下，这里我们只需要output: "text"的内容即可，由于我们在此之前已经限制了模型的相应格式形式，因此其结果一目了然。
{"status_code": 200, "request_id": "53876a07-b7c8-93ea-b8b3-823710b7306b", "code": "", "message": "", "output": {"text": "{\n \"action\": {\n \"name\": \"search\",\n \"args\": {\n \"query\": \"如何做一份土豆炖牛腩\"\n }\n },\n \"thoughts\": {\n \"plan\": [\n \"首先，我需要搜索如何制作土豆炖牛腩的步骤，以便为用户提供一个简单易懂的烹饪指南。\",\n \"在获取到相关信息后，我将整理成简洁的步骤，然后通过write_to_file动作记录下来。\",\n \"最后，我将总结这些步骤并使用finish动作告诉用户如何制作土豆炖牛腩。\"\n ],\n \"criticism\": \"在这个过程中，我依赖于搜索功能来获取详细步骤，但需要确保找到的信息是可靠且易于用户操作的。\",\n \"speak\": \"正在查找土豆炖牛腩的制作方法，请稍候。\",\n \"reasoning\": \"由于我不能直接与物理世界交互，搜索网络是获得烹饪方法的最佳途径。之后，我可以指导用户按照这些步骤进行操作。\"\n },\n \"observation\": \"任务开始，尚未获取具体烹饪步骤。\"\n}", "finish_reason": "stop", "choices": null}, "usage": {"input_tokens": 663, "output_tokens": 228, "total_tokens": 891}}
最终的到的response，根据我们对各个方法的描述，模型给出了要下一步执行的任务，即search。
随后便是根据要执行的方法，通过我们设计的任务映射来执行调用工具：
try:
"""action-name到函数的映射 map -> {"action_name":func}"""
# tools_map = {}
# 获得函数然后直接调用,获得函数的结果
func = tools_map.get(action_name)
call_function_result = func(**action_args)
except Exception as e:
print("调用工具异常:{}".format(e))
call_function_result = "{}".format(e)
工具定义
这里我们定义了在线查询、写入、读取以及完成等方法，具体实现如下
import json
import os
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 exit, 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):
f"{filename} not exit, 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', encoding='utf-8') as f:
f.write(content)
return "write content to file success."
def search(query):
"""
:param query:
:return:
"""
daily = TavilySearchResults(max_results=5)
try:
ret = daily.invoke(input=query)
print("搜索结果:{}".format(ret))
print("\n")
content_list = []
"""
# 从哪个网站上获取的内容
ret = [
{
"content": "",
"url": ""
}
]
"""
for obj in ret:
content_list.append(obj["content"])
return "\n".join(content_list)
except Exception as e:
return "search error:{}".format(e)
tools_info = [
{
"name": "read_file",
"description": "read file form 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": "finish",
"description": "完成用户目标",
"args": [
{
"name": "answer",
"type": "string",
"description": "最后的目标结果"
}
]
},
{
"name": "search",
"description": "this is a search engine, you can gain additional knowledge though this search engine "
"when you are unsure of large model return",
"args": [
{
"name": "query",
"type": "string",
"description": "search query to look up"
}
]
}
]
tools_map = {
"read_file": read_file,
"append_to_file": append_to_file,
"write_to_file": write_to_file,
"search": search
}
def gen_tools_desc():
"""
生成工具描述
:return:
"""
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
在这里，我们需要使用一个在线查询工具TavilySearchResults，网站如下，我们需要申请一个api从而完成在线查询功能。
https://app.tavily.com/home
至此，通过将任务拆解后的查询结果再次构造promot后输入到大模型中，经过多次提问，便能够完成相应的任务了。
最终任务结果：
当然这个问题比较简单，但也足以让我们理解Agent的具体流程了
