1.引入
聊天机器人：通用领域的知识问答，则依赖的是在线大模型/开源大模型的原生能力
私有数据：RAG
人工智能助手：调用工具完成任务
存在问题：如果同时有几个指令没法做
解决：AI AGENT
从AI代理被赋予特定的工作（Job(s)）开始，进而必须执行的操作（Action(s)）以完成这些工作，再到执行这些操作所需的特定能力（Capabilities）及其所需的熟练程度（Required Level of Proficiency）。为了达到这些能力的熟练程度，代理需要依赖于各种技术和技巧（Technologies and Techniques），而这些技术和技巧又必须通过精确的编排（Orchestration）来实现有效整合。整个过程形成了一个系统，其中每个部分都是实现AI代理高效运作的关键。
举例：
如果想出去订票，prompt如下：
prompt = """
你需要在“思考、行动、观察、回答”的循环中运行。
在循环的最后，你需要输出一个答案。
使用“思考”来描述你对被问及问题的思考。
使用“行动”来执行可用的行动之一。
“观察”将是执行这些行动后的结果。
“回答”将是对观察结果的分析。
你的可用行动有：
小红书:
例如：小红书: 北京旅游攻略
通过小红书API搜索，并返回北京旅游攻略和推荐。
携程:
例如：携程: 前往北京的航班
通过携程API搜索，并找到前往北京的可用航班。
尽可能使用小红书和携程API进行查询。
示例会话：
问题: 我正计划去北京旅游，我应该先做什么？
思考: 我应该在小红书上查找关于访问北京的景点和攻略。
行动: 小红书: 北京旅游攻略
观察: 搜索返回了北京的热门旅游攻略和必游景点的列表。
回答: 首先，你可以在小红书上了解北京的必游景点和旅游攻略。接着，在携程上查找可用的前往北京的航班，并考虑住宿选择。
.......
"""
这来源于react框架
2.react框架
ReAct 组成： Reason 和 Act
Reason：基于一种推理技术——思想链（CoT），将输入分解为多个逻辑思维步骤，
分解问题：当面对复杂的任务时，CoT 方法不是通过单个步骤解决它，而是将任务分解为更小的步骤，每个步骤解决不同方面的问题。
顺序思维：思维链中的每一步都建立在上一步的结果之上。这样，模型就能从头到尾构造出一条逻辑推理链。
存在问题：幻觉，且问题会被不断放大
解决：ReAct的技术，采用“思考、行动、观察、回答”循环
prompt：先告诉它动作以及每个动作范围，然后写个例子
prompt = """
您在一个由“思考、行动、观察、回答”组成的循环中运行。
在循环的最后，您输出一个答案。
使用“思考”来描述您对所提问题的思考。
使用“行动”来执行您可用的动作之一。
“观察”将是执行这些动作的结果。
“回答”将是分析“观察”结果后得出的答案。
您可用的动作包括：
calculate（计算）:
例如：calculate: 4 * 7 / 3
执行计算并返回数字 - 使用Python，如有必要请确保使用浮点数语法
wikipedia（维基百科）:
例如：wikipedia: Django
返回从维基百科搜索的摘要
如果有机会，请始终在维基百科上查找信息。
示例会话：
问题：法国的首都是什么？
思考：我应该在维基百科上查找关于法国的信息
行动：wikipedia: France
然后您应该调用适当的动作，并从结果中确定答案
您然后输出：
回答：法国的首都是巴黎
"""
面对不同的场景，其实我们只需要改变的是：1. 代理的身份设定 2. 代理完成任务所需要的工具。
3.实操：
流程：
精心设计代理的完整提示词，并在大模型的system角色设置中进行设定，以确保代理的行为和知识与其角色一致。
实时将用户的问题作为变量输入，填充到系统提示（System Prompt）中，确保代理能够根据当前的用户需求生成响应。
构建并整合所需的工具，使ReAct Agent能够完成预定任务，这些工具也应作为变量被嵌入到系统提示中，以便在运行时调用。
eg
Step 1. 设计完整的代理工程提示
system_prompt = """
You run in a loop of Thought, Action, Observation, Answer.
At the end of the loop you output an Answer
Use Thought to describe your thoughts about the question you have been asked.
Use Action to run one of the actions available to you.
Observation will be the result of running those actions.
Answer will be the result of analysing the Observation
Your available actions are:
calculate:
e.g. calculate: 4 * 7 / 3
Runs a calculation and returns the number - uses Python so be sure to use floating point syntax if necessary
fetch_real_time_info:
e.g. fetch_real_time_info: Django
Returns a real info from searching SerperAPI
Always look things up on fetch_real_time_info if you have the opportunity to do so.
Example session:
Question: What is the capital of China?
Thought: I should look up on SerperAPI
Action: fetch_real_time_info: What is the capital of China?
PAUSE
You will be called again with this:
Observation: China is a country. The capital is Beijing.
Thought: I think I have found the answer
Action: Beijing.
You should then call the appropriate action and determine the answer from the result
You then output:
Answer: The capital of China is Beijing
Example session
Question: What is the mass of Earth times 2?
Thought: I need to find the mass of Earth on fetch_real_time_info
Action: fetch_real_time_info : mass of earth
PAUSE
You will be called again with this:
Observation: mass of earth is 1,1944×10e25
Thought: I need to multiply this by 2
Action: calculate: 5.972e24 * 2
PAUSE
You will be called again with this:
Observation: 1,1944×10e25
If you have the answer, output it as the Answer.
Answer: The mass of Earth times 2 is 1,1944×10e25.
Now it's your turn:
""".strip()
提示词的第一部分告诉大模型如何通过我们之前看到的流程的标记部分循环处理问题，第二部分描述计算和搜索维基百科的工具操作，最后是一个示例的会话。
Step 2. 定义工具
一共需要两个工具，其一是用来根据关键词检索Serper API，返回详细的检索信息。其二是一个计算函数，接收的入参是需要执行计算操作的数值，返回最终的计算结果。
import requests
import json
def fetch_real_time_info(query):
# API参数
params = {
'api_key': '0f31d8c5561bdaa4c71ad7c86f6e63a4a26cead9', # 使用您自己的API密钥
'q': query, # 查询参数，表示要搜索的问题。
'num': 1 # 返回结果的数量设为1，API将返回一个相关的搜索结果。
}
# 发起GET请求到Serper API
api_result = requests.get('https://google.serper.dev/search', params)
# 解析返回的JSON数据
search_data = api_result.json()
# 提取并返回查询到的信息
if search_data["organic"]:
return search_data["organic"][0]["snippet"]
else:
return "没有找到相关结果。"
def calculate(operation: str) -> float:
return eval(operation)
最后，定义一个名为 available_actions 的字典，用来存储可用的函数引用，用来在后续的Agent 实际执行 Action 时可以根据需要调用对应的功能。
available_actions = {
"fetch_real_time_info": fetch_real_time_info,
"calculate": calculate,
}
Step 3. 开发大模型交互接口
接下来，定义大模型交互逻辑接口。这里我们实现一个聊天机器人的 Python 类，将系统提示（system）与用户（user）或助手的提示（assistant）分开，并在实例化ChatBot时对其进行初始化。 核心逻辑为 __call__函数负责存储用户消息和聊天机器人的响应，调用execute来运行代理。完整代码如下所示：
import openai
import re
import httpx
from openai import OpenAI
class ChatBot:
def __init__(self, system=""):
self.system = system
self.messages = []
if self.system:
self.messages.append({"role": "system", "content": system})
def __call__(self, message):
self.messages.append({"role": "user", "content": message})
result = self.execute()
self.messages.append({"role": "assistant", "content": result})
return result
def execute(self):
client = OpenAI()
completion = client.chat.completions.create(model="gpt-4o", messages=self.messages)
return completion.choices[0].message.content
如上所示，这段代码定义了一个ChatBot的类，用来创建和处理一个基于OpenAI GPT-4模型的聊天机器人。下面是每个部分的具体解释：
init 方法用来接收系统提示(System Prompt)，并追加到全局的消息列表中。
call 方法是 Python 类的一个特殊方法, 当对一个类的实例像调用函数一样传递参数并执行时，实际上就是在调用这个类的 call 方法。其内部会 调用execute 方法。
execute 方法实际上就是与OpenAI的API进行交互，发送累积的消息历史（包括系统消息、用户消息和之前的回应）到OpenAI的聊天模型,返回最终的响应。
Step 4. 定义代理循环逻辑
从Thought 到 Action ， 最后到 Observation 状态，是一个循环的逻辑，而循环的次数，取决于大模型将用户的原始 Goal 分成了多少个子任务。 所有在这样的逻辑中，我们需要去处理的是：
判断大模型当前处于哪一个状态阶段
如果停留在 Action 阶段，需要像调用 Function Calling 的过程一样，先执行工具，再将工具的执行结果传递给Obversation 状态阶段。
action这一步
首先需要明确，需要执行操作的过程是：大模型识别到用户的意图中需要调用工具，那么其停留的阶段一定是在 Action：xxxx : xxxx 阶段，其中第一个 xxx，就是调用的函数名称，第二个 xxxx，就是调用第一个 xxxx 函数时，需要传递的参数。这里就可以通过正则表达式来进行捕捉。如下所示：
# (\w+) 是一个捕获组，匹配一个或多个字母数字字符（包括下划线）。这部分用于捕获命令中指定的动作名称
# (.*) 是另一个捕获组，它匹配冒号之后的任意字符，直到字符串结束。这部分用于捕获命令的参数。
action_re = re.compile('^Action: (\w+): (.*)$')
由此，我们定义了如下的一个 AgentExecutor函数。该函数实现一个循环，检测状态并使用正则表达式提取当前停留的状态阶段。不断地迭代，直到没有更多的（或者我们已达到最大迭代次数）调用操作，再返回最终的响应。完整代码如下：
action_re = re.compile('^Action: (\w+): (.*)$')
def AgentExecutor(question, max_turns=5):
i = 0
bot = ChatBot(system_prompt)
# 通过 next_prompt 标识每一个子任务的阶段性输入
next_prompt = question
while i < max_turns:
i += 1
# 这里调用的就是 ChatBot 类的 __call__ 方法
result = bot(next_prompt)
print(f"result:{result}")
# 在这里通过正则判断是否到了需要调用函数的Action阶段
actions = [action_re.match(a) for a in result.split('\n') if action_re.match(a)]
if actions:
# 提取调用的工具名和工具所需的入参
action, action_input = actions[0].groups()
if action not in available_actions:
raise Exception("Unknown action: {}: {}".format(action, action_input))
print(f"running: {action} {action_input}")
observation = available_actions[action](action_input)
print(f"Observation: {observation}")
next_prompt = "Observation: {}".format(observation)
else:
return bot.messages
从上面我们实现的案例中，非常明显的发现，ReAct（推理和行动）框架通过将推理和行动整合到一个有凝聚力的操作范式中，能够实现动态和自适应问题解决，从而允许与用户和外部工具进行更复杂的交互。这种方法不仅增强了大模型处理复杂查询的能力，还提高了其在多步骤任务中的性能，使其适用于从自动化客户服务到复杂决策系统的广泛应用。
就目前的AI Agent 现状而言，流行的代理框架都有内置的 ReAct 代理，比如Langchain、LlamaIndex中的代理，或者 CrewAI这种新兴起的AI Agent开发框架，都是基于ReAct理念的一种变种。LangChain 的 ReAct 代理工程描述 👇
Answer the following questions as best you can. You have access
to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}
There are three placeholders {tool}, {input}, and {agent_scratchpad} in this prompt. These will be replaced with the appropriate text before sending it to LLM.