ReAct框架代码实现原理：核心机制与实战指南
ReAct Agent 的运行机制详解
“Reason Only”（内部推理优先）：该模式依赖类似思维链（CoT）的提示词，引导模型逐步拆解复杂问题，逐一攻克。逻辑分析能力出色，但致命短板在于无法调用外部工具或获取实时知识。形象地说，如同一位思维极其缜密、但对外界变化缺乏感知的学者。
“Act Only”（外部工具驱动）：典型的工具调用提示模式，擅长通过搜索引擎、数据库等外部资源解决问题，但自身的推理链条偏弱。好比一个好奇心旺盛、热衷于尝试各种新工具的实践者，但缺乏深度思考的习惯。
“ReAct”（推理+行动融合）：将前两者有机结合——既保持严谨的推理链路，又具备调用外部工具的能力，实现优势互补。简言之，让模型既能「想」又能「做」，内外兼修。
直观理解：ReAct = 推理能力 + 工具使用能力（或外部知识），帮助大模型攻克单靠内部参数无法完成的现实难题。
在 ReAct 提示中，模型会交错生成「推理轨迹」和「任务特定操作」，使二者产生协同放大效应：
- 推理轨迹：协助模型制定行动计划、追踪执行进度、处理异常情况。
- 执行操作：与外部源（如知识库、搜索引擎或环境）交互，获取必要信息。
接下来，我们逐层拆解 ReAct 框架的代码实现原理，共分三个核心步骤。
第一步：
定义 Prompt 模版
TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters} Format the arguments as a JSON object."""
REACT_PROMPT = """Answer the following questions as best you can. You ha ve access to the following tools:
{tool_descs}
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 be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {query}"""
工具构建函数
def build_planning_prompt(TOOLS, query):
tool_descs = []
tool_names = []
for info in TOOLS:
tool_descs.append(
TOOL_DESC.format(
name_for_model=info['name_for_model'],
name_for_human=info['name_for_human'],
description_for_model=info['description_for_model'],
parameters=json.dumps(
info['parameters'], ensure_ascii=False),
)
)
tool_names.append(info['name_for_model'])
tool_descs = 'nn'.join(tool_descs)
tool_names = ','.join(tool_names)
prompt = REACT_PROMPT.format(tool_descs=tool_descs, tool_names=tool_names, query=query)
return prompt
prompt_1 = build_planning_prompt(TOOLS[0:1], query="加拿大2023年人口统计数字是多少？")
print(prompt_1)
调用函数查看 response 结果
Answer the following questions as best you can. You ha ve access to the following tools:
Search: Call this tool to interact with the google search API. What is the google search API useful for? useful for when you need to answer questions about current events. Parameters: [{"name": "query", "type": "string", "description": "search query of google", "required": true}] Format the arguments as a JSON object.
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 [Search]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: 加拿大2023年人口统计数字是多少？
初始化模型
# 国内连 hugginface 网络不好，这段代码可能需要多重试
checkpoint = "Qwen/Qwen-7B-Chat"
TOKENIZER = AutoTokenizer.from_pretrained(checkpoint, trust_remote_code=True)
MODEL = AutoModelForCausalLM.from_pretrained(checkpoint, device_map="auto", trust_remote_code=True).eval()
MODEL.generation_config = GenerationConfig.from_pretrained(checkpoint, trust_remote_code=True)
MODEL.generation_config.do_sample = False # greedy
设置 "Observation" 为 stop word
stop = ["Observation:", "Observation:n"]
react_stop_words_tokens = [TOKENIZER.encode(stop_) for stop_ in stop]
response_1, _ = MODEL.chat(TOKENIZER, prompt_1, history=None, stop_words_ids=react_stop_words_tokens)
print(response_1)
当模型预测到即将生成的词为 "Observation" 时，立即中断生成。此时该 prompt 会输出以下内容：
Thought: 我应该使用搜索工具帮助我完成任务。search api能完成搜索的任务。
Action: Search
Action Input: {"query": "加拿大 2023年人口统计数字"}
Observation:
第二步：
从输出中解析需要使用的工具和入参
def parse_latest_plugin_call(text: str) -> Tuple[str, str]:
i = text.rfind('nAction:')
j = text.rfind('nAction Input:')
k = text.rfind('nObservation:')
if 0 <= i < j: # If the text has `Action` and `Action input`,
if k < j: # but does not contain `Observation`,
# then it is likely that `Observation` is ommited by the LLM,
# because the output text may ha ve discarded the stop word.
text = text.rstrip() + 'nObservation:' # Add it back.
k = text.rfind('nObservation:')
if 0 <= i < j < k:
plugin_name = text[i + len('nAction:'):j].strip()
plugin_args = text[j + len('nAction Input:'):k].strip()
return plugin_name, plugin_args
return '', ''
调用对应工具
def use_api(tools, response):
use_toolname, action_input = parse_latest_plugin_call(response)
if use_toolname == "":
return "no tool founds"
used_tool_meta = list(filter(lambda x: x["name_for_model"] == use_toolname, tools))
if len(used_tool_meta) == 0:
return "no tool founds"
api_output = used_tool_meta[0]["tool_api"](action_input)
return api_output
api_output = use_api(TOOLS, response_1)
print(api_output)
工具返回的结果
根据加拿大统计局预测，加拿大人口今天（2023年6月16日）预计将超过4000万。 联邦统计局使用模型来实时估计加拿大的人口，该计数模型预计加拿大人口将在北美东部时间今天下午3点前达到4000万。 加拿大的人口增长率目前为2.7％。
第三步：
根据工具返回结果继续调用模型
将工具返回的答案拼接到原有 prompt 中，构成新的提示，然后请求模型生成最终结果。
prompt_2 = prompt_1 + response_1 + ' ' + api_output
stop = ["Observation:", "Observation:n"]
react_stop_words_tokens = [TOKENIZER.encode(stop_) for stop_ in stop]
response_2, _ = MODEL.chat(TOKENIZER, prompt_2, history=None, stop_words_ids=react_stop_words_tokens)
print(prompt_2, response_2)
续写，生成最终答案
按 ReAct 格式继续产出 Thought 和 Final Answer，得到完整回答。
Answer the following questions as best you can. You ha ve access to the following tools:
Search: Call this tool to interact with the google search API. What is the google search API useful for? useful for when you need to answer questions about current events. Parameters: [{"name": "query", "type": "string", "description": "search query of google", "required": true}] Format the arguments as a JSON object.
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 [Search]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: 加拿大2023年人口统计数字是多少？Thought: 我应该使用搜索工具帮助我完成任务。search api能完成搜索的任务。
Action: Search
Action Input: {"query": "加拿大 2023年人口统计数字"}
Observation: 根据加拿大统计局预测，加拿大人口今天（2023年6月16日）预计将超过4000万。 联邦统计局使用模型来实时估计加拿大的人口，该计数模型预计加拿大人口将在北美东部时间今天下午3点前达到4000万。 加拿大的人口增长率目前为2.7％。
Thought: I now know the final answer.
Final Answer: 加拿大2023年人口统计数字预计为4000万。
归纳
def main(query, choose_tools):
# 组织 prompt
prompt = build_planning_prompt(choose_tools, query)
# 设置 stop word，模型停止生成
stop = ["Observation:", "Observation:n"]
react_stop_words_tokens = [TOKENIZER.encode(stop_) for stop_ in stop]
response, _ = MODEL.chat(TOKENIZER, prompt, history=None, stop_words_ids=react_stop_words_tokens)
while "Final Answer:" not in response: # 出现final Answer时结束
api_output = use_api(choose_tools, response) # 抽取入参并执行api
api_output = str(api_output) # 部分api工具返回结果非字符串格式需进行转化后输出
if "no tool founds" == api_output:
break
print(" 33[32m" + response + " 33[0m" + " 33[34m" + ' ' + api_output + " 33[0m")
prompt = prompt + response + ' ' + api_output # 合并api输出
response, _ = MODEL.chat(TOKENIZER, prompt, history=None, stop_words_ids=react_stop_words_tokens) # 继续生成
print(" 33[32m" + response + " 33[0m")
- 核心思路是利用大模型的自回归续写能力，按 ReAct 步骤逐轮迭代。
- 通过设置停止词（stop word）控制输出节奏，当生成内容命中停止词时模型立即停止。
- 解析工具名称与参数，正确调用外部工具获取实时数据。
- 将工具返回结果按 ReAct 格式拼接，再次调用模型完成续写。
- 持续循环直至输出中包含
Final Answer:，即得到最终答案。 - 每使用一次工具，至少需要发起两次大模型调用（第一次生成 Action，第二次生成最终答案）。