舆情分析是企业感知市场脉搏、预警公关危机的“听诊器”，然而传统的舆情分析系统更像是一个个“手工作坊”，面临数据收集效率低、分析深度不够、实时性差等问题，经常反馈之后，等企业拿到报告时，舆论热点早已转移，错过最佳时间。这些挑战，正是所有舆情系统开发者共同的痛点。

本方案将基于真实的代码实现，向您介绍如何使用函数计算 AgentRun 平台，构建一个现代化的“舆情分析专家”，该系统不仅实现了从数据采集到报告生成的可视化、全流程自动化，更通过流式架构，让洞察实时呈现。

系统架构设计

整个舆情分析系统采用分层架构设计，核心思想是通过代码严格控制流程执行顺序，而非依赖 LLM 的自主决策。

快速体验和效果预览

在深入技术细节前，我们先直观感受一下这套系统的效果。通过 AgentRun 平台，只需简单几步即可完成部署。

快速部署

打开阿里云函数计算 AgentRun 探索页面：

可以找到舆情分析专家案例，并点击卡片右下角进行部署，填写完整对应的参数信息即可点击右下角确定创建按钮：

此处需要稍等片刻，创建完之后可以看到体验地址，也可以跳转到运行时与沙箱看到部署完的Agent：

首页地址即右侧main_web地址，直接查看线上效果

也可以查看该应用案例代码，并进行在线二次开发：

效果体验

打开体验地址，可以看到舆情分析专家页面，此时可以输入一个词进行舆情分析：

分析过程中，系统会调用 函数计算 AgentRun 的 Sandbox 沙箱（确切说是创建的时候，选择的浏览器沙箱），可以看到 AI 控制云上的浏览器进行数据检索：

完成之后，系统会整理所有采集到的数据和信息：

最终生成文字+图表的可视化报告

AgentRun 相比传统方案的核心优势

安全隔离的执行环境

传统舆情系统通常直接在服务器上运行爬虫程序，面临着安全风险和环境污染问题。当某个网站的反爬机制触发时，可能影响整个服务器的稳定性。而 AgentRun Sandbox 提供了完全隔离的浏览器环境，即使单个采集任务出现问题，也不会影响系统的整体运行。

async def create_browser_sandbox() -> Optional[BrowserSandbox]: """创建隔离的浏览器环境，避免环境污染""" try: sandbox = await Sandbox.create_async( template_type=TemplateType.BROWSER, template_name=agentrun_browser_sandbox_name, ) _sandboxes[sandbox.sandbox_id] = sandbox return sandbox except Exception as e: # 单个Sandbox失败不影响其他实例 raise SandboxCreationError(f"创建 Sandbox 失败: {e}")

真实浏览器环境模拟

传统爬虫方案通常使用简单的HTTP请求库，容易被现代网站的反爬机制识别和拦截。AgentRun Sandbox 提供的是真实的 Chrome 浏览器环境，能够完整执行JavaScript、处理复杂的页面交互，大大提高了数据采集的成功率。从代码中可以看到，系统通过 Playwright 连接到真实的 Chrome 实例：

async with async_playwright() as playwright: browser = await playwright.chromium.connect_over_cdp(sandbox.get_cdp_url()) context = browser.contexts[0] if browser.contexts else await browser.new_context() page = context.pages[0] if context.pages else await context.new_page()

可视化调试能力

函数计算 AgentRun 最独特的优势是提供了实时的 VNC 预览功能，开发者和用户可以实时观察浏览器的操作过程。这种透明性在传统方案中是无法实现的，它不仅有助于调试和优化采集逻辑，还能让用户直观地了解系统的工作状态。

弹性扩展和故障恢复

传统系统在面临大规模采集任务时，往往需要复杂的分布式架构设计。而 函数计算 AgentRun 天然支持多 Sandbox 并行处理，系统可以根据需要动态创建和销毁浏览器实例。更重要的是，当某个实例出现故障时，系统能够自动检测并重建：

