1
2
3
4
5

from llama_index.core.workflow import Event

class NewsExtractedEvent(Event):
    content: str  
    xxx: int 

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

from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Event

class ProcessedDataEvent(Event):
    data: str

class MyWorkflow(Workflow):
    @step
    async def ingest(self, ev: StartEvent) -> ProcessedDataEvent:
        
        input_data = ev.get("input")
        return ProcessedDataEvent(data=input_data.upper())

    @step
    async def finish(self, ev: ProcessedDataEvent) -> StopEvent:
        
        return StopEvent(result=f"处理完成: {ev.data}")


wf = MyWorkflow()
await wf.run(input="hello world") 

1
2
3
4
5
6
7
8

!pip install -U uv





!uv pip install llama-index llama-index-llms-groq llama-index-llms-openai-like feedparser requests transformers

1
2
3
4
5
6
7
8

import os
from llama_index.llms.groq import Groq
from llama_index.core import Settings

os.environ["GROQ_API_KEY"] = user_secrets.get_secret("GROQ_API_KEY")


Settings.llm = Groq(model="llama-3.3-70b-versatile")

1
2
3
4
5
6
7
8

from llama_index.llms.openai_like import OpenAILike

Settings.llm = OpenAILike(
    model="Qwen/Qwen3-8B",
    api_key=user_secrets.get_secret("SILICON-API_KEY"),
    api_base="https://api.siliconflow.cn/v1",
    is_chat_model=True
)

1
2
3
4
5

from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Context, Event

class NewBlogPostEvent(Event):
    title: str
    link: str

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

from llama_index.core.agent.workflow import ReActAgent

CHECKER_SYSTEM_PROMPT = """
### ROLE
咱们是一个极其精准的数据提取专家。

### TASK
1. 调用工具获取最新博客文章。
2. 提取其标题和链接。
3. **必须**以 JSON 格式输出。

### OUTPUT_FORMAT
{"title": "文章标题", "link": "文章链接"}
"""
checker = ReActAgent(
            name="checker",
            system_prompt=CHECKER_SYSTEM_PROMPT,
            tools=[tool_get_blog],
            llm=Settings.llm
        )

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

NOTIFIER_SYSTEM_PROMPT = """
### ROLE
咱们是一个通知官。

### TASK
请根据提供的信息发送通知。
直接调用工具发送，不要解释咱们的行为。发送成功后回复“已发出”即可。
"""
notifier = ReActAgent(
            name="notifier",
            system_prompt=NOTIFIER_SYSTEM_PROMPT,
            tools=[tool_send_all],
            llm=Settings.llm
        )

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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

import os
import re
import json

global_last_title = ""



class DemoBlogMonitor(Workflow):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.checker = checker

        self.notifier = notifier

    @step
    async def check_step(self, ev: StartEvent) -> NewBlogPostEvent | StopEvent:
        global global_last_title
        rss_url = ev.get("rss_url")

        print(f"--- [Checker] 目标: {rss_url} ---")
        print(f"--- [Checker] 基准: '{global_last_title}' ---")

        response = await self.checker.run(user_msg=f"请提取此 RSS 的最新数据：{rss_url}")
        res_text = str(response)
        print(f"DEBUG [Agent Output]: {res_text}")

        try:
            json_match = re.search(r"(\{.*\})", res_text, re.DOTALL)
            if not json_match:
                raise ValueError("未在回复中找到 JSON 块")

            data = json.loads(json_match.group(1))
            current_title = data.get("title", "").strip()
            current_link = data.get("link", "").strip()

            if not current_title or not current_link:
                raise ValueError("JSON 数据缺失字段")

            
            if current_title != global_last_title:
                print(f"发现新动态: {current_title}")
                
                return NewBlogPostEvent(title=current_title, link=current_link)
            else:
                return StopEvent(result=f"判定：内容未更新")

        except Exception as e:
            
            print(f"WARN: JSON 解析失败 ({e}), 尝试正则兜底...")
            link_match = re.search(r"(https?://\S+)", res_text)
            if link_match:
                link = link_match.group(1).strip(" ”\"。")
                
                title_match = re.search(r"[是“\"'](.*?)[”\"']", res_text)
                title = title_match.group(1) if title_match else "未识别标题"

                if title != global_last_title:
                    return NewBlogPostEvent(title=title, link=link)

            return StopEvent(result=f"解析彻底失败。原文: {res_text}")

    @step
    async def notify_step(self, ev: NewBlogPostEvent) -> StopEvent:
        global global_last_title
        print(f"--- [Notifier] 正在发送: {ev.title} ---")

        await self.notifier.run(user_msg=f"发送通知：标题《{ev.title}》，链接 {ev.link}")

        
        global_last_title = ev.title
        return StopEvent(result=f"完成。新基准已设为: {ev.title}")

1
2
3
4
5
6
7
8
9

async def main():
    monitor = DemoBlogMonitor(timeout=120)

    print("\n>>> Run #1")
    await monitor.run(rss_url="https://blog.algieba12.cn/atom.xml")

    print("\n>>> Run #2")
    await monitor.run(rss_url="https://blog.algieba12.cn/atom.xml")
await main()