ReAct执行原理
参考资料
https://www.bilibili.com/video/BV1dw526tEMA
https://github.com/Wood-Q/MokioAgent
ReAct(ReasoningActing)范式即让模型在Thought → Action → Observation中循环，进行多步骤推理从而解决复杂问题的思想
- Thought（思考）：分析问题、规划下一步。例如判断是否需要外部数据。
- Action（行动）：执行具体操作，如调用搜索引擎、计算器等。
- Observation（观察）：接收工具返回结果，为下一轮推理提供依据。
DEFAULT_TASK = "请检查 inbox，把 a.txt 移动到 archive，然后告诉我整理后的目录变化。"
WORKSPACE = Path(__file__).resolve().parent / "demo_workspace"
def reset_workspace() -> None:
if WORKSPACE.exists():
shutil.rmtree(WORKSPACE)
(WORKSPACE / "inbox").mkdir(parents=True)
(WORKSPACE / "archive").mkdir()
(WORKSPACE / "inbox" / "a.txt").write_text(
"Hello from MokioClaw AgentLoop demo.",
encoding="utf-8",
)
def show_workspace() -> str:
items = sorted(WORKSPACE.rglob("*"))
lines = []
for item in items:
rel = item.relative_to(WORKSPACE).as_posix()
lines.append(f"- {rel}/" if item.is_dir() else f"- {rel}")
return "\n".join(lines) or "(empty)"
def workspace_path(path: str) -> Path:
path = path.strip().replace("\\", "/")
return WORKSPACE if path == "." else WORKSPACE / path.strip("/")
@tool
def list_files(path: str = ".") -> str:
"""List files in the demo workspace."""
target = workspace_path(path)
if target.is_file():
return target.relative_to(WORKSPACE).as_posix()
files = sorted(item for item in target.rglob("*") if item.is_file())
return "\n".join(f"- {item.relative_to(WORKSPACE).as_posix()}" for item in files) or "(empty)"
@tool
def move_file(source: str, target: str) -> str:
"""Move a file in the demo workspace."""
source_path = workspace_path(source)
target_path = workspace_path(target)
if "." not in target_path.name:
target_path = target_path / source_path.name
target_path.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(source_path), str(target_path))
return f"moved {source} -> {target_path.relative_to(WORKSPACE).as_posix()}"
def load_llm() -> ChatOpenAI:
load_dotenv(Path(__file__).resolve().parents[1] / ".env")
return ChatOpenAI(
model=os.getenv("MODEL", "qwen3.6-flash"),
base_url=os.getenv("BASE_URL"),
api_key=os.getenv("API_KEY"),
temperature=0,
)
SYSTEM_PROMPT = """
你是一个文件整理助手。
你可以反复调用工具，直到完成用户任务。
推荐顺序：先 list_files("inbox")，再 move_file("inbox/a.txt", "archive/a.txt")，再 list_files(".")，最后总结。
每一轮最多调用一个工具。
""".strip()
def main() -> None:
task = " ".join(sys.argv[1:]).strip() or DEFAULT_TASK
reset_workspace()
print("=== 01. 手写 while AgentLoop ===")
print("\n用户任务:")
print(task)
print("\n运行前 workspace:")
print(show_workspace())
tools = [list_files, move_file]
tool_map = {item.name: item for item in tools}
llm = load_llm().bind_tools(tools)
messages = [SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=task)]
for turn in range(1, 8): # 强制8轮结束
print(f"\n--- 第 {turn} 轮：模型思考 ---")
response = llm.invoke(messages)
messages.append(response)
if not response.tool_calls: # 如果最终回答不包含tool_calls代表需要退出工具循环链路
print("\n最终回答:")
print(response.content)
break
tool_call = response.tool_calls[0]
print("\n模型决定调用工具:")
print(f"tool_name = {tool_call['name']}")
print(f"tool_args = {tool_call['args']}")
result = tool_map[tool_call["name"]].invoke(tool_call["args"])
print("\n工具返回:")
print(result)
messages.append(
ToolMessage(
content=str(result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
print("\n运行后 workspace:")
print(show_workspace())
if __name__ == "__main__":
main()
代码中提供了列出文件和移动文件的方法，在Loop中，当模型输出的tool_calls为空，则代表工具调用结束，正常输出。
每一轮对话，都将response塞回messages历史上下文中
使用LangChain简化流程
LangChain中提供了create_agent方法，自动管理模型调用的Loop循环
def main() -> None:
from langchain.agents import create_agent
task = " ".join(sys.argv[1:]).strip() or DEFAULT_TASK
reset_workspace()
print("=== 02. LangChain create_agent：封装好的 ReAct Loop ===")
print("\n用户任务:")
print(task)
print("\n运行前 workspace:")
print(show_workspace())
agent = create_agent(
model=load_llm(),
tools=[list_files, move_file],
system_prompt=SYSTEM_PROMPT,
)
result = agent.invoke({"messages": [{"role": "user", "content": task}]})
print("\nAgent 消息流:")
for message in result["messages"]:
print(f"\n[{getattr(message, 'type', 'unknown')}]")
if getattr(message, "tool_calls", None):
print(message.tool_calls)
elif getattr(message, "content", ""):
print(message.content)
print("\n运行后 workspace:")
print(show_workspace())
其中，result = agent.invoke({"messages": [{"role": "user", "content": task}]})是一个阻塞的操作，最终任务执行完毕后才会执行消息流的打印
观察后发现，即使我们在SystenPrompt中声明了每一轮只调用一次工具，但是在实际的API通信中发现，模型会调用多从ToolCall。因此只靠提示词来约束模型是不稳定的。
但是在我们自己写的Agent Loop中，只处理来自ToolCall的第一条数据，也是不合理的，这会导致API厂商在处理工具反馈的时候，缺少对应的tool_call_id导致出现400报错
所以其实最佳的解决方案就是不解决（雾），让模型一轮能够调用多个工具，因为这样节省token。
当然有些厂商支持关闭并行tool_call，使用parallel_tool_calls=False，可以保证每轮只调用一次工具
评论（0）
暂无评论