LlamaIndex ReAct框架实战：全自动博客监控Agent开发指南

核心摘要 (TL;DR)

• 认知升级：从“缸中之脑”到“智能体 (Agent)”，理解大模型如何通过工具感知并影响现实世界。
• 原理解析：剖析 ReAct (Reasoning and Acting) 范式的核心流转机制。
• 实战目标：利用 LlamaIndex 构建一个全自动博客监控 Agent，赋予它获取博客更新、发送邮件和微信通知的“手脚”。

前言

在上一节我们快速了解了QLoRA微调技术，进行了一个简单的模型认知微调。微调可以说是折腾模型时的最后手段，其中细节和坑点都值得细聊，但篇幅有限，先快速过一遍。后续如果有机会，可以单独开个专栏深入微调。眼下，重点还是先把大模型的实战技术整个脉络理清楚。

好了，闲话少说。这一节咱们来做Agent，让它帮忙干点活儿。任务驱动，学起来更有目标，也更愉快。本篇博客的目的很直接：做一个能监控博客更新，并在更新时通知咱们的Agent。

1. Agent是啥？

通过前面的介绍，我们已经清楚：大模型本身只是一台拥有训练时数据的词语接龙机器。换个比喻——它像一个被关在囚牢里的智者，对外界一无所知，也无法改变外界。如果让它写写文章、讲讲故事，还能应付；但如果让它去查一下今天的金价，它只能回答“抱歉，作为一个对话大模型，我没有上网的能力”，或者干脆开始瞎编。

而Agent（智能体）呢？相当于给这颗“缸中之脑”装上了四肢和五感，让它能获取信息、操作实际事物。于是它从“建议者”变成了“行动者”，成为一个真正能干活儿的助手。

1.1 Agent的组件

Agent = 大模型（大脑）+ 定义的Tools（手脚双眼）+ 上下文记忆

• 大模型：作为核心，负责听懂自然语言指令（即咱们的输入），进行逻辑推理，并决定调用哪个工具。
• Tools：为Agent定义与外界交互的手段。需要自己定义，当然也有其他方式。
• 上下文记忆：用于推进复杂任务。很多时候Agent需要记住上一步的结果，再根据结果执行下一步。

1.2 ReAct范式

Agent的开发有很多范式，今天咱们用最简单、最快捷的一种——ReAct模式。

ReAct，全称Reasoning and Acting，字面意思就是“推理”+“行动”。这种范式最直观，因为它模拟了人类解决问题的过程：

• Thought（思考）：大模型先分析问题目标，思考下一步需要什么信息。对应本案例：“检查阿尔的代码屋最新博客，需要一个获取最新博客的工具”。
• Action（行动）：大模型决定调用哪个工具，并生成相应的参数去执行。
• Observation（观察）：工具执行完毕，把结果返回给大模型。大模型根据结果进行下一步思考、下一次行动、下一次观察……如此循环。

2. Kaggle实操

2.1 工具

开始干活之前，先明确这个任务需要至少两个工具：

• 获取最新博客的工具：阿尔的代码屋支持RSS订阅，可以直接用Python的feedparser库解决。感兴趣的同志可以去feedparser的GitHub深入了解这个优雅的RSS解析库。
• 发送通知的工具：本案例尝试两种通知——邮件和微信（其实一种就够了，多分享一种方法，方便以后自己扩展）。
  • 邮件通知：不打算配置复杂的SMTP，直接用resend API，注册后生成一个Key即可使用，方便快捷。
  • 微信通知：考虑到很多人看微信消息更多，这里用Server酱推送微信通知。不过免费额度每天只有5条，对咱们的监控任务来说绰绰有余。

2.2 注册并获取用于发邮件通知的Resend Api Key

• 进入官网 https://resend.com/
• 点击注册，使用Google账户或GitHub账户，当然也可以用普通邮箱注册

• 点击左侧的API Keys → Create API key

• 找个安全的地方记录下来，后面配置到项目里。

2.3 注册并获取用于发微信通知的Server酱 Api Key

