1
2
3
4

!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

1
2
3
4
5

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")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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}\n\n"
                "【系统强制指令】：请立刻将上面的 Title 与咱们已知最新标题进行严格比对！\n"
                "情况 A：如果 Title 与已知标题【完全相同】，请直接输出 Answer: 无更新，结束任务。\n"
                "情况 B：如果 Title 与已知标题【不一致】，咱们绝对不能直接输出 Answer！咱们必须严格按以下格式输出，以调用通知工具：\n\n"
                "Thought: 标题不一致，我必须立刻调用通知工具。\n"
                "Action: send_all_notifications\n"
                "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"

1

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

1
2
3

【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/\n\n【系统强制指令】：请立刻将上面的 Title 与咱们已知最新标题进行严格比对！\n情况 A：如果 Title 与已知标题【完全相同】，请直接输出 Answer: 无更新，结束任务。\n情况 B：如果 Title 与已知标题【不一致】，咱们绝对不能直接输出 Answer！咱们必须严格按以下格式输出，以调用通知工具：\n\nThought: 标题不一致，我必须立刻调用通知工具。\nAction: send_all_notifications\nAction Input: {"post_title": "完全复制上面的Title", "post_link": "完全复制上面的Link"}\n'

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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" : "[email protected]",
        "to":target_email,
        "subject":f"阿尔的代码屋更新咯：{post_title}",
        "text":f"检测到 阿尔的代码屋 更新了一篇新博客 \n\n 标题：{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"

1

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

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"检测到 阿尔的代码屋 更新了一篇新博客 \n\n 标题：{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"

1
2
3
4
5
6
7
8
9
10
11
12

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}"

1
2
3

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)

1
2
3

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/"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

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,
    }
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

agent = ReActAgent(
    name="blog_monitor_agent",
    description="自动监控博客并发送通知",
    system_prompt=(
        "咱们是一个极其严谨的后台监控机器人，严格遵循 Thought-Action-Observation 循环。\n"
        "咱们的任务是检查博客更新，并在有更新时发送通知。\n\n"
        "【操作规范与强制要求】\n"
        "步骤 1：咱们必须首先调用 get_latest_blog_post 工具获取真实数据。\n"
        "步骤 2：仔细阅读 Observation 返回的内容。如果获取到的真实标题与系统已知标题不同，说明有更新。\n"
        "步骤 3：如果有更新，咱们必须立刻调用 send_all_notifications 工具。\n\n"
        "【 **致命错误警告** 】\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
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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)}")

1

await main()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

Running step init_run
Step init_run produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentInput'>
Running step setup_agent
Step setup_agent produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentSetup'>
Running step run_agent_step
Step run_agent_step produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentOutput'>
Running step parse_agent_output
Running step call_tool
Step parse_agent_output produced event <class 'NoneType'>
【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 <class 'llama_index.core.agent.workflow.workflow_events.ToolCallResult'>
Running step aggregate_tool_results
Step aggregate_tool_results produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentInput'>
Running step setup_agent
Step setup_agent produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentSetup'>
Running step run_agent_step
Step run_agent_step produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentOutput'>
Running step parse_agent_output
Running step call_tool
Step parse_agent_output produced event <class 'NoneType'>
【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 <class 'llama_index.core.agent.workflow.workflow_events.ToolCallResult'>
Running step aggregate_tool_results
Step aggregate_tool_results produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentInput'>
Running step setup_agent
Step setup_agent produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentSetup'>
Running step run_agent_step
Step run_agent_step produced event <class 'llama_index.core.agent.workflow.workflow_events.AgentOutput'>
Running step parse_agent_output
Step parse_agent_output produced event <class 'workflows.events.StopEvent'>
Agent 执行完毕，返回信息: 已成功发送邮件和微信通知，标题为“在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋”，链接为https://blog.algieba12.cn/llm06-unsloth-qlora-ft/
系统内部状态已更新，最新标题为：在 Kaggle 上用 Unsloth 极速微调 Qwen3 - 大模型实战 06 | 阿尔的代码屋

1
2
3
4
5
6

import asyncio



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