React Agent的经典循环逻辑：Thought → Action → Observation → Thought → … → Answer

至于React Agent的论文详解，可以参考我之前的博客：

https://blog.csdn.net/weixin_55688516/article/details/152412653?spm=1001.2014.3001.5501

任务：让Agent打开一个文件text.txt，阅读其中的内容。

1、角色分配

设置了三个角色（速览）：

• system：对模型的“底层规则/约束”，像“规则手册”，只给模型看。

• assistant：模型自己的输出（思考/动作/答案），Thought与Action。

• user：外部世界输入（人类提问、工具反馈、环境观察），Observation。

1️⃣System

作用：定义“长期不变的指令与格式要求”。例如让模型严格用 Thought/Action/Answer 格式。

特点：通常在每轮调用时都放在 messages 的最前面，保证模型的一致性，因为我们调用模型，是用的无记忆的API。不会出现在用户界面的对话中，它只是影响模型的风格边界，也就是规则手册。

例如，我这里设置的system prompt，限制模型的输出模式为React模式

SYSTEM_PROMPT = """

You are an agent that can use tools.

When you need to read a file, follow the format below exactly:

Thought: (your reasoning)

Action: read_file("relative_path")

If you already have the final answer, output it directly as:

Answer: (your final answer)

You must use only the keywords: Thought / Action / Answer.

Do not output any other formats.

"""

在每一个STEP，都会把system的prompt加在最前面

2️⃣assistant 

模型自己的输出，包括 Thought、Action、Answer，都是以这个角色放回历史。ReAct 需要让模型“看见自己上一步的思考和动作”，方便它延续推理链

• 中间轮：Thought/Action

• 终止轮：Answer

3️⃣ User

代表外部世界的输入，包括：最初的自然语言问题（人类问的）；执行工具后的结果（环境反馈）。

即Observation。e.g.

messages.append({"role": "user", "content": f"问题: {query}"})  # 人类问题
history.append({"role": "user", "content": f"Observation: {observation}"})  # 工具反馈

4️⃣ 一轮完整对话的时间线

第 1 轮调用前的 messages：

system:  你是工具型 Agent，必须用 Thought/Action/Answer 格式
user:    问题: 请读取 test.txt 的内容

模型(assistant) 输出：

assistant: Thought: 我需要先读取文件
                  Action: read_file("test.txt")

我们在本地执行 read_file → 得到文件内容，并把它当作 Observation 回给模型：

user: Observation: Hello from test.txt!

第 2 轮调用时的 messages：

system:   规则手册（同上）
assistant:Thought: 我需要先读取文件
          Action: read_file("test.txt")
user:  Observation: Hello from test.txt!

模型(assistant) 输出：

assistant: Answer: 文件内容是一句问候。

到这里模型没有再给 Action，我们就视为结束。

2、角色分配进阶版（Tool角色版）

添加Tool角色，运用Function Calling / Tools API

模型的输出不再是一段文字，而是一个 带有结构化调用信息的消息。

项目	传统字符串版	Function Calling / Tools API
模型输出	"Action: read_file('test.txt')"（纯文本）	{"tool_calls": [{"function": {"name": "read_file", "arguments": {"path": "test.txt"}}}]}（结构化 JSON）
你做的事	手动用正则或 split() 解析 Action	SDK 自动解析 tool 调用
Observation 角色	user: "Observation: ..."	tool: "内容..."（有 tool_call_id 对应）
优点	简单直观、能看懂	安全可靠、机器可读、框架可自动串联
缺点	容易解析错，格式不稳定	代码略复杂，但稳定性高

Observation 不再用 user 角色，而是用 tool 角色携带 tool_call_id 回传给模型。

无需字符串解析 Action:，模型会以结构化 tool_calls 告诉你要调用哪个函数、参数是什么。

数据流示意：

第一步：你发消息给模型

messages = [
    {"role": "system", "content": "你可以调用工具"},
    {"role": "user", "content": "请读取 test.txt 的内容"}
]

第二步：模型响应（assistant 消息 + tool_call）

{
  "role": "assistant",
  "content": null,
  "tool_calls": [
    {
      "id": "call_001",
      "type": "function",
      "function": {
        "name": "read_file",
        "arguments": "{\"path\": \"test.txt\"}"
      }
    }
  ]
}

