内容创作者

ContentCreator 简化了在 LinkedIn 和 Twitter 上规划、创建和分发引人入胜的内容的过程。

创建一个名为 config.py 的文件，其中包含以下代码：

config.py

import os
from enum import Enum

from dotenv import load_dotenv

load_dotenv()


TYPEFULLY_API_URL = "https://api.typefully.com/v1/drafts/"
TYPEFULLY_API_KEY = os.getenv("TYPEFULLY_API_KEY")
HEADERS = {"X-API-KEY": f"Bearer {TYPEFULLY_API_KEY}"}


# 定义枚举
class PostType(Enum):
    TWITTER = "Twitter"
    LINKEDIN = "LinkedIn"

在 prompts.py 中添加提示

prompts.py

# 规划师代理配置
agents_config = {
    "blog_analyzer": {
        "role": "博客分析师",
        "goal": "分析博客并识别关键想法、部分和技术概念",
        "backstory": (
            "您是一位拥有多年撰写、编辑和审查技术博客经验的技术撰稿人。 "
            "您擅长理解和记录技术概念。\n\n"
        ),
        "verbose": False,
    },
    "twitter_thread_planner": {
        "role": "Twitter 线索规划师",
        "goal": "根据提供的博客分析创建 Twitter 线索计划",
        "backstory": (
            "您是一位拥有多年将长篇技术博客转换为 Twitter 线索经验的技术撰稿人。 "
            "您擅长将长篇内容分解成引人入胜且信息丰富的简短推文。 "
            "并识别可与推文关联的相关媒体 URL。\n\n"
        ),
        "verbose": False,
    },
    "linkedin_post_planner": {
        "role": "LinkedIn 帖子规划师",
        "goal": "根据提供的博客分析创建引人入胜的 LinkedIn 帖子",
        "backstory": (
            "您是一位拥有丰富撰写技术性 LinkedIn 内容经验的技术撰稿人。 "
            "您善于将技术概念提炼成清晰、权威的帖子，以引起专业受众的共鸣，"
            "同时保持技术的准确性。您知道如何平衡技术深度与易懂性，并整合"
            "相关的标签和提及以最大化参与度。\n\n"
        ),
        "verbose": False,
    },
}

# 规划师任务配置
tasks_config = {
    "analyze_blog": {
        "description": (
            "分析 {blog_path} 的 markdown 文件，创建一个以开发者为中心的技术概述\n\n"
            "1. 梳理博客讨论的核心思想\n"
            "2. 识别关键部分及其内容\n"
            "3. 对于每个部分，提取所有出现在图片 markdown 语法 ![](image_url) 中的 URL\n"
            "4. 必须将这些识别出的图片 URL 与其对应的部分关联起来，以便我们可以在推文中用作媒体。\n\n"
            "重点关注对全面理解博客很重要的细节。\n\n"
        ),
        "expected_output": (
            "一份技术分析，包含：\n"
            "- 博客标题和核心概念/思想\n"
            "- 识别出的关键技术部分及其要点\n"
            "- 涵盖的重要代码示例或技术概念\n"
            "- 开发者的关键收获\n"
            "- 与关键部分相关且可与推文关联的相关媒体 URL，必须这样做。\n\n"
        ),
    },
    "create_twitter_thread_plan": {
        "description": (
            "根据提供的博客分析开发一个引人入胜的 Twitter 线索，并严格遵循 {path_to_example_threads} 中提供的写作风格\n\n"
            "该线索应将复杂的技术概念分解为易于理解的、适合推文的块，"
            "在保持技术准确性的同时使其易于理解。\n\n"
            "计划应包括：\n"
            "- 一个引人注目的开场推文，应在 10 个字以内，必须与博客标题相同\n"
            "- 从基础到高级概念的逻辑流程\n"
            "- 适合 Twitter 格式的代码片段或关键技术亮点\n"
            "- 与关键部分相关且必须与其对应推文关联的相关媒体 URL\n"
            "- 为工程受众提供清晰的收获\n\n"
            "请确保涵盖：\n"
            "- 要解决的核心问题\n"
            "- 关键技术创新或方法\n"
            "- 有趣的实现细节\n"
            "- 实际应用或优势\n"
            "- 结论号召行动\n"
            "- 添加与推文相关的相关 URL\n\n"
            "专注于创造技术受众认为有价值的叙述，"
            "同时保持每条推文的简洁、易懂和有影响力。\n\n"
        ),
        "expected_output": (
            "一个包含推文列表的 Twitter 线索，每条推文都包含：\n"
            "- 内容\n"
            "- 可能相关的媒体 URL\n"
            "- is_hook：如果推文是开场推文，则为 true，否则为 false\n\n"
        ),
    },
    "create_linkedin_post_plan": {
        "description": (
            "根据提供的博客分析开发一个全面的 LinkedIn 帖子\n\n"
            "该帖子应以专业、长篇的格式呈现技术内容，"
            "同时保持参与度和可读性。\n\n"
            "计划应包括：\n"
            "- 一个引人注目的开场白，应与博客标题相同\n"
            "- 一个好的结构体来分解技术内容\n"
            "- 适合 LinkedIn 商务受众的专业语气\n"
            "- 一个主要博客 URL，策略性地放置在帖子末尾\n"
            "- 巧妙使用换行和格式\n"
            "- 相关标签（最多 3-5 个）\n\n"
            "请确保涵盖：\n"
            "- 核心技术问题及其业务影响\n"
            "- 关键解决方案和技术方法\n"
            "- 实际应用和效益\n"
            "- 专业见解或经验教训\n"
            "- 明确的号召行动\n\n"
            "专注于创造与技术专业人士和商业领袖都能产生共鸣的内容，"
            "同时保持技术准确性。\n\n"
        ),
        "expected_output": (
            "一个包含以下内容的 LinkedIn 帖子计划：\n- 内容\n- 与帖子关联的主要博客 URL\n\n"
        ),
    },
}

