作者：江昱
舆情分析是企业感知市场脉搏、预警公关危机的“听诊器”，然而传统的舆情分析系统更像是一个个“手工作坊”，面临数据收集效率低、分析深度不够、实时性差等问题，经常反馈之后，等企业拿到报告时，舆论热点早已转移，错过最佳时间。这些挑战，正是所有舆情系统开发者共同的痛点。
本方案将基于真实的代码实现，向您介绍如何使用函数计算 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_nameMODEL_NAME=qwen3-maxAGENTRUN_BROWSER_SANDBOX_NAME=your_browser_templateTIMEOUT=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 客户群”与我们交流，钉钉群号：134570017218。
快速了解函数计算 AgentRun：
一句话介绍： 函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。
函数计算 AgentRun 架构图
AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、Langchain、RAGFlow、Mem0 等主流开源生态。AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60%。
让开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施，让 Agentic AI 真正进入企业生产环境。
评论