Assistant与ReACT探究
最近OpenAI发布了Assistants API，允许在自己的应用程序中构建 AI 助手。Assistants API 目前支持三种类型的工具：Code Interpreter、Knowledge Retrieval、Function calling，还有对File的支持，其实网上已经有很多探讨Assistants API的使用方式的文章，主要参考API: https://platform.openai.com/docs/assistants/overview ，个人觉得这套工具其实本质实现了一个简单的ReAct模型，并且通过Assistants提供的 File、Function Call等能力，更好的与自己的应用程序进行交互，只不过目前还是相当不完备的，很多时候使用Assistants并不能一劳永逸。
关于Assistants的线程的概念，还有Messages等不再重复，因为这些官网都有比较详细的介绍。如果你使用的编程语言不是Python，也不是其他非常主流和常见的编程语言，那么直接用OpenAI提供的HTTP的接口效果是一样的，然后封装一下，开源共建一个API库。
看到官网有这张图，其实还是很容易理解的吧，Assistant其实就是一个带ReAct（行为反思链）的机器人，通过创建Thread的方式，开启与Assistant的交互之旅，在一个线程中，Assistant会处理你的消息，并且根据上下文分析出此时需要调用哪些Function Call，并且何时需要生成代码并且运行Code Interpreter生成的代码：
我们通过轮询一个线程的状态来判断此时应该做什么，且看下面的生命周期：
当首次创建并且运行的时候，转至排队状态，几乎立马会转到处理中的状态，我们需要关注的其实也就是处理中需要调用函数的时候（也就是Function Call），以及处理结果即可（这里结果就包含了完成、超时、失败、取消等状态）。如何轮询呢？提供了接口（官网有说明），其实大部分情况下1/2s一次就完全足够了，短期内多次轮询没有意义：
1https://api.openai.com/v1/threads/runs
创建线程，创建消息，Run起来都可以参考官方文档。
创建助手
https://platform.openai.com/docs/assistants/how-it-works/creating-assistants
1assistant = client.beta.assistants.create(
2 name="Data visualizer",
3 description="You are great at creating beautiful data visualizations. You analyze data present in .csv files, understand trends, and come up with data visualizations relevant to those trends. You also share a brief text summary of the trends observed.",
4 model="gpt-4-1106-preview",
5 tools=[{"type": "code_interpreter"}],
6 file_ids=[file.id]
7)
指定名称、描述、使用的LLM、工具（包括code_interpreter和你的自定义Function）
其实在playground页面创建也一样，都可以通过API来管理Assistant，以及更新Assistant。对于其他API也是一样的，Python的Assistant API库也是调用HTTP接口。
这里主要说一下如何定义工具，其实参考官方给的天气工具的例子就能写一个类似的，我常用的一个工具就是获得当前的时间，因为作为语言大模型，他不能已知当前的时间，需要我们告诉LLM，这个时候我们可以定义一个工具来做这个事情，让LLM在需要的时候自己去调用就知道当前时间：
1{
2 "type": "function",
3 "function": {
4 "name": "GetNowTime",
5 "description": "获取当前时间",
6 "parameters": {
7 "type": "object",
8 "properties": {},
9 "required": []
10 },
11 }
12}
这样的话，在创建Assistant的时候就把这个工具描述传递进去，由大模型自行决定何时调用：
1my_assistant = client.beta.assistants.create(
2 name=assistants_name,
3 instructions="我的数据分析助手",
4 model="gpt-4-1106-preview",
5 tools=[
6 {"type": "code_interpreter"},
7 {
8 "type": "function",
9 "function": {
10 "name": "GetNowTime",
11 "description": "获取当前时间",
12 "parameters": {
13 "type": "object",
14 "properties": {},
15 "required": []
16 }
17 }
18 }
19 ],
20 file_ids = []
21)
同时我们需要创建一个真实的Function，在大模型需要的时候去调用：
同时里面还有一个file_ids的参数，其实就是把一个file挂在一个Assistant下，大多数情况下当作知识库，或者数据文件与Assistant建立关系。
创建线程和消息
https://platform.openai.com/docs/assistants/how-it-works/managing-threads-and-messages
1thread = client.beta.threads.create(
2 messages=[
3 {
4 "role": "user",
5 "content": "Create 3 data visualizations based on the trends in this file.",
6 "file_ids": [file.id]
7 }
8 ]
9)
这个没啥好说的，可以在消息里携带图片或者数据文件。 有时候返回的消息有注释，目前有两种类型的注释：
- file_citation：文件引用是由检索工具创建的，它定义了对上传并由助手用于生成响应的特定文件中特定引用位置的引用。
- file_path：文件路径注释是由代码解释器工具创建，并包含对该工具生成的文件的引用。当消息对象中存在注释时，您将看到在文本中出现无法理解模型生成子字符串，您应该使用这些注释替换这些字符串。
1# Retrieve the message object
2message = client.beta.threads.messages.retrieve(
3 thread_id="...",
4 message_id="..."
5)
6
7# Extract the message content
8message_content = message.content[0].text
9annotations = message_content.annotations
10citations = []
11
12# Iterate over the annotations and add footnotes
13for index, annotation in enumerate(annotations):
14 # Replace the text with a footnote
15 message_content.value = message_content.value.replace(annotation.text, f' [{index}]')
16
17 # Gather citations based on annotation attributes
18 if (file_citation := getattr(annotation, 'file_citation', None)):
19 cited_file = client.files.retrieve(file_citation.file_id)
20 citations.append(f'[{index}] {file_citation.quote} from {cited_file.filename}')
21 elif (file_path := getattr(annotation, 'file_path', None)):
22 cited_file = client.files.retrieve(file_path.file_id)
23 citations.append(f'[{index}] Click <here> to download {cited_file.filename}')
24 # Note: File download functionality not implemented above for brevity
25
26# Add footnotes to the end of the message before displaying to user
27message_content.value += '\n' + '\n'.join(citations)
RUN起来
https://platform.openai.com/docs/assistants/how-it-works/runs-and-run-steps
轮训执行步骤，这块也是最关键的是监测到Function Call，并且执行后回复给Assistant结果：
1while True:
2 # 查询消息的状态
3 re_run = client.beta.threads.runs.retrieve(
4 thread_id=thread_id,
5 run_id=run.id
6 )
7 # "queued", "in_progress", "requires_action", "cancelling", "cancelled", "failed", "completed",
8 # "expired" 排队中, 进行中, 需要采取行动, 取消中, 已取消, 失败, 完成, 过期 如果状态完成，则获取结果，break
9 if re_run.status == "completed":
10 # 跑完了，获取全部消息，或者只获取最后的结果
11 messages = client.beta.threads.messages.list(thread_id=thread_id)
12 break
13 else:
14 run_step_dict = {}
15 if re_run.status == "requires_action":
16 # 请求工具
17 tool_calls = re_run.required_action.submit_tool_outputs.tool_calls
18 call_ids = []
19 outputs = []
20 for tool_call in tool_calls:
21 if tool_call.type == 'function':
22 # 开始调用函数
23 function_name = tool_call.function.name
24 tool_calls_id = tool_call.id
25 call_ids.append(tool_calls_id)
26 function_params = json.loads(tool_call.function.arguments)
27 if function_name == 'GetNowTime':
28 current_app.logger.info('Function Call --> GetNowTime()')
29 # 调用函数(核心就是这里啦！)
30 # 调用函数(核心就是这里啦！)
31 # 调用函数(核心就是这里啦！)
32 result = get_now_time()
33 current_app.logger.info('Function Call Result --> GetNowTime() :%s', str(result))
34 outputs.append(str(result))
35 tool_outputs = []
36 for i in range(len(call_ids)):
37 tool_outputs.append({
38 "tool_call_id": call_ids[i],
39 "output": outputs[i],
40 })
41 # 提交工具的输出
42 client.beta.threads.runs.submit_tool_outputs(
43 thread_id=thread_id,
44 run_id=run.id,
45 tool_outputs=tool_outputs
46 )
47 time.sleep(1)
如果你想快速验证你的 Function 能否被准确的调用到，直接在Playground面板进行调试是最好的选择：
从官网可以看到，后面OpenAI会逐步让这个工具变的更好用，包括以下的优化：
- 支持流式输出
- 以通知的方式更新，避免轮询，可能是SSE或者Websocket
- 支持以图片作为用户消息
搭建OpenAI的API代理
可以使用Nginx来搭建一个OpenAI API的代理服务，首先确保Nginx的环境可以访问到OpenAI API，然后采用如下配置：
1server {
2 listen 80; # 监听80端口，用于HTTP请求
3 location / {
4 proxy_pass https://api.openai.com/; # 反向代理到https://api.openai.com/这个地址
5 proxy_ssl_server_name on; # 开启代理SSL服务器名称验证，确保SSL连接的安全性
6 proxy_set_header Host api.openai.com; # 设置代理请求头中的Host字段为api.openai.com
7 chunked_transfer_encoding off; # 禁用分块编码传输，避免可能的代理问题
8 proxy_buffering off; # 禁用代理缓存，避免数据传输延迟
9 proxy_cache off; # 禁用代理缓存，确保实时获取最新的数据
10 proxy_set_header X-Forwarded-For $remote_addr; # 将客户端真实IP添加到代理请求头中的X-Forwarded-For字段中，用于记录客户端真实IP
11 }
12}
使用的时候需要设置OpenAI的BaseURL(0.27.x)
1import openai
2openai.api_key = os.environ.get("OPENAI_API_KEY")
3openai.api_base = "your_proxy_url" # 代理地址,如“http://www.test.com/v1”
新版本（>1.2.x）OpenAI API设置BaseURL
1from openai import OpenAI
2
3client = OpenAI(
4 api_key=os.environ.get("OPENAI_API_KEY"),
5 base_url="your_proxy_url" # 代理地址,如“http://www.test.com/v1”
6)
附上完整的nginx.conf与docker-compose.yml文件，以供参考： nginx.conf
1user nginx;
2worker_processes auto;
3
4error_log /var/log/nginx/error.log notice;
5pid /var/run/nginx.pid;
6
7
8events {
9 worker_connections 1024;
10}
11
12
13http {
14 include /etc/nginx/mime.types;
15 default_type application/octet-stream;
16
17 log_format main '$remote_addr - $remote_user [$time_local] "$request" '
18 '$status $body_bytes_sent "$http_referer" '
19 '"$http_user_agent" "$http_x_forwarded_for"';
20
21 access_log /var/log/nginx/access.log main;
22
23 sendfile on;
24 #tcp_nopush on;
25
26 keepalive_timeout 65;
27
28 #gzip on;
29
30 # include /etc/nginx/conf.d/*.conf;
31
32 server {
33 listen 80;
34 location / {
35 proxy_pass https://api.openai.com/;
36 proxy_ssl_server_name on;
37 proxy_set_header Host api.openai.com;
38 chunked_transfer_encoding off;
39 proxy_buffering off;
40 proxy_cache off;
41 proxy_set_header X-Forwarded-For $remote_addr;
42 }
43 }
44}
docker-compose.yaml
1services:
2 clash:
3 image: dreamacro/clash-premium:latest
4 volumes:
5 - "$PWD/TAG.yaml:/root/.config/clash/config.yml"
6 ports:
7 - "7890:7890"
8 - "7891:7891"
9 - "7892:7892"
10 restart: unless-stopped
11 proxy:
12 image: nginx
13 volumes:
14 - "$PWD/nginx.conf:/etc/nginx/nginx.conf"
15 ports:
16 - "80:80"
17 - "443:443"
18 restart: unless-stopped
附上我搭建的一个代理服务：
https://openai-proxy.zouchanglin.cn
Agent的原理——ReAct
ReAct其实就是Reasoning + Acting构成的一套反思、行动的思维链模型，不管是LangChain，还是AutoGen，其中比较核心的一个模块就是Agent。Agent的本质就是利用LLM去实现Reasoning + Acting的思维模式，从而完成复杂的任务。
下面直接动手实现一个这样的思维链Demo：
首先定义一些工具或者叫做函数：
1[
2 {
3 "name":"GetTimeTools",
4 "description":"获取当前时间",
5 "parameters":[],
6 "return":{
7 "type":"datetime",
8 "description":"时间"
9 }
10 },
11 {
12 "name":"AskHumanHelpTool",
13 "description":"如果需要人类帮助，请使用它",
14 "parameters":[
15 {
16 "type":"string",
17 "name":"answer",
18 "description":"你需要请求人类帮助的问题"
19 }
20 ]
21 },
22 {
23 "name":"TaskCompleteTool",
24 "description":"如果你认为你已经有了最终答案，请使用它",
25 "parameters":[
26 {
27 "name":"output",
28 "type":"string",
29 "description":"输出你的最终答案"
30 }
31 ]
32 },
33 {
34 "name":"CodeInterpreterTool",
35 "description":"如要执行与环境无关的Python代码，则执行的结果用print打出来，请使用它",
36 "parameters":[
37 {
38 "name":"code",
39 "type":"string",
40 "description":"你要执行的代码"
41 }
42 ]
43 }
44]
我这里准备了4个Tools以供LLM选择和调用，description用来定义这个工具的功能。
核心Prompt：
1core_prompt = '''You are an assistant, please fully understand the user's question, choose the appropriate tool, and \
2help the user solve the problem step by step.
3
4### CONSTRAINTS ####
51. The tool selected must be one of the tools in the tool list.
62. When unable to find the input for the tool, please adjust immediately and use the AskHumanHelpTool to ask the user \
7for additional parameters.
83. When you believe that you have the final answer and can respond to the user, please use the TaskCompleteTool.
95. You must response in Chinese.
10
11### Tool List ###
12%s
13
14
15### Tool respond format ###
16{
17 "tool_name": "tool name",
18 "response": "tool response content"
19}
20
21
22You should only respond in JSON format as described below
23
24### RESPONSE FORMAT ###
25{"thought": "Reasons and thought process for choosing these tools","tool_name": "tool_name","args_list": {"args_name_1\
26": "args_value_1","args_name_2": "args_value_2"}}
27Make sure that the response content you return is all in JSON format and does not contain any extra content.'''
28
这段Prompt其实就是Agent的核心了：其中包括了LLM的角色与任务，可调用的工具，每次返回的JSON格式等。接下来就是考验LLM分析问题的能力了：
1import json
2from datetime import datetime
3
4from openai_client import client
5from react import CORE_PROMPT
6from react.react_enum import FunctionCallEnum
7from tools.time_tools import get_now
8
9
10def load_tools():
11 with open('core_function.json', 'r', encoding='utf-8') as f:
12 try:
13 content = f.read()
14 tools = json.loads(content)
15 return core_prompt % content
16 except json.JSONDecodeError as e:
17 raise Exception('core_function.json文件格式错误，请检查！')
18
19
20def function_call(tool_name: str, messages: list[dict]):
21 if tool_name == FunctionCallEnum.GetTimeTools.name:
22 func_call_message = {
23 "name": FunctionCallEnum.GetTimeTools.name,
24 "response": get_now().strftime('%Y-%m-%d %H:%M:%S')
25 }
26 messages.append({
27 "role": "user",
28 "content": json.dumps(func_call_message)
29 })
30 else:
31 print('不支持的工具:' + tool_name)
32
33
34def exec_python_code_in_sandbox(llm_response, messages: list[dict]):
35 print('执行Python代码', llm_response)
36 code = llm_response['args_list']['code']
37 # 保存在文件中
38 with open('tmp.py', 'w', encoding='utf-8') as f:
39 f.write(code)
40 # shell方式执行获得返回值
41 import subprocess
42 result = subprocess.run(['python', 'tmp.py'], stdout=subprocess.PIPE)
43 print(result.stdout.decode('utf-8'))
44 # 返回结果
45 func_call_message = {
46 "tool_name": FunctionCallEnum.CodeInterpreterTool.name,
47 "response": result.stdout.decode('utf-8')
48 }
49 messages.append({
50 "role": "user",
51 "content": json.dumps(func_call_message)
52 })
53 # 删除文件
54 import os
55 os.remove('tmp.py')
56
57
58
59def start_react(user_input: str):
60 messages = [
61 {
62 "role": "system",
63 "content": CORE_PROMPT
64 },
65 {
66 "role": "user",
67 "content": user_input
68 }
69 ]
70 handle_count = 0
71 while True:
72 completion = client.chat.completions.create(
73 model="gpt-4-1106-preview",
74 messages=messages,
75 temperature=0.0,
76 response_format={
77 "type": "json_object",
78 }
79 )
80 handle_count += 1
81 print('OpenAI 交互处理次数:', handle_count)
82 response_msg = completion.choices[0].message.content.strip()
83 messages.append({
84 "role": "assistant",
85 "content": response_msg
86 })
87 try:
88 llm_response = json.loads(response_msg)
89 tool_name = llm_response['tool_name']
90 if tool_name == FunctionCallEnum.TaskCompleteTool.name:
91 print('任务完成', llm_response)
92 break
93 elif tool_name == FunctionCallEnum.AskHumanHelpTool.name:
94 print('需要人工帮助', llm_response)
95 input_content = input('请输入人工帮助的内容:')
96 func_call_message = {
97 "tool_name": FunctionCallEnum.AskHumanHelpTool.name,
98 "response": input_content
99 }
100 messages.append({
101 "role": "user",
102 "content": json.dumps(func_call_message)
103 })
104 elif tool_name == FunctionCallEnum.CodeInterpreterTool.name:
105 exec_python_code_in_sandbox(llm_response, messages)
106 else:
107 print('需要使用工具:', tool_name)
108 function_call(tool_name, messages)
109 except json.JSONDecodeError as ignored:
110 print('Json解析错误', ignored.msg, response_msg)
111 break
112
113
114
115if __name__ == '__main__':
116 load_tools()
117 # start_react('现在是几点?')
118 start_react('10分钟后是几点?')
119 # start_react('我想知道1.01的365次方减去100加上300是多少?')
以10分钟后是几点这个问题举例:
1
2{"thought":"首先需要获取当前的时间，然后在当前时间的基础上加上10分钟。","tool_name":"GetTimeTools","args_list":{}}
3{"thought":"获取到了当前时间，现在需要编写一个Python代码来计算10分钟后的时间。","tool_name":"CodeInterpreterTool","args_list":{"code":"
4from datetime import datetime, timedelta
5current_time = datetime.strptime('2024-02-18 17:31:43', '%Y-%m-%d %H:%M:%S')
6ten_minutes_later = current_time + timedelta(minutes=10)
7print(ten_minutes_later.strftime('%Y-%m-%d %H:%M:%S'))"}}
82024-02-18 17:41:43
9
10{"thought":"已经计算出10分钟后的时间，现在可以将结果告诉用户。","tool_name":"TaskCompleteTool","args_list":{"output":"10分钟后是2024年2月18日17点41分43秒。"}}
11
12Process finished with exit code 0
看起来很容易明白吧，其实就是利用LLM的推理能力，把复杂问题拆解为一步步可以使用工具解决的小问题，这样就实现了最基本的Agent的功能。
本文内容与代理服务仅供学习目的使用，并不代表任何法律建议。在阅读、分享或实践本文时，坚决遵守中华人民共和国现行法律法规，并尊重所有相关的知识产权和其他权利。提醒读者，在应用本文内容时，应确保其行为符合当地法律及规定，并承担相应的责任。本声明旨在促进法律知识的传播与教育，同时维护法治精神和社会秩序。任何违反相关法律规定的使用或行为，均与作者无关。