对于调度逻辑，创建 scheduler.py

scheduler.py

import datetime
from typing import Any, Dict, Optional

import requests
from agno.utils.log import logger
from dotenv import load_dotenv
from pydantic import BaseModel

from cookbook.workflows.content_creator_workflow.config import (
    HEADERS,
    TYPEFULLY_API_URL,
    PostType,
)

load_dotenv()


def json_to_typefully_content(thread_json: Dict[str, Any]) -> str:
    """将 JSON 线索格式转换为 Typefully 的格式，每个推文之间有 4 个换行符。"""
    tweets = thread_json["tweets"]
    formatted_tweets = []
    for tweet in tweets:
        tweet_text = tweet["content"]
        if "media_urls" in tweet and tweet["media_urls"]:
            tweet_text += f"\n{tweet['media_urls'][0]}"
        formatted_tweets.append(tweet_text)

    return "\n\n\n\n".join(formatted_tweets)


def json_to_linkedin_content(thread_json: Dict[str, Any]) -> str:
    """将 JSON 线索格式转换为 Typefully 的格式。"""
    content = thread_json["content"]
    if "url" in thread_json and thread_json["url"]:
        content += f"\n{thread_json['url']}"
    return content


def schedule_thread(
    content: str,
    schedule_date: str = "next-free-slot",
    threadify: bool = False,
    share: bool = False,
    auto_retweet_enabled: bool = False,
    auto_plug_enabled: bool = False,
) -> Optional[Dict[str, Any]]:
    """在 Typefully 上调度线索。"""
    payload = {
        "content": content,
        "schedule-date": schedule_date,
        "threadify": threadify,
        "share": share,
        "auto_retweet_enabled": auto_retweet_enabled,
        "auto_plug_enabled": auto_plug_enabled,
    }

    payload = {key: value for key, value in payload.items() if value is not None}

    try:
        response = requests.post(TYPEFULLY_API_URL, json=payload, headers=HEADERS)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f"错误：{e}")
        return None


