Agent Run：基於AgentRun構建輿情分析專家

準備工作

在開始之前，請確保您已完成以下準備工作：

1. 開通雲端服務並完成授權：

   • 確保您已擁有一個可用的阿里雲帳號。

   • 訪問並開通以下服務。首次訪問時，請根據頁面引導完成服務開通和RAM角色授權。

     • Function Compute FC 服務

     • FunctionAI AgentRun 服務

2. 擷取模型訪問憑證 (API-KEY)：

   • 本應用依賴大語言模型進行對話理解。請登入百鍊控制台，在密鑰管理頁面建立或複製一個API-KEY，後續步驟將使用此密鑰。

3. 通過模型管理建立LLM模型

   下文以配置百鍊中的qwen3-max模型為例，如果您想配置其他模型或自訂模型，請參見：大語言模型。

   • 進入模型管理頁面，點擊添加模型。

   • 選擇API模型，並配置以下參數：

     • 名称：qwen3-model；

     • 服務提供者：從下拉框中選擇阿里雲；

     • API端點：保持預設；

     • 配置具體模型：搜尋並選擇qwen3-max；

     • 憑證管理：選擇API密鑰，並粘貼上一步擷取的百鍊API Key；您也可以選擇使用已有憑證，並從下拉框中選擇您提前配置好的憑證，詳情可參見：憑證管理。

4. 通過Sandbox沙箱建立瀏覽器沙箱執行個體

   下文以控制台快速配置瀏覽器沙箱為例，如果您對其中的一些具體參數感興趣或想使用沙箱的API，請參見：BrowserTool瀏覽器。

   • 進入Sandbox沙箱頁面，點擊建立沙箱模板。

   • 選擇瀏覽器，點擊立即创建。在跳轉到的配置頁面中，填寫以下參數，其他保持預設即可：

     • 名稱：sandbox-in-opinion_analysis；

     • 執行角色ARN：AliyunAgentRunDefaultRole;

   • 點擊建立瀏覽器，等待沙箱建立成功。

選擇模板並部署

1. 在AgentRun控制台上方點擊探索按鈕，在熱門Agent專案模板中找到輿情分析專家，點擊部署。您也可以點擊卡片详情查看該模板概述。

2. 在部署頁面，參考以下樣本填寫配置參數：

   • 应用名称：為您的應用取一個易於識別的名稱，便於後續管理和識別，如：opinion_analysis。

   • 描述（可選）：簡要描述應用的功能，如：基於輿情分析專家模板建立的Agent。

   • 許可權配置：為AgentRun應用授權，以便其能訪問模型、工具等雲資源。

     • （推薦）點擊快速建立，系統將引導您建立一個名為 AliyunAgentRunDefaultRole 的預設角色，該角色已包含要求的權限；

     • 如需自訂許可權，您也可以點擊下拉框右側的添加按鈕，通過創建角色手動設定，並為其添加相應的權限原則，詳情可參見按需配置執行角色。

   • 大語言模型：從下拉框中選擇您剛剛配置的qwen3-model，並選擇qwen3-max模型。如果您未提前配置，可點擊下拉框右側的添加按鈕，按照準備工作步驟3立即配置。

   • 瀏覽器沙箱：從下拉框中選擇您剛剛配置的sandbox-in-opinion_analysis，如果您未提前配置，可點擊下拉框右側的添加按鈕，按照準備工作步驟4立即配置。

   • 點擊確認建立，等待應用部署完成。

   • 部署成功後，頁面會顯示應用的訪問連結。點選連結即可進入WebUI體驗頁面。後續您也可以通過在Agent控制台找到該應用，點擊應用详情，在概覽與配置頁面右側的整合配置中找到訪問地址。

應用體驗

1. 進入體驗頁面後，可以看到輿情分析系統頁面，此時可以輸入一個關鍵詞或主題進行輿情分析，樣本輸入：新能源汽車。分析過程中，系統會調用您配置的瀏覽器沙箱sandbox-in-opinion_analysis，AI控制雲上的瀏覽器進行資料檢索。

2. 分析完成之後，系統會整理所有採集到的資料和資訊，最終產生文字+圖表的可視化報告

查看應用詳情

應用部署成功後，您可以在AgentRun控制台對應用進行全方位的管理和營運。在應用列表中找到您建立的應用，點擊進入详情頁面，左側導覽列提供了以下核心功能：

• 概覽與配置：快速查看應用的基本資料、運行時配置（如模型、記憶體規格），並在此管理環境變數。

• 代碼與調試：支援線上查看、編輯和調試，並進行線上二次開發。

• 版本與灰階：提供版本管理能力，支援通過灰階發布將少量流量切換到新版本進行測試，驗證通過後再全量上線。每個版本都有一個臨時網域名稱供測試使用。

• 整合與發布：可參考Agent整合與發布，將開發的Agent快速整合到您的前端網頁、後端應用等。

• 彈性與執行個體：查看當前正在啟動並執行執行個體列表和狀態，並配置靈活的Auto Scaling策略，讓應用根據業務負載自動增減執行個體。

• 可觀測性：通過整合阿里雲應用即時監控服務（ARMS），可以獲得代碼級的效能診斷、請求追蹤、異常監控和警示能力，保障應用的穩定運行。

系統架構設計