async def recreate_sandbox_if_closed(sandbox_id: str, error_message: str): """智能故障检测和自动重建机制""" closed_error_patterns = [ "Target page, context or browser has been closed", "Browser has been closed", "Connection closed", ] is_closed_error = any(pattern.lower() in error_message.lower() for pattern in closed_error_patterns) if is_closed_error: await remove_sandbox(sandbox_id) new_sandbox = await create_browser_sandbox() return new_sandbox

AgentRun Sandbox 采用阿里云函数计算实现，支持百万沙箱模板（函数级别）并发运行，Serverless弹性伸缩，支持3.5w+沙箱/分钟，支持缩容到0，按请求感知调度。

后端核心实现

Agent 工具链设计

系统的核心是一个基于 PydanticAI 的智能体，该智能体包含四个关键工具，每个工具负责舆情分析的不同阶段。Agent 的设计遵循严格的执行顺序，确保数据收集的完整性和分析的准确性。

opinion_agent = Agent( agentrun_model, deps_type=StateDeps, system_prompt="""你是舆情分析系统的执行者。 你的任务是按照以下严格流程执行舆情分析： 【流程】 1. 收到关键词后，调用 collect_data 工具收集数据 2. 数据收集完成后，调用 analyze_data 工具分析数据 3. 分析完成后，调用 write_report 工具撰写报告 4. 报告完成后，调用 render_html 工具生成 HTML 【重要规则】 - 必须按顺序调用工具 - 每个工具只调用一次 - 不要跳过任何步骤 - 不要编造数据 """, retries=3, )

流式输出与实时反馈

传统舆情系统通常采用批处理模式，用户需要等待很长时间才能看到结果。而基于 函数计算 AgentRun 的系统实现了真正的流式输出，用户可以实时观察每个处理步骤的进展。这种实时性不仅提升了用户体验，也便于及时发现和解决问题。

async def push_state_event(run_id: str, state: OpinionState): """实时推送状态更新，用户无需等待""" event = StateSnapshotEvent( type=EventType.STATE_SNAPSHOT, snapshot=state.model_dump(), timestamp=int(time.time() * 1000) ) await event_manager.push_event(run_id, event)

智能数据质量控制

系统实现了严格的数据质量控制机制，通过多维度评估确保收集到的数据具有较高的相关性和价值。这种质量控制在传统系统中往往是缺失的，导致大量噪音数据影响分析结果。

async def evaluate_relevance(keyword: str, title: str, snippet: str) -> float: """多维度相关性评估，确保数据质量""" text = f"{title} {snippet}" text_lower = text.lower() # 检测关键词匹配度 has_chinese_keyword = any('\u4e00' <= char <= '鿿' for char in keyword) result_has_chinese = any('\u4e00' <= char <= '鿿' for char in text) # 中文关键词必须在结果中有中文内容 if has_chinese_keyword and not result_has_chinese: return 0.0 # 排除明显的无关网站 irrelevant_patterns = [ "calculator", "deepseek", "chegg", "stackoverflow", "翻译", "dictionary", "词典" ] if any(pattern in text_lower for pattern in irrelevant_patterns): return 0.0 # 计算相关性得分 score = 0.0 if keyword in text: score += 0.6 # 基础分 # 时效性加分 time_keywords = ["最新", "今日", "近日", "2024", "2025"] if any(tk in text for tk in time_keywords): score += 0.1 return max(0.0, min(1.0, score))

深度内容抓取技术

平台适配策略

不同的社交媒体平台具有不同的页面结构和内容组织方式，传统系统往往采用统一的抓取策略，导致数据质量参差不齐。AgentRun 系统针对不同平台实现了定制化的抓取逻辑：