def schedule(
    thread_model: BaseModel,
    hours_from_now: int = 1,
    threadify: bool = False,
    share: bool = True,
    post_type: PostType = PostType.TWITTER,
) -> Optional[Dict[str, Any]]:
    """
    从 Pydantic 模型调度线索。

    参数：
        thread_model: 包含线索数据的 Pydantic 模型
        hours_from_now: 从现在开始的小时数用于调度线索（默认为 1）
        threadify: 是否让 Typefully 分割内容（默认为 False）
        share: 是否在响应中获取共享 URL（默认为 True）

    返回：
        API 响应字典或失败时返回 None
    """
    try:
        thread_content = ""
        # 将 Pydantic 模型转换为字典
        thread_json = thread_model.model_dump()
        logger.info("######## 线索 JSON：", thread_json)
        # 转换为 Typefully 格式
        if post_type == PostType.TWITTER:
            thread_content = json_to_typefully_content(thread_json)
        elif post_type == PostType.LINKEDIN:
            thread_content = json_to_linkedin_content(thread_json)

        # 计算调度时间
        schedule_date = (
            datetime.datetime.utcnow() + datetime.timedelta(hours=hours_from_now)
        ).isoformat() + "Z"

        if thread_content:
            # 调度线索
            response = schedule_thread(
                content=thread_content,
                schedule_date=schedule_date,
                threadify=threadify,
                share=share,
            )

            if response:
                logger.info("线索调度成功！")
                return response
            else:
                logger.error("未能调度线索。")
                return None
        return None

    except Exception as e:
        logger.error(f"错误：{str(e)}")
        return None

在 workflow.py 中定义工作流：

workflow.py

import json
from typing import List, Optional

from agno.agent import Agent, RunResponse
from agno.models.openai import OpenAIChat
from agno.run.response import RunEvent
from agno.tools.firecrawl import FirecrawlTools
from agno.utils.log import logger
from agno.workflow import Workflow
from dotenv import load_dotenv
from pydantic import BaseModel, Field

from cookbook.workflows.content_creator_workflow.config import PostType
from cookbook.workflows.content_creator_workflow.prompts import (
    agents_config,
    tasks_config,
)
from cookbook.workflows.content_creator_workflow.scheduler import schedule

# 加载环境变量
load_dotenv()


# 定义 Pydantic 模型来构建响应
class BlogAnalyzer(BaseModel):
    """
    表示来自博客分析师代理的响应。
    包含博客标题和 Markdown 格式的内容。
    """

    title: str
    blog_content_markdown: str


class Tweet(BaseModel):
    """
    表示 Twitter 线索中的单个推文。
    """

    content: str
    is_hook: bool = Field(
        default=False, description="标记此推文是否为“钩子”（第一条推文）"
    )
    media_urls: Optional[List[str]] = Field(
        default_factory=list, description="关联的媒体 URL，如果有的话"
    )  # type: ignore


class Thread(BaseModel):
    """
    表示包含多条推文的完整 Twitter 线索。
    """

    topic: str
    tweets: List[Tweet]


class LinkedInPost(BaseModel):
    """
    表示 LinkedIn 帖子。
    """

    content: str
    media_url: Optional[List[str]] = None  # 可选的媒体附件 URL