第三步：你的程序执行工具（Python 函数）

result = read_file("test.txt")  # => "Hello from test.txt!"

第四步：把执行结果回传给模型（tool 角色）

messages.append({
    "role": "assistant",
    "tool_calls": resp.choices[0].message.tool_calls
})
messages.append({
    "role": "tool",
    "tool_call_id": "call_001",   # 对应上面 tool_calls 的 id
    "content": result             # 工具执行结果内容
})

第五步：模型看到工具返回结果，继续生成最终答案

resp2 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages
)

{
  "role": "assistant",
  "content": "文件 test.txt 的内容是：Hello from test.txt!"
}

与传统ReAct版本的比对：

传统版本：

system: 你是Agent
user: 请读取 test.txt 的内容
assistant: Thought: 我需要读文件
            Action: read_file("test.txt")
user: Observation: Hello from test.txt!
assistant: Answer: 文件内容是一句问候。

---

Function Calling版本：

角色	内容
system	你是一个会使用工具的Agent
user	请读取 test.txt 的内容
assistant	tool_calls = [{“function”:{“name”:“read_file”,“arguments”:{“path”:“test.txt”}}}]
tool	tool_call_id=call_001, content=“Hello from test.txt!”
assistant	content=“文件内容是一句问候。”

可以看到：

• Action → tool_calls（assistant 消息中）

• Observation → tool（工具返回消息）

• Answer → assistant（模型最终消息）

3、工具函数的编写与调用

大模型会根据推理给出决策，但其无法直接打开对应的文件，执行操作，具体的打开操作的执行，还是需要我们在本地进行编写，编写的这些便是工具函数。

编写完具体的工具函数后，将它们的函数名共同集成在一个变量中，等模型返回指令后，直接在其中调用，类似于这样：

这里的read_file函数，是一个打开文件进行读取的函数。

后续的调用是这样的：

4、工具函数的编写与调用（Function Calling版）

有Tool角色和Function Calling后，工具函数的书写工作并不会被简化，简化的部分是上图中的第二步，也就是返回信息正则化这个部分，由于Tool角色的存在，模型的返回内容会直接包含工具调用模块，我们逐步来看。

1️⃣工具函数书写。这一部分与常规ReAct的是一致的，根据需要实现的功能书写对应的函数。

2️⃣工具集

工具集需要写的规范详实，包括函数名、参数等，主要目的是告诉模型这个函数是做什么，怎么使用。类似于⬇️，我这里只展示了一个函数，可以放入更多的函数，让模型可做的事更丰富。

分发函数相当于模型和工具集之间的接口，模型会返回需要调用的工具名，而这个函数则根据这个工具名去工具集寻找，之后进行调用，完成功能。

5、代码附录（详细注释版）

1️⃣无function calling 版本

# Thought → Action → Observation → Thought → … → Answer
# 传统ReAct
import os # 读取环境变量（API Key）与做系统相关操作
from pathlib import Path # 更安全/直观的路径处理
from openai import OpenAI # OpenAI 官方 Python 客户端

# ========= 配置 =========
MODEL = "gpt-4o-mini" # 选用的模型；可按你账号可用的模型替换
MAX_STEPS = 5 # ReAct 循环最大步数，防止死循环
CWD = Path.cwd().resolve() # 限定“只允许读取当前工作目录及其子目录”
MAX_BYTES = 1024 * 1024 # 单文件最大读取 1MB，避免大文件阻塞/耗费过多token

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # 从环境变量读取 API Key 初始化客户端

# ========= 工具实现：只读当前目录的相对路径 =========
def safe_read_file(path_str: str) -> str: # 定义安全的文件读取工具（供 Agent 调用）
try:
# 去掉包裹引号与首尾空白（因为模型可能输出 read_file("test.txt") 有引号）
p = path_str.strip().strip('"').strip("'")
# 将相对路径与 CWD 合并得到绝对路径，并做规范化/解析
abs_path = (CWD / p).resolve()
# 路径逃逸保护：必须以 CWD 起始，防止读到工作目录外的敏感文件
if not str(abs_path).startswith(str(CWD)):
return "[ERROR] 仅允许读取当前目录内的文件。"