• 进入官网 https://sct.ftqq.com/
• 微信扫码登录

免费额度每天5条
Server酱支持在线测试，可以消耗一次当日额度试试通知效果。

2.4 安装库环境

• 导入模型input：可以在Qwen3-8B-unsloth-bnb-4bit这个input中拿到模型。因为unsloth的量化模型不在Kaggle官方模型文件中，已经下载好作为input。照例在右侧边栏点击input，添加这个模型的input。

• 安装库：显卡选择T4×2就足够。用uv管理Python库，运行以下命令：

!pip install uv
!uv pip install feedparser llama-index requests llama-index-llms-huggingface
!uv pip install -U transformers peft accelerate bitsandbytes
!uv pip install --reinstall torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121 --system

Kaggle默认的transformers版本有点老，直接升级。安装完成后，一定要点击重启并清理(Restart & clear all cell outputs)，让新库生效，否则仍会使用旧库。

2.5 配置Api Keys

之前注册的用于发送通知的Api Key属于隐私资产，不能直接写在代码中，需要用密钥方式加到环境里。

• 点击Add-ons中的Secrets，在其左下角点击Add Secret按钮。
• LABEL中先填写EMAIL_API_KEY，VALUE中填上Resend获取的Api Key。
• 同样将微信通知的Key填为WECHAT_API_KEY。
• 另外，接收通知的邮箱也属于隐私信息，一并配置为TARGET_EMAIL。

配置完成后，确认新加的这几个Secrets都已勾选。这些Secrets在项目间可见，添加后其他项目也能通过勾选激活。

Kaggle也给出了获取Secrets的代码示例：

from kaggle_secrets import UserSecretsClient
user_secrets = UserSecretsClient()
secret_value_0 = user_secrets.get_secret("EMAIL_API_KEY")
secret_value_1 = user_secrets.get_secret("TARGET_EMAIL")
secret_value_2 = user_secrets.get_secret("WECHAT_API_KEY")

后面会用到。

2.6 编写获取博客信息的函数get_latest_blog_post

import feedparser

def get_latest_blog_post(rss_url:str)->str:
    """
    用于请求并解析目标博客的 RSS feed，获取最新的一篇博客文章信息。
    当咱们需要检查博客是否有更新时，必须首先调用此工具。
    """
    print(f"【calling】 get_latest_blog_post with rss_url:{rss_url}")
    try:
        feed = feedparser.parse(rss_url)
        if feed.entries:
            latest_entry = feed.entries[0]
            print(f"debug: title {latest_entry.title} link:{latest_entry.link}")

            return (
                f"Title: {latest_entry.title}n"
                f"Link: {latest_entry.link}nn"
                "【系统强制指令】：请立刻将上面的 Title 与咱们已知最新标题进行严格比对！n"
                "情况 A：如果 Title 与已知标题【完全相同】，请直接输出 Answer: 无更新，结束任务。n"
                "情况 B：如果 Title 与已知标题【不一致】，咱们绝对不能直接输出 Answer！咱们必须严格按以下格式输出，以调用通知工具：nn"
                "Thought: 标题不一致，我必须立刻调用通知工具。n"
                "Action: send_all_notificationsn"
                "Action Input: {"post_title": "完全复制上面的Title", "post_link": "完全复制上面的Link"}n"
            )

        return f"No posts found in rss feed {rss_url}"
    except Exception as e:
        return f"Exception: {str(e)} on fetching blog"

代码通过feedparser获取最新博客的title和link，中间打印了调试信息。但返回值里塞了一大段像提示词的内容？这里先卖个关子，后面解密。

调用测试一下：

get_latest_blog_post("https://blog.algieba12.cn/atom.xml")

结果如下（最新博客可能已变化，但能获取到即可）：

