作者：江昱
輿情分析是企業感知市場脈搏、預警公關危機的“聽診器”，然而傳統的輿情分析系統更像是一個個“手工作坊”，面臨數據收集效率低、分析深度不夠、實時性差等問題，經常反饋之後，等企業拿到報告時，輿論熱點早已轉移，錯過最佳時間。這些挑戰，正是所有輿情繫統開發者共同的痛點。
本方案將基於真實的代碼實現，向您介紹如何使用函數計算 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_sandboxAgentRun 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 客户羣”與我們交流，釘釘羣號：134570017218。
快速瞭解函數計算 AgentRun：
一句話介紹： 函數計算 AgentRun 是一個以高代碼為核心的一站式 Agentic AI 基礎設施平台。秉持生態開放和靈活組裝的理念，為企業級 Agent 應用提供從開發、部署到運維的全生命週期管理。
函數計算 AgentRun 架構圖
AgentRun 運行時基於阿里雲函數計算 FC 構建，繼承了 Serverless 計算極致彈性、按量付費、零運維的核心優勢。通過深度集成 AgentScope、Langchain、RAGFlow、Mem0 等主流開源生態。AgentRun 將 Serverless 的極致彈性、零運維和按量付費的特性與 AI 原生應用場景深度融合，助力企業實現成本與效率的極致優化，平均 TCO 降低 60%。
讓開發者只需專注於 Agent 的業務邏輯創新，無需關心底層基礎設施，讓 Agentic AI 真正進入企業生產環境。