async def explore_page_with_llm(page, keyword: str, url: str, source: str, initial_content: str): """基于平台特性的智能内容抓取""" if "weibo.com" in url: # 微博特定的评论和转发抓取 available_actions = [ {"action": "view_comments", "selector": ".WB_feed_expand, [class*='comment']"}, {"action": "view_retweets", "selector": ".WB_feed_expand, [class*='repost']"}, ] elif "zhihu.com" in url: # 知乎回答和评论抓取 available_actions = [ {"action": "view_more_answers", "selector": ".AnswerItem, .List-item"}, {"action": "view_comments", "selector": ".Comments-container, .CommentItem"}, ] elif "bilibili.com" in url: # B站视频评论抓取 available_actions = [ {"action": "view_comments", "selector": ".reply-item, .root-reply"}, {"action": "view_related", "selector": ".video-page-card, .recommend-list"}, ]

LLM 驱动的智能探索

系统创新性地引入了 LLM 驱动的智能探索机制，让 AI 决定是否需要深入抓取某个页面的额外内容，如评论区、相关推荐等。这种智能决策大大提高了数据采集的效率和针对性。

async def llm_decide_exploration(keyword: str, page_url: str, page_content: str, source: str): """LLM 智能决策是否进行深度探索""" prompt = f"""请根据以下信息决定是否需要进一步探索页面获取更多舆情数据。 【搜索关键词】{keyword} 【当前页面】{page_url} 【已获取内容预览】{page_content[:500]} 【决策标准】 1. 如果当前内容已经足够丰富，可能不需要进一步探索 2. 如果是微博/B站等平台，评论区通常包含重要的舆情信息 3. 权衡时间成本，每个页面最多探索1-2个操作 请返回 JSON 格式的决策结果。 """ result = await explorer.run(prompt) return json.loads(result.output)

前端 VNC 集成实现

动态库加载机制

前端 VNC 客户端需要动态加载 noVNC 库，系统实现了智能的加载机制，支持本地资源和 CDN 回退：