【calling】 get_latest_blog_post with rss_url:https://blog.algieba12.cn/atom.xml
debug: title 在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋 link:https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
'Title: 在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋nLink: https://blog.algieba12.cn/llm06-unsloth-qlora-ft/nn【系统强制指令】：请立刻将上面的 Title 与咱们已知最新标题进行严格比对！n情况 A：如果 Title 与已知标题【完全相同】，请直接输出 Answer: 无更新，结束任务。n情况 B：如果 Title 与已知标题【不一致】，咱们绝对不能直接输出 Answer！咱们必须严格按以下格式输出，以调用通知工具：nnThought: 标题不一致，我必须立刻调用通知工具。nAction: send_all_notificationsnAction Input: {"post_title": "完全复制上面的Title", "post_link": "完全复制上面的Link"}n'

2.7 定义发送邮件的函数send_email_notification

import requests
def send_email_notification(post_title:str, post_link:str)->str:
    """
    当发现博客有更新时，必须调用此工具发送邮件通知。
    参数 post_title: 新博客的标题
    参数 post_link: 新博客的链接
    """
    print(f"【calling】 send_email_notification with post_title {post_title} link {post_link}")
    target_email= user_secrets.get_secret("TARGET_EMAIL")
    email_api_key = user_secrets.get_secret("EMAIL_API_KEY")
    headers = {
        "Authorization": f"Bearer {email_api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "from" : "onboarding@resend.dev",
        "to":target_email,
        "subject":f"阿尔的代码屋更新咯：{post_title}",
        "text":f"检测到 阿尔的代码屋 更新了一篇新博客 nn 标题：{post_title}]n 链接: {post_link}"
    }

    try:
        response = requests.post("https://api.resend.com/emails", headers=headers, json=payload)
        if response.status_code == 200:
            return "Email sent successfully via API"
        return f"Email sent failed. {response.text}"
    except Exception as e:
        return f"Exception: {str(e)} on sending email"

这里使用了之前Secret代码示例获取收信邮箱和API Key。注意from字段是Resend官方的邮箱，如果希望用自己的邮箱发邮件，需要用SMTP。大家可以自行探索，有兴趣的话后面可以写篇博客介绍。

测试一下：

send_email_notification(post_title="email notification test",post_link="none")

能收到邮件就OK。

2.8 定义发送微信通知的函数

def send_wechat_notification(post_title:str, post_link:str)->str:
    """
    当发现博客有更新时，必须调用此工具发送微信通知。
    参数 post_title: 新博客的标题
    参数 post_link: 新博客的链接
    """
    print(f"【calling】 send_wechat_notification with title:{post_title} link: {post_link}")
    wechat_api_key = user_secrets.get_secret("WECHAT_API_KEY")
    url = f"https://sctapi.ftqq.com/{wechat_api_key}.send"
    data = {
        "title":f"阿尔的代码屋更新咯：{post_title}",
        "desp":f"检测到 阿尔的代码屋 更新了一篇新博客 nn 标题：{post_title}]n 链接: {post_link}"
    }
    try:
        response = requests.post(url,data=data)
        if response.status_code != 200:
            return f"Message sent failed. {response.text}"
        result = response.json()
        if result.get("code") != 0:
            return f"Failed to send notification. API response: {result.get('message')}"
        return "Message sent successfully via API"

    except Exception as e:
        return f"Exception: {str(e)} on sending wechat notification"

这个函数跟邮件类似，就不重复测试了。

2.9 合并通知函数

经过几轮测试发现，让当前模型调用太多工具的效果不好。这里先把所有通知函数合并到一个函数中：

def send_all_notifications(post_title: str, post_link: str) -> str:
    """
    当发现博客有更新时，必须调用此工具。调用此工具会自动同时发送邮件和微信通知。
    参数 post_title: 新博客的标题
    参数 post_link: 新博客的链接
    """
    print(f"【Agent】 触发了联合通知工具: {post_title}")

    email_res = send_email_notification(post_title, post_link)
    wechat_res = send_wechat_notification(post_title, post_link)

    return f"邮件通知结果: {email_res} | 微信通知结果: {wechat_res}"

2.10 包装工具

