AgentRun 实战：快速构建 AI 舆情实时分析专家

舆情分析是企业感知市场脉搏、预警公关危机的“听诊器”，然而传统的舆情分析系统更像是一个个“手工作坊”，面临数据收集效率低、分析深度不够、实时性差等问题，经常反馈之后，等企业拿到报告时，舆论热点早已转移，错过最佳时间。这些挑战，正是所有舆情系统开发者共同的痛点。
本方案将基于真实的代码实现，向您介绍如何使用函数计算 AgentRun 平台，构建一个现代化的“舆情分析专家”，该系统不仅实现了从数据采集到报告生成的可视化、全流程自动化，更通过流式架构，让洞察实时呈现。
系统架构设计

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

快速体验和效果预览

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

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

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

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

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

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

效果体验

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

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

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

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

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

安全隔离的执行环境

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

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

复制代码

真实浏览器环境模拟

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

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

复制代码

可视化调试能力

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

弹性扩展和故障恢复

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

1. async def recreate_sandbox_if_closed(sandbox_id: str, error_message: str):
2.     """智能故障检测和自动重建机制"""
4.     closed_error_patterns = [
5.         "Target page, context or browser has been closed",
6.         "Browser has been closed",
7.         "Connection closed",
8.     ]
9.    
10.     is_closed_error = any(pattern.lower() in error_message.lower()
11.                          for pattern in closed_error_patterns)
12.    
13.     if is_closed_error:
14.         await remove_sandbox(sandbox_id)
15.         new_sandbox = await create_browser_sandbox()
16.         return new_sandbox

复制代码

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

Agent 工具链设计

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

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

复制代码

流式输出与实时反馈

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

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

复制代码

智能数据质量控制

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

1. async def evaluate_relevance(keyword: str, title: str, snippet: str) -> float:
2.     """多维度相关性评估，确保数据质量"""
4.     text = f"{title} {snippet}"
6.     text_lower = text.lower()
7.    
8.     # 检测关键词匹配度
10.     has_chinese_keyword = any('\u4e00' <= char <= '鿿' for char in keyword)
11.     result_has_chinese = any('\u4e00' <= char <= '鿿' for char in text)
12.    
13.     # 中文关键词必须在结果中有中文内容
15.     if has_chinese_keyword and not result_has_chinese:
16.         return 0.0
18.    
19.     # 排除明显的无关网站
21.     irrelevant_patterns = [
22.         "calculator", "deepseek", "chegg", "stackoverflow",
23.         "翻译", "dictionary", "词典"
25.     ]
26.     if any(pattern in text_lower for pattern in irrelevant_patterns):
27.         return 0.0
29.         
30.     # 计算相关性得分
32.     score = 0.0
34.     if keyword in text:
35.         score += 0.6  # 基础分
37.    
38.     # 时效性加分
40.     time_keywords = ["最新", "今日", "近日", "2024", "2025"]
41.     if any(tk in text for tk in time_keywords):
42.         score += 0.1
44.    
45.     return max(0.0, min(1.0, score))

复制代码

部署与运维优势

简化的部署流程

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

1. async def explore_page_with_llm(page, keyword: str, url: str, source: str, initial_content: str):
2.     """基于平台特性的智能内容抓取"""
4.    
5.     if "weibo.com" in url:
6.         # 微博特定的评论和转发抓取
8.         available_actions = [
9.             {"action": "view_comments", "selector": ".WB_feed_expand, [class*='comment']"},
10.             {"action": "view_retweets", "selector": ".WB_feed_expand, [class*='repost']"},
11.         ]
12.     elif "zhihu.com" in url:
13.         # 知乎回答和评论抓取
15.         available_actions = [
16.             {"action": "view_more_answers", "selector": ".AnswerItem, .List-item"},
17.             {"action": "view_comments", "selector": ".Comments-container, .CommentItem"},
18.         ]
19.     elif "bilibili.com" in url:
20.         # B站视频评论抓取
22.         available_actions = [
23.             {"action": "view_comments", "selector": ".reply-item, .root-reply"},
24.             {"action": "view_related", "selector": ".video-page-card, .recommend-list"},
25.         ]

复制代码

自动化运维能力

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

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

复制代码

性能与扩展性分析

并发处理能力

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

1. function loadScript(url) {
2.     return new Promise(function(resolve, reject) {
3.         var script = document.createElement('script');
4.         script.src = baseUrl + url;
5.         script.onload = resolve;
6.         script.onerror = function() {
7.             // 本地加载失败，尝试 CDN
9.             var fallbackUrl = url.includes('wordcloud')
10.                 ? 'https://cdn.jsdelivr.net/npm/echarts-wordcloud@2.1.0/dist/echarts-wordcloud.min.js'
12.                 : 'https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js';
13.             var fallbackScript = document.createElement('script');
14.             fallbackScript.src = fallbackUrl;
15.             fallbackScript.onload = resolve;
16.             fallbackScript.onerror = reject;
17.             document.head.appendChild(fallbackScript);
18.         };
19.         document.head.appendChild(script);
20.     });
21. }

复制代码

资源弹性管理

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

1. const adjustWebSocketUrl = useCallback((url: string): string => {
2.     const isHttps = window.location.protocol === 'https:';
3.    
4.     if (!isHttps && url.startsWith('wss://')) {
5.         return url.replace('wss://', 'ws://');
6.     }
7.    
8.     if (isHttps && url.startsWith('ws://')) {
9.         return url.replace('ws://', 'wss://');
10.     }
11.    
12.     return url;
13. }, []);

复制代码

总结

基于 函数计算 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 真正进入企业生产环境。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！