# 存在性与类型校验：必须存在且是普通文件
if not abs_path.exists() or not abs_path.is_file():
return f"[ERROR] 文件不存在: {p}"

# 大小限制：超过阈值则拒绝读取
size = abs_path.stat().st_size
if size > MAX_BYTES:
return f"[ERROR] 文件过大（{size} bytes），上限 {MAX_BYTES} bytes。"

# 优先按 UTF-8 文本读取；如果失败，再按二进制读取前 200 字节作为签名展示
try:
content = abs_path.read_text(encoding="utf-8")
# 内容过长则截断，避免把超长文本都回传给模型导致费用/上下文溢出
if len(content) > 2000:
content = content[:2000] + "\n[TRUNCATED]"
return content
except UnicodeDecodeError:
# 非文本（或编码不为 UTF-8）时，读取前 200 字节以便调试观察
with open(abs_path, "rb") as f:
data = f.read(200)
return f"[BINARY FILE HEAD 200B] {data!r}"
except Exception as e:
# 任何未预料到的异常，都以 [ERROR] 返回，避免抛出到主流程
return f"[ERROR] {e}"

# 将工具注册到一个名字到函数的映射中，供“Action: read_file(...)”解析后调用
TOOLS = {
"read_file": safe_read_file
}

# 这是给 LLM 的“系统提示词”，约束它输出固定格式，便于我们解析
SYSTEM_PROMPT = """You are an agent that can use tools.
When you need to read a file, follow the format below exactly:

Thought: (your reasoning)
Action: read_file("relative_path")

If you already have the final answer, output it directly as:

Answer: (your final answer)

You must use only the keywords: Thought / Action / Answer.
Do not output any other formats.
"""

def react_loop(query: str): # 核心：ReAct 循环（Reason → Act → Observe …）
history = [] # 保存多轮对话历史（含模型输出与Observation）
for step in range(MAX_STEPS): # 最多执行 MAX_STEPS 轮
# 1) 组装消息：system（约束格式）+ 历史（含 Observation）+ 当前用户问题
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages += history # 把前面轮次的 Thought/Action/Observation 放回上下文
messages.append({"role": "user", "content": f"问题: {query}"})

# 2) 调用 LLM 生成下一步（可能是 Thought+Action，也可能直接给 Answer）
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0.2, # 降低随机性，让格式更稳定
)
msg = resp.choices[0].message.content.strip() # 拿到模型文本输出
print(f"\n=== LLM 输出（第 {step+1} 轮）===\n{msg}\n") # 打印观测，便于调试

# 3) 判断本轮是否包含 Action（如果有，我们就执工具；否则认为它已给出最终 Answer）
if "Action:" in msg:
# 从输出里抽取“Action:”之后的一行（或同一行）作为动作调用串
act_line = msg.split("Action:", 1)[1].strip().splitlines()[0].strip()
# 解析函数名与参数：支持 read_file("test.txt") 这种带括号的形式
if "(" in act_line and act_line.endswith(")"):
act_name = act_line.split("(", 1)[0].strip() # 函数名：read_file
arg = act_line.split("(", 1)[1][:-1] # 参数："test.txt"（去掉末尾右括号）
else:
# 防御式：如果模型没按格式来，给空参数，后面会出错提示
act_name, arg = act_line, ""

# 查找对应工具函数；没找到则返回“未知工具”错误
tool_fn = TOOLS.get(act_name)
if tool_fn is None:
observation = f"[ERROR] 未知工具: {act_name}"
else:
# 调用工具函数，把参数字符串传入，并获取执行结果（Observation）
observation = tool_fn(arg)

# 把本轮的模型输出（含 Thought/Action）与我们的 Observation 一起写入历史
# 注意：这里把 Observation 作为 user 角色回给模型，模拟“环境反馈”
history.append({"role": "assistant", "content": msg})
history.append({"role": "user", "content": f"Observation: {observation}"})
print(f"--- 工具执行结果（Observation）---\n{observation}\n")
else:
# 没有 Action：认为模型已经产出最终答案（Answer），结束循环
print("✅ 最终回答：")
print(msg)
break