class ContentPlanningWorkflow(Workflow):
    """
    此工作流自动执行以下过程：
    1. 使用博客分析师代理抓取博客帖子。
    2. 根据抓取的内容为 Twitter 或 LinkedIn 生成内容计划。
    3. 调度和发布计划的内容。
    """

    # 此描述仅用于工作流 UI
    description: str = (
        "根据博客帖子规划、调度和发布社交媒体内容。"
    )

    # 博客分析师代理：提取博客内容（标题、部分）并将其转换为 Markdown 格式以供进一步使用。
    blog_analyzer: Agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        tools=[
            FirecrawlTools(scrape=True, crawl=False)
        ],  # 启用博客抓取功能
        description=f"{agents_config['blog_analyzer']['role']} - {agents_config['blog_analyzer']['goal']}",
        instructions=[
            f"{agents_config['blog_analyzer']['backstory']}",
            tasks_config["analyze_blog"][
                "description"
            ],  # 用于博客分析的任务特定说明
        ],
        response_model=BlogAnalyzer,  # 期望响应遵循 BlogAnalyzer Pydantic 模型
    )

    # Twitter 线索规划师：从博客内容创建 Twitter 线索，每条推文都简洁、引人入胜，
    # 并与相关媒体逻辑地连接起来。
    twitter_thread_planner: Agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        description=f"{agents_config['twitter_thread_planner']['role']} - {agents_config['twitter_thread_planner']['goal']}",
        instructions=[
            f"{agents_config['twitter_thread_planner']['backstory']}",
            tasks_config["create_twitter_thread_plan"]["description"],
        ],
        response_model=Thread,  # 期望响应遵循 Thread Pydantic 模型
    )

    # LinkedIn 帖子规划师：将博客内容转换为结构化的 LinkedIn 帖子，针对专业受众进行优化，并包含相关标签。
    linkedin_post_planner: Agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        description=f"{agents_config['linkedin_post_planner']['role']} - {agents_config['linkedin_post_planner']['goal']}",
        instructions=[
            f"{agents_config['linkedin_post_planner']['backstory']}",
            tasks_config["create_linkedin_post_plan"]["description"],
        ],
        response_model=LinkedInPost,  # 期望响应遵循 LinkedInPost Pydantic 模型
    )

    def scrape_blog_post(self, blog_post_url: str, use_cache: bool = True):
        if use_cache and blog_post_url in self.session_state:
            logger.info(f"使用缓存的博客帖子：{blog_post_url}")
            return self.session_state[blog_post_url]
        else:
            response: RunResponse = self.blog_analyzer.run(blog_post_url)
            if isinstance(response.content, BlogAnalyzer):
                result = response.content
                logger.info(f"博客标题：{result.title}")
                self.session_state[blog_post_url] = result.blog_content_markdown
                return result.blog_content_markdown
            else:
                raise ValueError("收到的博客分析师内容类型意外。")

    def generate_plan(self, blog_content: str, post_type: PostType):
        plan_response: RunResponse = RunResponse(content=None)
        if post_type == PostType.TWITTER:
            logger.info(f"正在为 {post_type} 生成帖子计划")
            plan_response = self.twitter_thread_planner.run(blog_content)
        elif post_type == PostType.LINKEDIN:
            logger.info(f"正在为 {post_type} 生成帖子计划")
            plan_response = self.linkedin_post_planner.run(blog_content)
        else:
            raise ValueError(f"不支持的帖子类型：{post_type}")

        if isinstance(plan_response.content, (Thread, LinkedInPost)):
            return plan_response.content
        elif isinstance(plan_response.content, str):
            data = json.loads(plan_response.content)
            if post_type == PostType.TWITTER:
                return Thread(**data)
            else:
                return LinkedInPost(**data)
        else:
            raise ValueError("收到的规划师内容类型意外。")

    def schedule_and_publish(self, plan, post_type: PostType) -> RunResponse:
        """
        利用 Typefully API 调度和发布内容。
        """
        logger.info(f"# 发布帖子类型的内容：{post_type}")

        # 直接使用 `scheduler` 模块来调度内容
        response = schedule(
            thread_model=plan,
            post_type=post_type,  # “Twitter”或“LinkedIn”
        )

        logger.info(f"响应：{response}")

        if response:
            return RunResponse(content=response, event=RunEvent.workflow_completed)
        else:
            return RunResponse(
                content="未能调度内容。", event=RunEvent.workflow_completed
            )

    def run(self, blog_post_url, post_type) -> RunResponse:
        """
        参数：
            blog_post_url：要分析的博客帖子的 URL。
            post_type：要生成的帖子类型（例如，Twitter 或 LinkedIn）。
        """
        # 抓取博客帖子
        blog_content = self.scrape_blog_post(blog_post_url)

        # 根据博客和帖子类型生成计划
        plan = self.generate_plan(blog_content, post_type)

        # 调度和发布内容
        response = self.schedule_and_publish(plan, post_type)

        return response


if __name__ == "__main__":
    # 初始化并运行工作流
    blogpost_url = "https://blog.dailydoseofds.com/p/5-chunking-strategies-for-rag"
    workflow = ContentPlanningWorkflow()
    post_response = workflow.run(
        blog_post_url=blogpost_url, post_type=PostType.TWITTER
    )  # PostType.LINKEDIN 用于 LinkedIn 帖子
    logger.info(post_response.content)

用法

1. 创建虚拟环境

   打开 Terminal 并创建一个 python 虚拟环境。

   Mac

   python3 -m venv .venv
   source .venv/bin/activate
   

2. 安装库

   pip install agno firecrawl-py openai packaging requests python-dotenv
   

3. 运行代理

   python workflow.py
   

Assistant

Responses are generated using AI and may contain mistakes.