传媒行业AI Agent应用：内容创作与用户运营的效率提升
在这样的背景下，AI Agent（智能体）作为一种新兴的人工智能应用形式，正在为传媒行业带来革命性的变化。与传统的AI工具不同，AI Agent不仅能够执行特定任务，还具备自主决策、学习进化和多任务协调的能力。它就像是一位不知疲倦的智能助手，能够在内容创作、用户运营、数据分析等多个维度为传媒机构赋能。内容创作方面，AI Agent可以帮助记者快速收集信息、生成初稿、优化标题，甚至能够根据不同平台的
·
传媒行业AI Agent应用：内容创作与用户运营的效率提升
一、引言
1.1 钩子：传媒行业的效率困境
你是否曾在凌晨三点还在为一篇即将发布的新闻稿绞尽脑汁？是否曾面对堆积如山的用户评论感到无从下手？在当今信息爆炸的时代，传媒行业正面临着前所未有的挑战：内容生产速度跟不上用户需求，用户运营成本居高不下，而传统的工作方式似乎已经触及了效率的天花板。
据统计，一家中型媒体机构每天需要处理超过500条新闻线索，撰写数十篇原创内容，同时要回复数千条用户评论。在这种高压环境下，编辑和运营人员往往疲于奔命，难以兼顾质量与效率。更令人担忧的是，随着用户对个性化内容需求的不断增长，传统的"一刀切"内容生产方式已经越来越难以满足受众的期待。
1.2 定义问题/阐述背景
在这样的背景下，AI Agent（智能体）作为一种新兴的人工智能应用形式，正在为传媒行业带来革命性的变化。与传统的AI工具不同，AI Agent不仅能够执行特定任务，还具备自主决策、学习进化和多任务协调的能力。它就像是一位不知疲倦的智能助手，能够在内容创作、用户运营、数据分析等多个维度为传媒机构赋能。
内容创作方面，AI Agent可以帮助记者快速收集信息、生成初稿、优化标题，甚至能够根据不同平台的特点自动调整内容风格。用户运营方面，AI Agent可以实现24小时不间断的用户互动、个性化推荐、智能客服等功能，大幅提升用户体验和运营效率。
1.3 亮明观点/文章目标
本文将带你深入了解AI Agent在传媒行业的应用，重点聚焦于内容创作与用户运营两大场景。我们将从基础概念讲起，逐步深入到技术实现和实战案例，最后探讨最佳实践和未来趋势。
读完这篇文章，你将：
- 理解AI Agent的核心概念和工作原理
- 掌握如何利用AI Agent提升内容创作效率
- 学会运用AI Agent优化用户运营策略
- 了解传媒行业AI Agent应用的技术架构
- 获取可直接落地的实战代码和最佳实践建议
无论你是传媒行业的从业者，还是对AI应用感兴趣的技术人员，相信这篇文章都能为你带来有价值的 insights。
二、基础知识/背景铺垫
2.1 核心概念定义
2.1.1 什么是AI Agent？
AI Agent（智能体）是一种能够感知环境、做出决策并执行动作的智能系统。它可以被视为一个具有目标导向的自主实体，能够在特定环境中独立或半独立地完成任务。
从技术角度来看，一个完整的AI Agent通常包含以下几个核心组件：
- 感知模块（Perception Module）：负责收集和处理环境信息
- 决策引擎（Decision Engine）：基于感知信息和目标制定行动计划
- 执行模块（Execution Module）：将决策转化为具体行动
- 学习模块（Learning Module）：通过反馈不断优化自身行为
- 记忆系统（Memory System）：存储历史信息和经验知识
与传统的AI工具相比，AI Agent的最大特点在于其自主性和适应性。它不仅仅是被动地执行人类指令，而是能够根据环境变化主动调整策略，甚至在没有明确指令的情况下自主完成任务。
2.1.2 传媒行业的AI Agent应用场景
在传媒行业，AI Agent的应用场景非常广泛，主要包括：
-
内容创作领域：
- 新闻线索收集与筛选
- 内容初稿生成
- 内容优化与风格调整
- 多模态内容生成（图文、视频等）
- 事实核查与内容审核
-
用户运营领域：
- 智能客服与用户互动
- 个性化内容推荐
- 用户行为分析与画像
- 社群管理与舆情监控
- 精准营销与活动策划
-
内容分发领域：
- 多渠道自动发布
- 发布时间优化
- A/B测试自动化
- 传播效果追踪与分析
-
管理决策领域：
- 内容策略规划
- 资源分配优化
- 竞争对手分析
- 趋势预测与预警
2.2 AI Agent与传统AI工具的区别
为了更好地理解AI Agent的价值，我们可以通过下表对比AI Agent与传统AI工具的主要区别：
| 对比维度 | 传统AI工具 | AI Agent |
|---|---|---|
| 自主性 | 被动执行指令 | 主动感知环境，自主决策 |
| 适应性 | 固定逻辑，难以适应变化 | 能够学习和进化，灵活适应新情况 |
| 任务范围 | 单一任务导向 | 多任务协调，可处理复杂场景 |
| 交互方式 | 单向输入输出 | 双向交互，可进行对话式协作 |
| 学习能力 | 需人工重新训练 | 持续在线学习，自动优化 |
| 目标导向 | 完成特定功能 | 追求长期目标，可动态调整优先级 |
下面，我们通过一个简单的ER图来展示AI Agent的核心组件及其关系：
2.3 传媒行业AI Agent的技术基础
要构建一个有效的传媒行业AI Agent，需要整合多种前沿技术：
2.3.1 大语言模型（LLM）
大语言模型是AI Agent的"大脑"，负责理解和生成自然语言。在传媒行业，LLM可以用于内容生成、摘要提取、情感分析等任务。目前主流的LLM包括GPT系列、Claude、文心一言等。
2.3.2 多模态AI
现代传媒内容通常包含文字、图片、视频等多种形式，因此AI Agent需要具备多模态理解和生成能力。这涉及计算机视觉、语音识别、图像生成等技术。
2.3.3 知识图谱
知识图谱可以帮助AI Agent更好地理解和组织信息，特别是在新闻报道、事实核查等场景中。通过知识图谱，AI Agent可以建立实体之间的关联，提供更有深度的内容。
2.3.4 强化学习
强化学习使AI Agent能够通过试错不断优化自身行为。在用户运营场景中，强化学习可以帮助AI Agent学习最佳的互动策略，提升用户满意度。
2.3.5 RAG（检索增强生成）
RAG技术可以让AI Agent在生成内容时参考外部知识库，提高内容的准确性和时效性。这对于新闻报道等需要准确信息的场景尤为重要。
2.4 传媒行业AI Agent的发展历程
传媒行业的AI应用经历了从简单工具到智能体的演变过程：
| 阶段 | 时间 | 主要特征 | 代表性应用 |
|---|---|---|---|
| 自动化工具阶段 | 2010-2015 | 规则驱动，执行特定任务 | 自动排版工具、基础语法检查 |
| 基础AI应用阶段 | 2015-2020 | 机器学习，单任务优化 | 推荐算法、简单新闻生成 |
| 多模态AI阶段 | 2020-2022 | 多模态融合，内容理解增强 | 图文生成、视频摘要 |
| AI Agent阶段 | 2022至今 | 自主决策，多任务协调 | 智能采编助手、全渠道运营Agent |
三、核心内容/实战演练
3.1 内容创作AI Agent的设计与实现
3.1.1 需求分析与系统设计
在传媒行业，内容创作是一个复杂的过程，通常包括以下环节：
- 选题策划
- 资料收集与调研
- 内容撰写
- 编辑审核
- 多渠道适配发布
我们的内容创作AI Agent将覆盖这些环节，帮助编辑人员提高工作效率。
系统架构设计：
3.1.2 核心功能实现
让我们通过Python代码来实现内容创作AI Agent的几个核心功能。首先，我们需要设置基本的环境和依赖：
import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import json
import requests
from openai import OpenAI
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
# 初始化OpenAI客户端（实际项目中应使用环境变量管理密钥）
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# 模拟数据库
class MockDatabase:
def __init__(self):
self.contents = []
self.user_behaviors = []
self.knowledge_base = []
self.materials = []
def add_content(self, content: Dict[str, Any]):
self.contents.append(content)
def get_recent_contents(self, days: int = 7) -> List[Dict[str, Any]]:
cutoff = datetime.now() - timedelta(days=days)
return [c for c in self.contents if c.get('publish_time', datetime.now()) > cutoff]
def add_user_behavior(self, behavior: Dict[str, Any]):
self.user_behaviors.append(behavior)
def get_trending_topics(self) -> List[str]:
# 简化的热门话题提取逻辑
all_texts = " ".join([b.get('content', '') for b in self.user_behaviors[-100:]])
# 在实际项目中，这里应该使用更复杂的话题提取算法
return ["AI技术", "可持续发展", "数字经济", "健康生活"]
# 初始化数据库
db = MockDatabase()
现在，让我们实现选题策划Agent：
@dataclass
class TopicSuggestion:
title: str
description: str
relevance_score: float
trend_score: float
difficulty_score: float
estimated_engagement: float
class TopicPlanningAgent:
def __init__(self, db: MockDatabase):
self.db = db
self.vectorizer = TfidfVectorizer(max_features=1000)
def analyze_current_trends(self) -> List[str]:
"""分析当前热点趋势"""
return self.db.get_trending_topics()
def evaluate_content_performance(self) -> Dict[str, float]:
"""评估历史内容表现"""
recent_contents = self.db.get_recent_contents()
if not recent_contents:
return {}
# 简化的内容表现评估
performance = {}
for content in recent_contents:
category = content.get('category', 'general')
engagement = content.get('engagement', 0)
if category not in performance:
performance[category] = []
performance[category].append(engagement)
# 计算各类别的平均表现
return {cat: np.mean(scores) for cat, scores in performance.items()}
def generate_topic_suggestions(self,
focus_areas: Optional[List[str]] = None,
count: int = 5) -> List[TopicSuggestion]:
"""生成选题建议"""
trends = self.analyze_current_trends()
performance = self.evaluate_content_performance()
# 构建提示词
system_prompt = """
你是一位专业的媒体选题策划专家。请根据提供的热点趋势和历史内容表现数据，
为媒体机构生成有吸引力的选题建议。每个选题应该包含：
1. 一个吸引人的标题
2. 简短的选题描述
3. 相关性评分（0-1）
4. 趋势评分（0-1）
5. 制作难度评分（0-1，越高越难）
6. 预计互动量评分（0-1）
请以JSON格式返回结果。
"""
user_prompt = f"""
热点趋势：{trends}
历史内容表现：{performance}
关注领域：{focus_areas if focus_areas else '无特定限制'}
需要生成的选题数量：{count}
"""
# 调用LLM生成选题建议
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"}
)
# 解析结果
try:
result = json.loads(response.choices[0].message.content)
suggestions = result.get('suggestions', [])
# 转换为TopicSuggestion对象
topic_suggestions = []
for s in suggestions:
topic_suggestions.append(TopicSuggestion(
title=s.get('title', ''),
description=s.get('description', ''),
relevance_score=s.get('relevance_score', 0.5),
trend_score=s.get('trend_score', 0.5),
difficulty_score=s.get('difficulty_score', 0.5),
estimated_engagement=s.get('estimated_engagement', 0.5)
))
return topic_suggestions
except Exception as e:
print(f"解析选题建议时出错：{e}")
return []
接下来，让我们实现内容生成Agent：
@dataclass
class ContentData:
title: str
content: str
summary: str
tags: List[str]
category: str
word_count: int
read_time: int
class ContentGenerationAgent:
def __init__(self, db: MockDatabase):
self.db = db
def research_topic(self, topic: str, depth: int = 3) -> List[Dict[str, Any]]:
"""调研选题，收集相关资料"""
# 实际项目中，这里应该调用搜索引擎、API等获取真实资料
# 这里我们使用模拟数据
mock_research_data = [
{
"source": "权威新闻网站",
"title": f"{topic}的最新发展",
"content": f"关于{topic}的详细报道内容...",
"relevance": 0.9,
"credibility": 0.95
},
{
"source": "行业报告",
"title": f"{topic}市场分析",
"content": f"{topic}的市场规模、增长趋势等数据...",
"relevance": 0.85,
"credibility": 0.9
},
{
"source": "专家观点",
"title": f"专家谈{topic}",
"content": f"行业专家对{topic}的看法和预测...",
"relevance": 0.8,
"credibility": 0.85
}
]
return mock_research_data[:depth]
def generate_content(self,
topic: str,
research_data: List[Dict[str, Any]],
style: str = "news",
tone: str = "neutral",
target_length: int = 1000) -> ContentData:
"""生成内容"""
# 构建研究资料摘要
research_summary = "\n".join([
f"来源：{item['source']}\n标题：{item['title']}\n内容：{item['content'][:200]}..."
for item in research_data
])
# 风格和语气映射
style_descriptions = {
"news": "客观、准确的新闻报道风格",
"blog": "轻松、亲切的博客风格",
"opinion": "观点鲜明的评论风格",
"feature": "深入、细致的特写风格"
}
tone_descriptions = {
"neutral": "中立客观",
"positive": "积极正面",
"critical": "批评质疑",
"humorous": "幽默风趣"
}
# 构建提示词
system_prompt = f"""
你是一位专业的媒体内容创作者。请根据提供的研究资料，创作一篇{style_descriptions.get(style, '通用风格')}的文章。
文章应该保持{tone_descriptions.get(tone, '中立客观')}的语气，大约{target_length}字左右。
请确保：
1. 内容准确，基于提供的研究资料
2. 结构清晰，有引人入胜的标题和开头
3. 逻辑连贯，段落过渡自然
4. 包含有价值的见解和信息
请以JSON格式返回结果，包含以下字段：
- title: 文章标题
- content: 完整文章内容
- summary: 100字左右的摘要
- tags: 3-5个标签
- category: 文章分类
"""
user_prompt = f"""
主题：{topic}
研究资料：
{research_summary}
"""
# 调用LLM生成内容
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"}
)
# 解析结果
try:
result = json.loads(response.choices[0].message.content)
# 计算字数和阅读时间
content = result.get('content', '')
word_count = len(content)
read_time = max(1, word_count // 300) # 假设每分钟阅读300字
return ContentData(
title=result.get('title', ''),
content=content,
summary=result.get('summary', ''),
tags=result.get('tags', []),
category=result.get('category', 'general'),
word_count=word_count,
read_time=read_time
)
except Exception as e:
print(f"生成内容时出错：{e}")
return ContentData(
title="",
content="",
summary="",
tags=[],
category="",
word_count=0,
read_time=0
)
def optimize_for_platform(self,
content: ContentData,
platform: str = "website") -> ContentData:
"""针对不同平台优化内容"""
platform_requirements = {
"website": {
"title_length": (10, 60),
"content_length": (800, 2000),
"format": "html"
},
"wechat": {
"title_length": (15, 30),
"content_length": (500, 1500),
"format": "rich_text"
},
"weibo": {
"title_length": (0, 20),
"content_length": (100, 500),
"format": "short_text"
},
"linkedin": {
"title_length": (10, 80),
"content_length": (300, 1000),
"format": "professional"
}
}
requirements = platform_requirements.get(platform, platform_requirements["website"])
# 这里是简化的平台优化逻辑，实际项目中应该更复杂
system_prompt = f"""
请将提供的内容优化为适合{platform}平台发布的版本。
标题长度应该在{requirements['title_length'][0]}-{requirements['title_length'][1]}字之间。
内容长度应该在{requirements['content_length'][0]}-{requirements['content_length'][1]}字之间。
格式应该符合{requirements['format']}特点。
"""
# 调用LLM进行平台优化
# 这里省略具体实现，类似于前面的生成逻辑
return content # 简化返回，实际应该返回优化后的内容
最后，让我们创建一个整合的内容创作工作流：
class ContentCreationWorkflow:
def __init__(self, db: MockDatabase):
self.topic_agent = TopicPlanningAgent(db)
self.content_agent = ContentGenerationAgent(db)
self.db = db
def execute(self,
focus_areas: Optional[List[str]] = None,
style: str = "news",
tone: str = "neutral",
target_platforms: Optional[List[str]] = None) -> Dict[str, Any]:
"""执行完整的内容创作工作流"""
print("步骤1: 生成选题建议...")
topic_suggestions = self.topic_agent.generate_topic_suggestions(
focus_areas=focus_areas,
count=3
)
if not topic_suggestions:
return {"success": False, "error": "无法生成选题建议"}
# 选择综合评分最高的选题
best_topic = max(topic_suggestions, key=lambda t: (
t.relevance_score * 0.3 +
t.trend_score * 0.3 +
t.estimated_engagement * 0.4
))
print(f"步骤2: 选择选题 - {best_topic.title}")
print("步骤3: 调研选题...")
research_data = self.content_agent.research_topic(best_topic.title, depth=3)
print("步骤4: 生成内容...")
content = self.content_agent.generate_content(
topic=best_topic.title,
research_data=research_data,
style=style,
tone=tone
)
if not content.title:
return {"success": False, "error": "无法生成内容"}
print(f"步骤5: 内容生成完成 - {content.title}")
# 针对不同平台优化内容
platform_contents = {}
if target_platforms:
for platform in target_platforms:
print(f"步骤6: 优化内容用于{platform}平台...")
platform_content = self.content_agent.optimize_for_platform(content, platform)
platform_contents[platform] = platform_content
# 保存内容到数据库
content_record = {
"title": content.title,
"content": content.content,
"summary": content.summary,
"tags": content.tags,
"category": content.category,
"word_count": content.word_count,
"read_time": content.read_time,
"topic_info": {
"title": best_topic.title,
"description": best_topic.description,
"relevance_score": best_topic.relevance_score,
"trend_score": best_topic.trend_score
},
"platform_contents": platform_contents,
"creation_time": datetime.now(),
"status": "draft"
}
self.db.add_content(content_record)
return {
"success": True,
"topic": best_topic,
"content": content,
"platform_contents": platform_contents,
"content_record": content_record
}
3.2 用户运营AI Agent的设计与实现
3.2.1 需求分析与系统设计
用户运营是传媒行业的另一个核心环节，主要目标是提升用户活跃度、留存率和忠诚度。一个有效的用户运营AI Agent应该能够：
- 理解用户需求和偏好
- 提供个性化内容推荐
- 进行智能互动和客服
- 管理用户社群
- 分析用户行为并提供洞察
系统架构设计：
3.2.2 核心功能实现
让我们继续用Python代码实现用户运营AI Agent的核心功能。首先，我们需要定义用户画像的数据结构：
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json
import numpy as np
from collections import defaultdict
# 用户活跃度枚举
class UserActivityLevel(Enum):
NEW = "new"
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
VIP = "vip"
# 用户兴趣标签
@dataclass
class InterestTag:
name: str
weight: float
last_updated: datetime
# 用户行为记录
@dataclass
class UserBehavior:
behavior_type: str # view, like, share, comment, etc.
content_id: str
content_category: str
timestamp: datetime
duration: Optional[int] = None # 浏览时长（秒）
details: Dict[str, Any] = field(default_factory=dict)
# 用户画像
@dataclass
class UserProfile:
user_id: str
created_at: datetime
last_active: datetime
activity_level: UserActivityLevel
interests: List[InterestTag] = field(default_factory=list)
behaviors: List[UserBehavior] = field(default_factory=list)
preferences: Dict[str, Any] = field(default_factory=dict)
engagement_score: float = 0.0
lifetime_value: float = 0.0
现在，让我们实现用户画像Agent：
class UserProfilingAgent:
def __init__(self, db: MockDatabase):
self.db = db
self.user_profiles: Dict[str, UserProfile] = {}
def create_user_profile(self, user_id: str) -> UserProfile:
"""创建新用户画像"""
now = datetime.now()
profile = UserProfile(
user_id=user_id,
created_at=now,
last_active=now,
activity_level=UserActivityLevel.NEW
)
self.user_profiles[user_id] = profile
return profile
def get_user_profile(self, user_id: str) -> Optional[UserProfile]:
"""获取用户画像"""
if user_id not in self.user_profiles:
return None
return self.user_profiles[user_id]
def update_user_activity(self, user_id: str, behavior: UserBehavior) -> None:
"""更新用户活跃度"""
profile = self.get_user_profile(user_id)
if not profile:
profile = self.create_user_profile(user_id)
# 添加行为记录
profile.behaviors.append(behavior)
profile.last_active = behavior.timestamp
# 计算活跃度
self._calculate_activity_level(profile)
# 更新兴趣标签
self._update_interests(profile, behavior)
# 计算参与度分数
self._calculate_engagement_score(profile)
def _calculate_activity_level(self, profile: UserProfile) -> None:
"""计算用户活跃度等级"""
now = datetime.now()
thirty_days_ago = now - timedelta(days=30)
seven_days_ago = now - timedelta(days=7)
# 统计不同时间范围内的行为数量
total_behaviors_30d = len([b for b in profile.behaviors if b.timestamp > thirty_days_ago])
total_behaviors_7d = len([b for b in profile.behaviors if b.timestamp > seven_days_ago])
# 简化的活跃度计算逻辑
if total_behaviors_30d == 0:
profile.activity_level = UserActivityLevel.NEW
elif total_behaviors_7d < 2:
profile.activity_level = UserActivityLevel.LOW
elif total_behaviors_7d < 7:
profile.activity_level = UserActivityLevel.MEDIUM
elif total_behaviors_30d > 30:
profile.activity_level = UserActivityLevel.VIP
else:
profile.activity_level = UserActivityLevel.HIGH
def _update_interests(self, profile: UserProfile, behavior: UserBehavior) -> None:
"""更新用户兴趣标签"""
# 查找或创建兴趣标签
interest_found = False
for interest in profile.interests:
if interest.name == behavior.content_category:
# 更新现有兴趣标签
interest.weight = min(1.0, interest.weight + 0.1)
interest.last_updated = behavior.timestamp
interest_found = True
break
if not interest_found:
# 创建新的兴趣标签
profile.interests.append(InterestTag(
name=behavior.content_category,
weight=0.3,
last_updated=behavior.timestamp
))
# 衰减其他兴趣标签
for interest in profile.interests:
if interest.name != behavior.content_category:
interest.weight = max(0.0, interest.weight - 0.01)
# 按权重排序兴趣标签
profile.interests.sort(key=lambda x: x.weight, reverse=True)
# 保留前10个兴趣标签
profile.interests = profile.interests[:10]
def _calculate_engagement_score(self, profile: UserProfile) -> None:
"""计算用户参与度分数"""
now = datetime.now()
thirty_days_ago = now - timedelta(days=30)
# 获取最近30天的行为
recent_behaviors = [b for b in profile.behaviors if b.timestamp > thirty_days_ago]
if not recent_behaviors:
profile.engagement_score = 0.0
return
# 计算不同类型行为的权重
behavior_weights = {
"view": 1,
"like": 3,
"comment": 5,
"share": 7,
"create": 10
}
# 计算加权行为分数
total_weighted_behaviors = sum(
behavior_weights.get(b.behavior_type, 1)
for b in recent_behaviors
)
# 计算活跃度衰减因子（最近活跃的用户得分更高）
days_since_last_active = (now - profile.last_active).days
recency_factor = max(0.1, 1.0 - (days_since_last_active / 30))
# 计算最终参与度分数（0-100）
profile.engagement_score = min(100.0, total_weighted_behaviors * recency_factor)
def get_user_segment(self, user_id: str) -> Dict[str, Any]:
"""获取用户细分信息"""
profile = self.get_user_profile(user_id)
if not profile:
return {}
# 简化的用户细分逻辑
top_interests = [i.name for i in profile.interests[:3]] if profile.interests else []
return {
"user_id": user_id,
"activity_level": profile.activity_level.value,
"top_interests": top_interests,
"engagement_score": profile.engagement_score,
"days_since_last_active": (datetime.now() - profile.last_active).days,
"total_behaviors": len(profile.behaviors)
}
接下来，让我们实现个性化推荐Agent：
@dataclass
class RecommendedContent:
content_id: str
title: str
summary: str
category: str
relevance_score: float
novelty_score: float
diversity_score: float
final_score: float
class PersonalizationAgent:
def __init__(self, db: MockDatabase, profiling_agent: UserProfilingAgent):
self.db = db
self.profiling_agent = profiling_agent
def calculate_content_relevance(self,
user_profile: UserProfile,
content: Dict[str, Any]) -> float:
"""计算内容与用户的相关性"""
if not user_profile.interests:
return 0.5 # 中性评分
content_category = content.get('category', 'general')
# 查找用户对该类别的兴趣权重
category_interest = 0.0
for interest in user_profile.interests:
if interest.name == content_category:
category_interest = interest.weight
break
# 内容标签匹配
content_tags = content.get('tags', [])
tag_match_score = 0.0
for interest in user_profile.interests:
if interest.name in content_tags:
tag_match_score += interest.weight
# 综合相关性评分
relevance = 0.7 * category_interest + 0.3 * min(1.0, tag_match_score)
return relevance
def calculate_content_novelty(self,
user_profile: UserProfile,
content: Dict[str, Any]) -> float:
"""计算内容的新颖性（对用户而言）"""
content_id = content.get('content_id', '')
# 检查用户是否已经看过该内容
viewed_content_ids = set(b.content_id for b in user_profile.behaviors if b.behavior_type == 'view')
if content_id in viewed_content_ids:
return 0.0 # 已看过的内容新颖性为0
# 检查内容发布时间
publish_time = content.get('publish_time', datetime.now() - timedelta(days=30))
days_since_publish = (datetime.now() - publish_time).days
# 越新的内容新颖性越高
if days_since_publish <= 1:
time_factor = 1.0
elif days_since_publish <= 7:
time_factor = 0.8
elif days_since_publish <= 30:
time_factor = 0.5
else:
time_factor = 0.2
# 检查内容类别新颖性
content_category = content.get('category', 'general')
user_categories = set(i.name for i in user_profile.interests[:5]) # 用户主要兴趣类别
if content_category not in user_categories:
category_novelty = 0.8 # 非主要兴趣类别的内容有一定新颖性
else:
category_novelty = 0.3 # 主要兴趣类别的内容新颖性较低
# 综合新颖性评分
novelty = 0.6 * time_factor + 0.4 * category_novelty
return novelty
def calculate_diversity(self,
candidate_contents: List[Dict[str, Any]],
current_recommendations: List[RecommendedContent]) -> float:
"""计算添加候选内容后的多样性"""
if not current_recommendations:
return 1.0 # 如果还没有推荐，多样性为1
# 获取已推荐内容的类别
recommended_categories = set(rc.category for rc in current_recommendations)
# 检查候选内容的类别
candidate_category = candidate_contents[0].get('category', 'general') if candidate_contents else 'general'
# 如果候选内容类别不在已推荐类别中，多样性较高
if candidate_category not in recommended_categories:
return 0.8
else:
return 0.3 # 同类内容多样性较低
def generate_recommendations(self,
user_id: str,
count: int = 10,
diversity_weight: float = 0.3,
novelty_weight: float = 0.2) -> List[RecommendedContent]:
"""生成个性化推荐内容"""
user_profile = self.profiling_agent.get_user_profile(user_id)
if not user_profile:
return []
# 获取所有可用内容（实际项目中应该有更高效的检索方式）
available_contents = self.db.contents
if not available_contents:
return []
# 计算每个内容的相关性和新颖性
scored_contents = []
for content in available_contents:
relevance = self.calculate_content_relevance(user_profile, content)
novelty = self.calculate_content_novelty(user_profile, content)
scored_contents.append({
"content": content,
"relevance": relevance,
"novelty": novelty
})
# 初步筛选：按相关性排序，取前20个
scored_contents.sort(key=lambda x: x["relevance"], reverse=True)
candidates = scored_contents[:20]
# 使用贪婪算法进行多样性优化
recommendations = []
remaining_candidates = candidates.copy()
while len(recommendations) < count and remaining_candidates:
best_candidate = None
best_score = -1
# 为每个候选内容计算综合分数
for i, candidate in enumerate(remaining_candidates):
# 计算添加此候选后的多样性
temp_recommendations = recommendations.copy()
temp_recommendations.append(
RecommendedContent(
content_id=candidate["content"].get("content_id", ""),
title=candidate["content"].get("title", ""),
summary=candidate["content"].get("summary", ""),
category=candidate["content"].get("category", "general"),
relevance_score=candidate["relevance"],
novelty_score=candidate["novelty"],
diversity_score=0, # 临时值
final_score=0 # 临时值
)
)
diversity = self.calculate_diversity([candidate["content"]], temp_recommendations)
# 计算综合分数
relevance_weight = 1.0 - diversity_weight - novelty_weight
final_score = (
relevance_weight * candidate["relevance"] +
diversity_weight * diversity +
novelty_weight * candidate["novelty"]
)
if final_score > best_score:
best_score = final_score
best_candidate = {
"index": i,
"candidate": candidate,
"diversity": diversity,
"final_score": final_score
}
if best_candidate:
# 添加最佳候选到推荐列表
recommendations.append(
RecommendedContent(
content_id=best_candidate["candidate"]["content"].get("content_id", ""),
title=best_candidate["candidate"]["content"].get("title", ""),
summary=best_candidate["candidate"]["content"].get("summary", ""),
category=best_candidate["candidate"]["content"].get("category", "general"),
relevance_score=best_candidate["candidate"]["relevance"],
novelty_score=best_candidate["candidate"]["novelty"],
diversity_score=best_candidate["diversity"],
final_score=best_candidate["final_score"]
)
)
# 从候选列表中移除已选择的内容
remaining_candidates.pop(best_candidate["index"])
# 按最终分数排序推荐内容
recommendations.sort(key=lambda x: x.final_score, reverse=True)
return recommendations
最后，让我们实现智能互动Agent：
from typing import List, Dict, Any, Optional, Tuple
from enum import Enum
import random
import re
# 对话意图枚举
class ConversationIntent(Enum):
GREETING = "greeting"
CONTENT_RECOMMENDATION = "content_recommendation"
QUESTION = "question"
COMPLAINT = "complaint"
FEEDBACK = "feedback"
CHITCHAT = "chitchat"
FAREWELL = "farewell"
UNKNOWN = "unknown"
# 对话状态
@dataclass
class ConversationState:
user_id: str
conversation_id: str
history: List[Dict[str, str]] = field(default_factory=list)
current_intent: Optional[ConversationIntent] = None
entities: Dict[str, Any] = field(default_factory=dict)
waiting_for: Optional[str] = None
started_at: datetime = field(default_factory=datetime.now)
last_updated: datetime = field(default_factory=datetime.now)
class SmartInteractionAgent:
def __init__(self,
db: MockDatabase,
profiling_agent: UserProfilingAgent,
personalization_agent: PersonalizationAgent):
self.db = db
self.profiling_agent = profiling_agent
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
10
0- 0
已为社区贡献295条内容
所有评论(0)