if __name__ == "__main__": # 脚本入口
# 你可以把 query 换成更具体的自然语言任务
# 例如：query = "请读取 test.txt 的内容，并告诉我它的前 20 个字符。"
query = "请读取 test.txt 的内容，并简要复述。"
react_loop(query) # 启动 ReAct 循环

2️⃣ 有function calling版

# ============================================================
# 目标：让模型以“结构化的 tool_calls”发起动作，你的程序执行工具，
# 再用 role="tool" 把结果回传给模型，直到得到最终回答。
# 使用：
# export OPENAI_API_KEY=你的key
# echo "Hello from test.txt!" > test.txt
# python react_read_file_fc.py
# ============================================================

import os
import json
from pathlib import Path
from typing import Any, Dict
from openai import OpenAI

# ========= 配置 =========
MODEL = "gpt-4o-mini" # 按你账号可用的模型替换
MAX_STEPS = 8 # 主循环最多迭代轮数（包含工具调用与最终回答）
CWD = Path.cwd().resolve() # 只允许读取“当前工作目录”及其子目录（路径白名单）
MAX_BYTES = 1024 * 1024 # 单文件最大读取 1MB，防止读入异常大文件
TRUNCATE_CHARS = 2000 # 文本截断，避免把超长内容全部塞入上下文导致费用/溢出

# 初始化 OpenAI 客户端：Chat Completions 是“无状态”的，
# 每轮都要把完整 messages 传入（包括 system、历史 assistant、tool 等）
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ========= 安全的文件读取工具 =========
def safe_read_file(path: str) -> str:
"""
受控读取文件：
- 仅允许相对路径，并限制在 CWD 子树内（防止路径逃逸）
- 大小上限 1MB
- 优先按 UTF-8 文本读取，失败则展示二进制前 200B
- 对超长文本进行截断（TRUNCATE_CHARS）
"""
try:
# 模型可能会带引号，这里做一次清洗
p = path.strip().strip('"').strip("'")
abs_path = (CWD / p).resolve()

# 路径逃逸保护：必须以 CWD 开头
if not str(abs_path).startswith(str(CWD)):
return "[ERROR] Access outside working directory is not allowed."

# 必须存在且为普通文件
if not abs_path.exists() or not abs_path.is_file():
return f"[ERROR] File not found: {p}"

# 大小上限
size = abs_path.stat().st_size
if size > MAX_BYTES:
return f"[ERROR] File too large ({size} bytes). Limit: {MAX_BYTES} bytes."

# 文本优先；若不是 UTF-8 文本，则返回二进制前 200B 作为签名
try:
content = abs_path.read_text(encoding="utf-8")
if len(content) > TRUNCATE_CHARS:
content = content[:TRUNCATE_CHARS] + "\n[TRUNCATED]"
return content
except UnicodeDecodeError:
with open(abs_path, "rb") as f:
head = f.read(200)
return f"[BINARY FILE HEAD 200B] {head!r}"
except Exception as e:
# 任何未考虑到的异常都用 ERROR 返回，避免直接抛出导致程序终止
return f"[ERROR] {e}"

# ========= 工具规范（Tools API 的 schema）=========
# 这是“告诉模型有哪些工具可用、如何调用”的契约（强约束 JSON Schema）
# 在 Function Calling 模式中，“Action 不再是文本”，而是 assistant 消息里的 tool_calls 数组
TOOLS_SPEC = [
{
"type": "function",
"function": {
"name": "read_file", # 工具名：模型会在 tool_calls 中引用此 name
"description": "Read the content of a local file (relative to the current working directory).",
"parameters": { # 参数结构（JSON Schema）
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Relative path to the file, e.g., 'test.txt' or 'subdir/file.txt'."
}
},
"required": ["path"] # 必填参数
}
}
}
]

# ========= 将 tool 调用分发到本地函数 =========
def dispatch_tool_call(name: str, arguments: Dict[str, Any]) -> str:
"""
根据工具名分派到对应的本地实现：
- 在这里做权限控制、参数校验、异常处理
- 返回的字符串会作为“tool 角色消息”的 content 回给模型（Observation）
"""
if name == "read_file":
return safe_read_file(arguments.get("path", ""))
return f"[ERROR] Unknown tool: {name}"