将这些工具函数包装成LlamaIndex能识别的FunctionTool，供Agent使用。

from llama_index.core.tools import FunctionTool
tool_get_blog = FunctionTool.from_defaults(fn=get_latest_blog_post)
tool_send_all = FunctionTool.from_defaults(fn=send_all_notifications)

2.11 定义配置

把接下来要用到的配置先定义好：

• 要监控的RSS链接
• 维护一个最新标题，用于判断博客是否更新
• 当前大模型路径

rss_link = "https://blog.algieba12.cn/atom.xml"
last_title = ""
local_model_path = "/kaggle/input/datasets/algieba12/qwen3-8b-unsloth-bnb-4bit/Qwen3-8B-unsloth-bnb-4bit/"

2.12 导入大模型

import torch

from llama_index.core import Settings
from llama_index.llms.huggingface import HuggingFaceLLM
from llama_index.core.agent.workflow import ReActAgent
from llama_index.core.workflow import Context

Settings.llm = HuggingFaceLLM(
    model_name=local_model_path,
    tokenizer_name=local_model_path,
    context_window=8192,
    max_new_tokens=4096,
    generate_kwargs={
        "do_sample":False, # 关闭采样，直接贪婪解码，不随机，用可能性最高的结果
    },
    device_map="auto",
    model_kwargs={
        "dtype":torch.float16,
        "trust_remote_code":True,
    }
)

为了求稳，直接关闭采样，每次返回的结果必然一致。当然也可以用温度控制，建议设定在0.1以下。

2.13 定义ReActAgent

重头戏来了：

agent = ReActAgent(
    name="blog_monitor_agent",
    description="自动监控博客并发送通知",
    system_prompt=(
        "咱们是一个极其严谨的后台监控机器人，严格遵循 Thought-Action-Observation 循环。n"
        "咱们的任务是检查博客更新，并在有更新时发送通知。nn"
        "【操作规范与强制要求】n"
        "步骤 1：咱们必须首先调用 get_latest_blog_post 工具获取真实数据。n"
        "步骤 2：仔细阅读 Observation 返回的内容。如果获取到的真实标题与系统已知标题不同，说明有更新。n"
        "步骤 3：如果有更新，咱们必须立刻调用 send_all_notifications 工具。nn"
        "【 **致命错误警告** 】n"
        "当调用 send_all_notifications 时，传入的 post_title 和 post_link 参数必须 100% 完全复制自 get_latest_blog_post 返回的真实 Observation 数据！n"
        "绝对禁止凭空捏造参数！绝对禁止使用诸如“新文章”、“test”、“新链接”之类的占位符！必须原样提取真实文本！n"
        "只有当两个工具都执行完毕，且通知发送成功后，咱们才能输出最终的 Answer 结束任务。"
    ),
    tools=[tool_get_blog, tool_send_all],
    llm=Settings.llm,
    max_iterations=8,  # 允许足够的步数进行多轮工具调用
    verbose=True
)

ReActAgent的name和description不重要，但system_prompt至关重要，需要将智能体要做的事情尽可能清晰地描述出来。tools参数配置两个FunctionTool，开启verbose以便观察agent的执行步骤。

2.14 运行函数

import re

async def main():
    global last_title

    task_prompt = (
        f"请去检查博客 RSS：'{rss_link}'。目前系统已知最新标题是 '{last_title}'。n"
        "请严格按以下逻辑执行：n"
        "1. 立即获取最新的真实博客信息。n"
        "2. 拿到真实信息后，与已知标题进行比对。n"
        "3. 若发现标题不一致，必须将刚才获取到的真实标题和真实链接提取出来，准确无误地传给通知工具进行发送。"
    )
    ctx = Context(agent)
    try:
        response = await agent.run(user_msg=task_prompt, ctx=ctx)
        result_text = response.response.content.strip()
        print(f"Agent 执行完毕，返回信息: {result_text}")

        import feedparser
        feed = feedparser.parse(rss_link)
        if feed.entries:
            current_actual_title = feed.entries[0].title
            if current_actual_title != last_title:
                last_title = current_actual_title
                print(f"系统内部状态已更新，最新标题为：{last_title}")

    except Exception as e:
        print(f"Exception: {str(e)}")