整個輿情分析系統採用分層架構設計，核心思想是通過代碼嚴格控制流程程執行順序，而非依賴 LLM 的自主決策。

核心優勢

1. 安全隔離的執行環境

   傳統輿情系統通常直接在伺服器上運行爬蟲程式，面臨著安全風險和環境汙染問題。當某個網站的反爬機制觸發時，可能影響整個伺服器的穩定性。而 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}")

2. 真實瀏覽器環境類比

   傳統爬蟲方案通常使用簡單的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()

3. 可視化調試能力

   Function Compute AgentRun 最獨特的優勢是提供了即時的 VNC 預覽功能，開發人員和使用者可以即時觀察瀏覽器的操作過程。這種透明性在傳統方案中是無法實現的，它不僅有助於調試和最佳化採集邏輯，還能讓使用者直觀地瞭解系統的工作狀態。

4. 彈性擴充和故障恢複

   傳統系統在面臨大規模採集任務時，往往需要複雜的分布式架構設計。而 Function Compute 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

前端VNC整合實現

1. 動態庫載入機制

   前端 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);
       });
   }

2. 多協議適配

   考慮到部署環境的複雜性，VNC 組件實現了 HTTP/HTTPS 環境下的 WebSocket 通訊協定自適應。

   import { useCallback } from 'react';
   
   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;
   }, []);

後端核心實現

1. 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,
   )

2. 流式輸出與即時反饋

   傳統輿情系統通常採用批處理模式，使用者需要等待很長時間才能看到結果。而基於 Function Compute AgentRun 的系統實現了真正的流式輸出，使用者可以即時觀察每個處理步驟的進展。這種即時性不僅提升了使用者體驗，也便於及時發現和解決問題。

   import time
   
   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)

3. 智能資料品質控制

   系統實現了嚴格的資料品質控制機制，通過多維度評估確保收集到的資料具有較高的相關性和價值。這種品質控制在傳統系統中往往是缺失的，導致大量噪音資料影響分析結果。

   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))

深度內容抓取技術

1. 平台適配策略

   不同的社交媒體平台具有不同的頁面結構和內容組織方式，傳統系統往往採用統一的抓取策略，導致資料品質參差不齊。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"},
           ]

2. LLM 驅動的智能探索

   系統創新性地引入了 LLM 驅動的智能探索機制，讓 AI 決定是否需要深入抓取某個頁面的額外內容，如評論區、相關推薦等。這種智能決策大大提高了資料擷取的效率和針對性。

   import json
   
   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)
   

智能分析與報告產生

1. 標準化情感分析

   系統實現了基於關鍵詞詞典的情感分析演算法，相比傳統的機器學習模型，這種方法更加透明和可控。

   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

2. 流式報告產生

   報告產生過程採用流式輸出，使用者可以即時觀察報告的撰寫過程，這種體驗是傳統系統無法提供的。

   import asyncio
   
   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)

部署與營運優勢

1. 簡化的部署流程

   相比傳統輿情系統需要複雜的分布式爬蟲叢集部署，AgentRun 系統的部署相對簡單。只需要配置好環境變數和 AgentRun Sandbox 模板，系統就能自動管理瀏覽器執行個體的建立和銷毀。

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

2. 自動化營運能力

   系統內建了完善的監控和自恢複機制，大大降低了營運複雜度。當檢測到異常時，系統能夠自動重建資源，保證服務的連續性。

   import { useEffect, useRef } from 'react';
   
   // 串連失敗時自動重連（每 10 秒嘗試一次）
   useEffect(() => {
       if (status === 'error' && active && rfbLoaded) {
           reconnectTimerRef.current = setTimeout(() => {
               cleanupRfb();
               lastUrlRef.current = null;
               fetchVncUrl(true);
           }, RECONNECT_INTERVAL);
       }
   }, [status, active, rfbLoaded]);

效能與擴充性分析

1. 並發處理能力

   傳統系統的並發能力往往受限於單機資源，而 Function Compute AgentRun 系統可以根據需要動態建立多個 Sandbox 執行個體，實現真正的水平擴充。系統通過非同步編程模型和串連池管理，能夠高效處理大量並發請求。

   import uvicorn
   
   uvicorn.run(
       "main:app",
       host="0.0.0.0",
       port=8000,
       log_level="info",
       timeout_keep_alive=120,
       limit_concurrency=100,  # 支援高並發
   )

2. 資源彈性管理

   系統實現了智能的資源管理原則，能夠根據任務負載動態調整 Sandbox 執行個體數量。這種彈性擴充能力是傳統固定架構難以實現的。

   from typing import Any, Dict, List
   
   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

總結

基於Function Compute AgentRun 構建的輿情分析系統，在安全性、可靠性、可觀測性和擴充性方面，均展現出相較於傳統方案的顯著優勢。它通過隔離的 Sandbox 環境解決了安全與環境汙染的核心痛點；以即時的 VNC 預覽提供了前所未有的過程透明度；並藉助內建的故障檢測與自愈機制，極大降低了營運複雜度。更重要的是，該系統實現了從資料擷取、智能分析到流式報告產生的端到端自動化。使用者可以即時觀察分析全過程，獲得了遠超傳統批處理模式的卓越體驗。這種將複雜流程高度自動化的能力，結合 AI 技術的持續進步，將為企業和機構的輿情洞察與決策支援帶來深刻的價值提升。