# ========= 主循环 =========
def run_agent(user_query: str):
"""
Function Calling 版 ReAct 主循环：
1) 准备 messages：system（规则手册）+ user（自然语言问题）
2) 调用模型 → 如果 assistant 消息包含 tool_calls（= Action 的结构化形式）
- 逐个执行工具，并以 role="tool" + tool_call_id 回传结果（= Observation）
- 再进入下一轮，让模型基于工具结果继续思考
3) 如果没有 tool_calls，视为模型给出最终回答（assistant.content），结束
"""

# 强烈建议每轮都把 system 放在最前（ChatCompletions 无记忆，需要反复注入规则）
messages: list[dict] = [
{
"role": "system",
"content": (
"You are an agent that can use tools. "
"If you need to read a file, call the 'read_file' tool with a JSON argument. "
"When you have enough information, answer the user's question clearly."
),
},
{"role": "user", "content": user_query}, # 初始自然语言问题
]

# 多轮迭代：模型可能先 tool_calls，看到 tool 响应后再继续生成，直到最终 Answer
for step in range(1, MAX_STEPS + 1):
print(f"\n=== Step {step}: model thinking ===")

# 关键调用：把“上下文 + 工具规范”一起交给模型
# - tools=TOOLS_SPEC 让模型知道可用函数及参数结构
# - tool_choice="auto" 允许模型自行决定是否调用工具
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS_SPEC,
tool_choice="auto",
temperature=0.2, # 低温更稳定，减少胡乱输出
)
msg = resp.choices[0].message # assistant 消息对象（可能有 content / tool_calls）

# =========== 分支 1：模型请求调用工具（Action 的结构化形式） ===========
if msg.tool_calls:
# 把 assistant 消息（含 tool_calls）加入对话历史
# 这样我们随后添加的 tool 响应可以用 tool_call_id 精准对应这一次调用
messages.append(
{
"role": "assistant",
"content": msg.content or "", # 有时模型会同时输出少量解释；没有就给空串
"tool_calls": msg.tool_calls, # 关键：结构化的“要调用哪些工具、用什么参数”
}
)

# 逐个执行工具，并把结果以 role="tool" 回传，同时带上 tool_call_id
for tc in msg.tool_calls:
name = tc.function.name
try:
# tool 调用参数是 JSON 字符串，要解析成字典
args = json.loads(tc.function.arguments or "{}")
except Exception:
args = {} # 防御式：解析失败时给空参数，具体错误留在工具内部返回

print(f"-> calling tool: {name}({args})")
# 这里是真正“落地执行动作”的地方（Action 执行）
result = dispatch_tool_call(name, args)
# 只打印前 200 个字符，避免刷屏
print(f"<- tool result: {result[:200]}{'...' if len(result) > 200 else ''}")

# 把工具执行结果作为“Observation”回给模型
# 注意：在 Function Calling 中，Observation 不再是 user，而是 role="tool"
# 且必须携带 tool_call_id 以与 assistant 的调用一一对应
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"content": result,
}
)

# 本轮到这里就结束了；继续下一轮，让模型“看到”工具结果并继续推理/回答
continue

# =========== 分支 2：没有 tool_calls → 模型直接给出最终答案 ===========
final = msg.content or "" # 若 content 为 None，用空串兜底
print("\n✅ Final Answer:\n" + final)
return # 结束主循环

# 超出最大步数仍未得到最终回答（例如模型不断请求工具但没有收敛）
print("\n[INFO] Reached MAX_STEPS without a final answer.")

# 脚本入口
if __name__ == "__main__":
# 你可以自由修改问题；如果文件在子目录，记得保证相对路径在 CWD 内
query = "Please read the content of test.txt and give me a brief summary."
run_agent(query)

6、测试结果

这里我只测试了Function Calling版本的代码，由于目前只是在测试阶段，不想花费太多。

我这里先使用deepseek 的API：https://platform.deepseek.com/

我把模型细节贴在这里，方便快速查阅。

在原代码中，将model和client信息改成下面这样：

MODEL = "deepseek-chat"

client = OpenAI(

api_key="YOUR_KRY",

base_url="https://api.deepseek.com/v1", # 注意 base_url)

进行测试，结果如下：