function loadScript(url) { return new Promise(function(resolve, reject) { var script = document.createElement('script'); script.src = baseUrl + url; script.onload = resolve; script.onerror = function() { // 本地加载失败，尝试 CDN var fallbackUrl = url.includes('wordcloud') ? 'https://cdn.jsdelivr.net/npm/echarts-wordcloud@2.1.0/dist/echarts-wordcloud.min.js' : 'https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js'; var fallbackScript = document.createElement('script'); fallbackScript.src = fallbackUrl; fallbackScript.onload = resolve; fallbackScript.onerror = reject; document.head.appendChild(fallbackScript); }; document.head.appendChild(script); }); }

多协议适配

考虑到部署环境的复杂性，VNC 组件实现了 HTTP/HTTPS 环境下的 WebSocket 协议自适应：

const adjustWebSocketUrl = useCallback((url: string): string => { const isHttps = window.location.protocol === 'https:'; if (!isHttps && url.startsWith('wss://')) { return url.replace('wss://', 'ws://'); } if (isHttps && url.startsWith('ws://')) { return url.replace('ws://', 'wss://'); } return url; }, []);

智能分析与报告生成

标准化情感分析

系统实现了基于关键词词典的情感分析算法，相比传统的机器学习模型，这种方法更加透明和可控：

class SentimentStandards: """情感倾向标准化计算""" POSITIVE_KEYWORDS = [ "优秀", "卓越", "创新", "领先", "突破", "成功", "赞", "好评", "支持", "认可", "满意", "信赖", "期待", "看好", "值得", "推荐", "喜欢" ] NEGATIVE_KEYWORDS = [ "差", "糟糕", "失败", "落后", "问题", "缺陷", "批评", "质疑", "担忧", "失望", "不满", "抱怨", "投诉", "差评", "垃圾", "骗局" ] @staticmethod def calculate_sentiment_score(text: str) -> float: """计算情感得分 (-1.0 到 1.0)""" positive_count = sum(1 for word in SentimentStandards.POSITIVE_KEYWORDS if word in text) negative_count = sum(1 for word in SentimentStandards.NEGATIVE_KEYWORDS if word in text) total_count = positive_count + negative_count if total_count == 0: return 0.0 return (positive_count - negative_count) / total_count

流式报告生成

报告生成过程采用流式输出，用户可以实时观察报告的撰写过程，这种体验是传统系统无法提供的：

async with writer.run_stream(report_prompt) as result: async for text in result.stream_text(): report_content = text state.report_text = report_content current_time = asyncio.get_event_loop().time() content_delta = len(report_content) - last_event_length time_delta = current_time - last_event_time # 每 100 字符或每 0.3 秒发送一次更新 if content_delta >= 100 or time_delta >= 0.3: await push_state_event(run_id, state)

部署与运维优势

简化的部署流程

相比传统舆情系统需要复杂的分布式爬虫集群部署，AgentRun 系统的部署相对简单。只需要配置好环境变量和 AgentRun Sandbox 模板，系统就能自动管理浏览器实例的创建和销毁：

# 核心配置 AGENTRUN_MODEL_NAME=your_model_name MODEL_NAME=qwen3-max AGENTRUN_BROWSER_SANDBOX_NAME=your_browser_template TIMEOUT=180

自动化运维能力

系统内置了完善的监控和自恢复机制，大大降低了运维复杂度。当检测到异常时，系统能够自动重建资源，保证服务的连续性：

# 连接失败时自动重连（每 10 秒尝试一次） useEffect(() => { if (status === 'error' && active && rfbLoaded) { reconnectTimerRef.current = setTimeout(() => { cleanupRfb(); lastUrlRef.current = null; fetchVncUrl(true); }, RECONNECT_INTERVAL); } }, [status, active, rfbLoaded]);

性能与扩展性分析

并发处理能力

传统系统的并发能力往往受限于单机资源，而 函数计算 AgentRun 系统可以根据需要动态创建多个 Sandbox 实例，实现真正的水平扩展。系统通过异步编程模型和连接池管理，能够高效处理大量并发请求：

uvicorn.run( "main:app", host="0.0.0.0", port=8000, log_level="info", timeout_keep_alive=120, limit_concurrency=100, # 支持高并发 )

资源弹性管理

系统实现了智能的资源管理策略，能够根据任务负载动态调整 Sandbox 实例数量。这种弹性扩展能力是传统固定架构难以实现的：

async def get_all_sandboxes() -> List[Dict[str, Any]]: """动态获取所有可用的Sandbox实例""" result = [] async with _sandbox_lock: for sandbox_id, sandbox in _sandboxes.items(): try: # 检查实例健康状态 vnc_url = sandbox.get_vnc_url() result.append({ "sandbox_id": sandbox_id, "vnc_url": vnc_url, "active": True, }) except Exception: # 自动清理失效实例 result.append({ "sandbox_id": sandbox_id, "active": False, }) return result

总结

基于 函数计算 AgentRun 构建的舆情分析系统展现了现代 AI 技术在实际业务场景中的强大应用潜力。相比传统方案，函数计算 AgentRun 系统在安全性、可靠性、可观测性和扩展性方面都具有显著优势。

通过隔离的浏览器环境，系统解决了传统爬虫面临的安全风险和环境污染问题。实时的 VNC 预览功能提供了前所未有的透明度，让开发者和用户能够直观地观察系统工作状态。智能的故障检测和自恢复机制大大降低了运维复杂度，而流式输出设计则显著提升了用户体验。

更重要的是，函数计算 AgentRun 系统将复杂的舆情分析任务完全自动化，从多平台数据采集、深度内容抓取、智能情感分析到专业报告生成，整个流程无需人工干预。这种端到端的自动化能力，结合 AI 技术的持续进步，将为企业和机构的舆情分析工作带来革命性的改变。

随着技术的不断发展，基于 函数计算 AgentRun 的舆情系统将在准确性、智能化程度和处理效率方面持续提升，成为现代舆情分析和危机管理的重要工具。

欢迎加入“函数计算 AgentRun 客户群”与我们交流，钉钉群号：134570017218。

快速了解函数计算 AgentRun：

​一句话介绍：​函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。

函数计算 AgentRun 架构图

AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、Langchain、RAGFlow、Mem0 等主流开源生态。AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60%。

​让​开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施，让 Agentic AI 真正进入企业生产环境。