AI 代理:掌握3个基本模式(ReAct)。第2部分,共3部分
Source: Dev.to
文章 – 第 1 部分
这些模式的代码可在 GitHub 上获取。 → [Repo]
如果说使用 “Tool Using”(第 1 级)模式我们给了 AI “手”,让它能够触及外部世界,那么使用 ReAct 则是为它装上了一个 “功能性大脑”。
回顾第 1 级
一切都非常线性:
刺激 → 响应
你请求一个数据,代理触发一个工具,就结束了。没有疑问,也没有计划。
但现实世界是混乱的。
- 当一个问题不能通过一次点击解决时会怎样?
- 当需要调查、比较数据然后再进行计算时会怎样?
在这些情况下,线性模型会崩溃,ReAct(由普林斯顿和谷歌的研究者在 2022 年提出)就会介入。如今它基本上已经成为行业标准。
ReAct = Reasoning + Acting(推理 + 行动)
ReAct 有何不同?
它不仅仅是使用工具(这我们已经有了)。
我们强制 LLM 保持一个 内部独白——一个 “思考轨迹”。
代理不会盲目地直接回答,而是进入一个认知循环:
思考 → 行动 → 观察 → 重复
代理实际上在日志中 “自言自语”。它会根据刚刚发现的内容规划下一步,这使得它能够在发现情况不如预期时即时纠正方向。
步骤示例
用户提问:
“法国现任总统的年龄的平方根是多少?”
| 迭代 | 思想 (Thought) | 行动 (Action) | 观察 (Observation) |
|---|---|---|---|
| 1 | “好吧,我根本不知道总统的年龄。事实上,我甚至不知道现在的总统是谁。先查一下。” | search_tool("Presidente de Francia actual") | “Emmanuel Macron.” |
| 2 | “好的,我们已经有名字:Macron。现在我需要他的年龄。” | search_tool("Emmanuel Macron edad") | “47岁。” |
| 3 | “我得到了47。用户想要它的平方根。” | calculator_tool("sqrt(47)") | “6.8556…” |
最终回答:
“埃马纽埃尔·马克龙(47岁)的年龄的平方根约为 6.85。”
看到魔法了吗?没有先完成第2步,第3步是不可能的。代理人一直在“串联线索”。
使用 Agno 实现 ReAct 模式
import os
import sys
import logging
import traceback
from typing import List, Optional
from dotenv import load_dotenv, find_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.tavily import TavilyTools
# ----------------------------------------------------------------------
# 1. 全局日志配置和异常处理
# ----------------------------------------------------------------------
LOG_DIR = os.path.join(os.path.dirname(__file__), "log")
LOG_FILE = os.path.join(LOG_DIR, "logs.txt")
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(LOG_FILE, encoding="utf-8"),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
def global_exception_handler(exctype, value, tb):
"""捕获未处理的异常并记录。"""
error_msg = "".join(traceback.format_exception(exctype, value, tb))
logger.error(f"Unhandled exception:\n{error_msg}")
sys.__excepthook__(exctype, value, tb)
sys.excepthook = global_exception_handler
# ----------------------------------------------------------------------
# 2. 加载环境变量
# ----------------------------------------------------------------------
env_path = find_dotenv()
if env_path:
load_dotenv(env_path)
logger.info(f".env file loaded from: {env_path}")
else:
logger.warning(".env file not found")
# ----------------------------------------------------------------------
# 3. 工具定义
# ----------------------------------------------------------------------
def calculate(expression: str) -> str:
"""
求解一个简单的数学表达式(加、减、乘、除)。
用于计算年份或日期差异。
Args:
expression (str): 待求值的表达式,例如 "2024 - 1789"。
Returns:
str: 结果或错误信息。
"""
try:
# 只允许安全字符
allowed_chars = "0123456789+-*/(). "
if all(c in allowed_chars for c in expression):
result = eval(expression) # noqa: S307 (受控使用)
return f"Result: {result}"
else:
return "Error: Disallowed characters in expression."
except Exception as e:
return f"Error while calculating: {str(e)}"
# ----------------------------------------------------------------------
# 4. Agno 代理配置(ReAct 模式)
# ----------------------------------------------------------------------
model_id = os.getenv("BASE_MODEL", "gpt-4o")
agent = Agent(
model=OpenAIChat(id=model_id),
tools=[TavilyTools(), calculate],
instructions=[
"You are a researcher using the ReAct (Reason + Act) method.",
"1. Think step‑by‑step about what information you need to answer the user's question.",
"2. Use the search tool (Tavily) to find specific dates, facts, or data.",
"3. Use the calculator ('calculate') for any mathematical operation or time calculation.",
"4. Do not guess historical information. If you don't have a piece of data, look it up.",
"5. Show your reasoning clearly: 'Thought:', 'Action:', 'Observation:'.",
"6. Continue investigating until you have a complete and verified answer."
],
)
# ----------------------------------------------------------------------
# 5. 用户界面
# ----------------------------------------------------------------------
def main():
logger.info("Starting Historical Detective Agent (ReAct)...")
print("--- Historical Detective - ReAct Pattern ---")
print("Type 'exit' to quit.\n")
while True:
> **Source:** ...
```python
try:
user_input = input("Researcher, what is your question?: ")
if user_input.lower() == "exit":
logger.info("The user has ended the session.")
break
if not user_input.strip():
continue
logger.info(f"User query: {user_input}")
print("\nInvestigating...\n")
# Ejecutar el agente con el prompt del usuario
response = agent.run(user_input)
print("\n--- Answer ---")
print(response)
print("\n----------------\n")
except KeyboardInterrupt:
logger.info("Session interrupted by user (Ctrl+C).")
break
except Exception as e:
logger.error(f"Error during processing: {e}")
if __name__ == "__main__":
main()
这个脚本的功能
- 日志记录 → 将所有活动保存到
log/logs.txt并在控制台显示。 - 异常处理 → 捕获任何意外错误并记录。
- 工具 →
TavilyTools用于网页搜索。calculate用于简单的算术运算。
- ReAct 代理 → 按照强制模型的指令进行:
- 步骤化思考。
- 执行(搜索或计算)。
- 观察结果。
- 重复上述过程,直至得到完整且已验证的答案。
- 交互界面 → 在控制台循环,用户输入问题,代理回复,并在日志中展示内部过程。
注意:
- 确保有一个
.env文件,其中包含BASE_MODEL变量(例如gpt-4o)。- 在运行脚本前安装所需依赖(
agno、python-dotenv等)。
完成!现在你可以尝试 ReAct 模式,观察代理如何“自我对话”来解决复杂问题。
nt.print_response(user_input, stream=True, show_tool_calls=True)
print("\n")
except KeyboardInterrupt:
logger.info("Keyboard interrupt detected.")
break
except Exception as e:
logger.error(f"Error in main loop: {str(e)}")
print(f"\nAn error occurred: {e}")
if __name__ == "__main__":
main()
ReAct 在代码中的位置?
PENSAR (Think): 在代理的 instructions 中定义。
ACTUAR (Act): 代理在 tools 中拥有外部能力。
OBSERVAR (Observe): 框架(Agno)执行工具并将结果返回给代理。
REPETIR (Repeat): 循环持续,直到代理拥有足够的信息。
为什么我们如此喜欢这个模式? (优点)
-
解决“多跳”问题(Multi‑hop)
这是能够回答 A 引导到 B,B 再引导到 C 的问题的能力。 -
“自我修复”能力(Self‑Healing)
这很关键。想象一下你搜索“马克龙年龄”,而 Google 失败。普通脚本会崩溃。ReAct 代理会想:“哎,搜索失败了。好,我去找他的出生日期,然后自己计算年龄”。 -
告别黑箱
作为开发者,你可以读取 Thoughts。你确切知道代理为何决定使用计算器而不是别的工具。 -
更少的谎言(幻觉)
强制它每一步都基于真实的观察(Observation),就更难出现捏造数据。 -
速度慢(延迟)
ReAct 是顺序执行的。思考 3 次意味着调用 LLM 3 次并执行工具。准备好等待 10 到 30 秒。 -
API 费用(成本)
那段“内部独白”会消耗大量 token,仿佛没有明天。你的上下文历史会很快被填满。 -
沉迷的风险
有时会进入循环:“搜索 X → 未找到 → 认为需要再次搜索 X → 再搜索 X…”。 -
提示设计微妙
需要一个调校得非常好的系统 Prompt,让模型知道何时停止思考并给出答案。
生产中的最佳实践(Agno、LangChain 等)
-
设置一个刹车(最大迭代次数)
始终配置迭代上限(例如 10 步)。如果在 10 步内没有解决问题,100 步也不会解决,而且只会烧钱。 -
教它保持沉默(停止序列)
从技术上讲,LLM 在触发 Action 后应立即停止写作。如果不在此处截断,它会自行捏造 Observation(即对工具结果的幻觉)。现代框架通常会处理此问题,但仍需留意。 -
清理垃圾(上下文管理)
在长对话中,删除旧的思考痕迹,只保留最终答案。否则,代理会很快耗尽“RAM”(上下文窗口)记忆。
ReAct 工作流示例
-
编程代理(如 Devin 或 Copilot)
“写代码 → 运行 → 看到错误 → 思考如何修复 → 重写”。 -
二级技术支持
“阅读工单 → 查看服务器状态 → 查看用户日志 → 交叉数据 → 回复”。 -
金融分析师
“查询当前价格 → 查询最新新闻 → 与历史数据比较 → 生成建议”。
编码愉快! 🤖