这里的task_prompt相当于平时使用大模型的提示词，简要描述了任务。Context用于保留模型对话记忆，但由于每次扫描都是独立的，最新标题已经维护在变量里，所以每次调用main函数都重新生成上下文。

因为这是异步函数，通过await调用：

await main()

结果如下（输出日志）：

Running step init_run
Step init_run produced event 
Running step setup_agent
Step setup_agent produced event 
Running step run_agent_step
Step run_agent_step produced event 
Running step parse_agent_output
Running step call_tool
Step parse_agent_output produced event 
【calling】 get_latest_blog_post with rss_url:https://blog.algieba12.cn/atom.xml
debug: title 在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋 link:https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
Step call_tool produced event 
Running step aggregate_tool_results
Step aggregate_tool_results produced event 
Running step setup_agent
Step setup_agent produced event 
Running step run_agent_step
Step run_agent_step produced event 
Running step parse_agent_output
Running step call_tool
Step parse_agent_output produced event 
【Agent】 触发了联合通知工具: 在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋
【calling】 send_email_notification with post_title 在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋 link https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
【calling】 send_wechat_notification with title:在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋 link: https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
Step call_tool produced event 
Running step aggregate_tool_results
Step aggregate_tool_results produced event 
Running step setup_agent
Step setup_agent produced event 
Running step run_agent_step
Step run_agent_step produced event 
Running step parse_agent_output
Step parse_agent_output produced event 
Agent 执行完毕，返回信息: 已成功发送邮件和微信通知，标题为“在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋”，链接为https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
系统内部状态已更新，最新标题为：在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋

可以看到工具函数都被成功调用。

2.15 解密

在定义获取最新博客的函数时，返回值里塞了一大段提示词。这种技术叫做观察结果劫持。因为咱们用的模型不大（8B级别），能力有限，容易忘记后续步骤，或者“早退”——拿到最新博客后就直接退出，以为自己已经完成发送，忘了去调用通知。

通过在获取函数中反复提醒它进行检查和通知，让它能够完整地执行任务。

3. 代码

所有代码可以在本教程对应的Kaggle笔记本中获取。后面还写了些测试案例做正反面测试和稳定性测试，有兴趣可以参考。当前的实现可以稳定完成监控和通知任务。

4. 常见问题 (Q&A)

Q1: 代码中的“观察结果劫持”是什么高级操作？如果用GPT-4还需要这么写吗？
A: 这是一个针对中小参数模型（如8B级别）的“工程化妥协”。

• 中小模型痛点：多步逻辑推理时容易发生“幻觉”或提前终止任务（比如看到博客标题就以为完事了，忘了发通知）。加上长上下文中难以记住初始指令，甚至忘记标准输出格式。
• 劫持原理：把提示词强行塞进工具的返回值（Observation）里，附上严格的ReAct语法模板。相当于每一步都在耳边“敲黑板划重点”，逼迫它按格式填空。
• 大模型对比：如果用GPT-4或Claude 4.5这类模型，它们逻辑规划能力极强，通常不需要这种劫持，直接返回干净数据即可。

Q2: 加载模型时为什么要设置do_sample=False？开启随机采样会怎样？
A: Agent执行任务需要极其严格的格式输出。

• 开启随机采样 (do_sample=True)：带温度的采样会让模型在生成JSON格式（如Action Input）时“自由发挥”，容易漏括号或拼错工具名，导致解析失败、Agent崩溃。
• 关闭随机采样 (do_sample=False)：使用贪婪解码，最大程度保证模型按最稳妥、概率最高的路径输出结构化文本，大幅提升稳定性。

Q3: 运行出现The following generation flags are not valid...警告需要处理吗？
A: 这是因为生成策略参数发生逻辑冲突，建议清理。

• 参数冲突：当设置了do_sample=False（关闭采样）时，模型不再有随机性。此时传入的temperature、top_p、top_k等控制发散程度的参数已经无效。
• 最佳实践：为保持代码纯净并消除底层框架的警告日志，直接从generate_kwargs字典中删除这些控制随机性的参数即可。

Q4: 为什么要把发送邮件和发送微信合并成一个send_all_notifications工具？
A: 同样是为了照顾8B模型的能力上限。

• 决策复杂度：Agent拥有的工具越多，它在Thought阶段要做的决策就越复杂。
• 翻车风险：如果拆开，模型需要先调邮件工具，观察结果后再调微信工具。链路越长，小模型断片或翻车的概率呈指数级上升。
• 最佳实践：将功能高度相关的动作封装成一个高级别工具（Macro Tool），是提升小模型Agent稳定性的绝佳手段。如果不合并，就必须依赖极其严苛的系统提示词和观察结果劫持。

Q5: 为什么Agent会把Prompt里的示例（如“新文章”、“新链接”）当成真实参数传给工具？
A: 这是小模型在ReAct框架中的“死记硬背（Overfitting）”现象。

• 伪代码误导：如果在系统提示词中给出了带有假数据的Few-Shot示例，小模型很容易将其当成标准答案直接照抄，而不是从上一步的Observation里提取真实数据。
• 解决思路：去掉提示词中的假定值，改用“致命错误警告”这种强烈的上下文刺激，明确要求它“必须100%完全复制上一步的真实返回结果”。

Q6: 为什么明明设置了timeout=120，模型却像死机一样一直卡着，不触发超时机制？
A: 这涉及Python异步机制和PyTorch底层C++执行的根本冲突。

• CUDA上下文死锁：在Kaggle等环境中，如果使用asyncio.to_thread将本地模型的同步推理任务推到后台线程，极易引发GPU资源的跨线程调度冲突。底层C++算子一旦锁死，Python主线程的超时设定根本无力干预。
• 有效解法：在单路自动化工作流中，直接在主线程中同步调用模型（会短暂冻结进程但不会死机）。或者更彻底地，使用vLLM将模型部署为独立的本地API服务，让工作流通过HTTP请求调用，实现完美的进程隔离和超时控制。

Q7: 在Kaggle中执行包安装后报错RuntimeError: operator torchvision::nms does not exist怎么办？
A: 这是一个环境破坏导致的底层C++链接崩溃问题。

• 根本原因：新库更新覆盖了Kaggle环境中原有的依赖，把预装的PyTorch和Torchvision的底层链接带崩了，导致后续所有涉及transformers导包的代码全部崩溃。
• 修复步骤：新建代码块运行强制重装指令（uv pip install --reinstall torch torchvision torchaudio）来重新对齐底层库。安装完毕后，必须点击Run → Restart Session重启内核，否则内存中加载的还是损坏的旧库。

Q8: 这个代码能在Kaggle里7×24小时一直跑吗？
A: 不太现实。本教程旨在演示Agent的核心逻辑与搭建方法，并非生产级托管方案。

• Kaggle限制：Kaggle Kernel有最长12小时的运行时间限制，且长时间无交互会自动断开休眠。
• 生产级建议：如果希望真正24小时监控，建议将这段代码部署到轻量级云服务器（如1核2G实例），结合系统的crontab定时任务，或者在Python脚本外层包裹一个while True: sleep(3600)循环即可。

Q9: 如果我把代码保存在普通的.py文件里，怎么运行这个异步的main函数？
A: 运行环境不同，调用异步函数的方式也不同。

• 在Jupyter Notebook / Kaggle中：环境本身已经运行在一个异步事件循环中（自带Event Loop），所以可以直接使用await main()来运行。
• 在普通的Python脚本 (.py) 中：标准Python脚本里直接写await会报错。需要引入asyncio库，使用asyncio.run()启动事件循环。代码结尾应该改成：

import asyncio

# ... main 函数和其他代码 ...

if __name__ == "__main__":
    asyncio.run(main())