Multi Agents Collaboration OS：Web DeepSearch System

背景：多智能體協作驅動網絡信息處理的范式革新

隨著大型語言模型（LLM）能力的突破性進展，人工智能正從“單點賦能”向“系統協同”演進。傳統單一智能體在復雜業務場景中逐漸顯露局限：面對需多維度知識整合、動態任務拆解、閉環反饋優化的深度信息處理需求時，其往往難以滿足效率與精度的雙重訴求。尤其在金融分析、市場調研、學術研究等依賴多源數據融合的領域，人工主導的信息采集與分析流程存在效率低下、易出錯、難擴展等問題，而現有工具鏈又缺乏對“搜索-解析-評估-總結”全流程的自動化閉環支持。

在此背景下，多智能體協作系統（Multi-Agent System, MAS）成為突破瓶頸的關鍵路徑。通過模擬團隊協作模式，系統可將復雜任務拆解為子任務，由具備差異化能力的智能體并行或串行處理，最終通過知識融合與邏輯整合輸出結果。然而，構建此類系統面臨三大核心挑戰：

高定制化門檻：不同業務場景需定義差異化智能體角色、協作流程及知識庫接入方式，傳統開發模式周期長、成本高；
動態協作復雜度：如何實現智能體間任務路由、沖突消解與結果驗證，需解決非線性交互帶來的工程難題；
容錯與自優化能力：面對網絡波動、數據缺失、模型幻覺等干擾因素，系統需具備評估反饋與自動糾錯機制。

針對上述痛點，我們推出Multi Agents Collaboration OS ：Web DeepSearch System系統聚焦于信息檢索與深度分析場景，通過多智能體動態協作架構，實現了從“人工信息篩選”到“自動化深度報告生成”的跨越式升級。其核心創新在于：

多智能體協作 實現動態規劃的智能體工作流，支持多個智能體協同完成復雜任務
文檔解析器 支持解析Web結構化文件，包括PDF、DOCX、XLSX等格式的文檔處理
多引擎搜索 集成DuckDuckGo與SerpAPI等多種搜索引擎能力，提供全面的網絡信息檢索
智能提取代理 構建網頁內容結構化提取管道，自動抽取關鍵信息并組織為可用數據格式
研究報告生成 基于收集和處理的數據，自動化生成結構化的研究報告
自糾錯機制 對生成的信息進行評估與校驗，自動檢測錯誤并進行修正，同時反饋優化結果

核心模塊與代碼實踐

智能體設計

Web Search Agent ：聯網搜索獲取與用戶訴求相關的網頁地址，集合serpapi、tavily_search、duck duckgo等聯網搜索能力框架

# --- 聯網搜索智能體 ---
def web_search(query, max_results=10):"""Perform web search using SerpAPI"""params = {"engine": "google","q": query,"api_key": "your api key",  # Use environment variable"num": max_results  # Explicitly request number of results}try:search = GoogleSearch(params)time.sleep(5)results_dict = search.get_dict()  # Get full responsereturn [{"url": item.get("link"),"title": item.get("title"),"content": item.get("snippet")[:2000]}for item in results_dict.get("organic_results", [])if all([item.get("link"), item.get("title"), item.get("snippet")])][:max_results]except Exception as e:print(f"Search failed: {str(e)}")return []
from tavily import TavilyClientdef tavily_search(query, max_results=10):"""Perform a search using the Tavily API and return the results.Args:query (str): The search query.max_results (int): The maximum number of results to return.Returns:list: A list of dictionaries containing the search results."""# Initialize the Tavily clienttavily = TavilyClient(api_key="your API key")# Perform the searchresults = tavily.search(query=query, max_results=max_results)# Extract relevant information from the resultssearch_results = [{"url": result["url"],"title": result["title"],"content": result["content"]}for result in results["results"]]return search_results
#聯網搜索智能體：根據用戶的訴求獲取網址
def search_duckduckgo(query, max_results=10, timelimit_='y'):"""Search DuckDuckGo with error handling."""try:with DDGS() as ddgs:results = ddgs.text(query, max_results=max_results, timelimit=timelimit_)time.sleep(5)return [result['href'] for result in results]except Exception as e:print(f"Search failed: {e}")return []

Web Page System Agent：網頁頁面操作，獲取具體網頁頁面相關的二級甚至多級頁面地址及信息，帶自糾錯機制

#網頁操作自糾錯及評估
def evaluate_page_relevance(html_content: str, user_request: str) -> bool:"""使用LLM判斷頁面內容是否包含與用戶訴求相關的信息。返回True表示相關，False表示不相關。"""llm = ChatOpenAI(model=model_use, temperature=0.2)prompt = f"""你是一個網頁內容分析助手。請根據以下HTML內容，判斷該頁面是否包含與用戶訴求相關的信息。用戶訴求為: "{user_request}"頁面HTML內容如下:{html_content[:4000]}  # 只取前4000字符防止超長請只回答"相關"或"不相關"。"""human_message = f"Evaluating relevance for the request: {user_request}"response = llm.invoke([("system", prompt), ("human", human_message)])answer = response.content.strip()return "相關" in answer
#網頁系統智能體：自動判斷頁面是否包含用戶訴求相關信息，若無則嘗試翻頁/搜索等操作，獲取新的業務系統url
def webpage_system_agent(url_list: list, user_request: str, max_iterations: int = 5):"""處理多個URL的增強版本，保持原始輸出格式不變"""final_results = []for target_url in url_list:if not target_url.startswith(('http://', 'https://')):print(f"?? 無效URL格式已跳過: {target_url}")continueprint(f"\n🚀 處理初始URL: {target_url}")try:headers = {'User-Agent': 'MyGenericAgent/1.0 (LanguageModelBot; +http://mybot.example.com/info)'}response = requests.get(target_url, headers=headers, timeout=30)response.raise_for_status()current_url = response.urlhtml_content = response.textprint(f"\n🔄 從頁面提取模塊URL: {current_url}")# 原有提取邏輯保持不變llm = ChatOpenAI(model=model_use, temperature=0.2)prompt = f"""..."""  # 保持原有prompt不變human_message = f"Extracting relevant module URLs for: {user_request}"response = llm.invoke([("system", prompt), ("human", human_message)])# 保持原有JSON解析邏輯json_str = response.content.replace("```json", "").replace("```", "").strip()try:result = json.loads(json_str)except Exception as e:print(f"JSON解析錯誤: {e}")result = {"urls": []}collected_urls = result.get("urls", [])print(f"🧠 提取到{len(collected_urls)}個相關URL")# 保持原有評估邏輯if evaluate_page_relevance(html_content, user_request):print(f"? 當前頁面包含所需信息: {current_url}")final_results.append(current_url)continue# 保持原有迭代邏輯visited = set()for iteration in range(max_iterations):found = Falsefor url in collected_urls:abs_url = url if url.startswith("http") else urljoin(current_url, url)if abs_url in visited:continuevisited.add(abs_url)print(f"🔗 嘗試訪問模塊頁面: {abs_url}")try:response = requests.get(abs_url, headers=headers, timeout=30)response.raise_for_status()html = response.textif evaluate_page_relevance(html, user_request):print(f"? 找到相關頁面: {abs_url}")final_results.append(abs_url)found = Truebreakexcept Exception as e:print(f"訪問失敗: {abs_url} - {str(e)}")if not found:print(f"第{iteration+1}輪未找到新頁面")breakexcept Exception as e:print(f"處理URL {target_url} 時發生錯誤: {str(e)}")# 去重并保持原始輸出格式return list(set(final_results)) if final_results else None

Web Structure Agent：網頁結構化信息提取，表格、文件、圖片及網頁文本內容

def fetch_html(url: str):import requeststry:headers = {'User-Agent': 'MyGenericAgent/1.0 (LanguageModelBot; +http://mybot.example.com/info)'}response = requests.get(url, headers=headers, timeout=30)response.raise_for_status()return response.textexcept Exception as e:print(f"Error fetching URL {url}: {e}")return Nonedef extract_text_content(html_content: str, url: str) -> str:"""從HTML中提取所有文本內容并保存到文件（支持中文）"""soup = BeautifulSoup(html_content, 'html.parser')# 移除腳本和樣式標簽for script_or_style in soup(["script", "style"]):script_or_style.decompose()# 獲取文本，用換行符分隔，并去除多余空白text = soup.get_text(separator='\n', strip=True)# 文件名基于URL (簡化版)time_ = datetime.now().strftime("%Y%m%d_%H%M%S")parsed_url = urlparse(url)filename = f"{parsed_url.netloc.replace('.', '_')}_{time_}_text.txt"filepath = OUTPUT_DIR / filenametry:with open(filepath, 'w', encoding='utf-8') as f:  # 確保使用UTF-8編碼f.write(text)return str(filepath.relative_to(pathlib.Path.cwd())) # 返回相對路徑except IOError as e:print(f"Error saving text content to {filepath}: {e}")return Nonedef extract_image_urls(html_content: str, base_url: str) -> list:"""從HTML中提取所有圖片的絕對URL"""soup = BeautifulSoup(html_content, 'html.parser')image_urls = []for img_tag in soup.find_all('img'):src = img_tag.get('src')if src:# 將相對URL轉換為絕對URLabsolute_url = urljoin(base_url, src)image_urls.append(absolute_url)return list(set(image_urls)) # 去除重復項
def extract_file_links(html_content: str, base_url: str) -> list:"""提取PDF/Excel/Word/TXT等文件鏈接"""soup = BeautifulSoup(html_content, 'html.parser')file_links = []# 匹配常見文檔格式的正則表達式pattern = re.compile(r'\.(pdf|xlsx?|docx?|txt)$', re.IGNORECASE)for tag in soup.find_all(['a', 'link']):href = tag.get('href')if href and pattern.search(href):absolute_url = urljoin(base_url, href)file_links.append(absolute_url)return list(set(file_links))  # 去除重復項
def extract_structured_data_tables(html_content: str, url: str) -> str:"""從HTML中提取所有表格數據，并保存為JSON文件。這是一個非常基礎的表格提取示例。"""soup = BeautifulSoup(html_content, 'html.parser')tables_data = []for table_index, table_tag in enumerate(soup.find_all('table')):current_table_data = []headers = [header.get_text(strip=True) for header in table_tag.find_all('th')]for row_tag in table_tag.find_all('tr'):cells = row_tag.find_all(['td', 'th']) # 包括表頭單元格以防萬一if not cells:continuerow_data = [cell.get_text(strip=True) for cell in cells]# 如果表頭存在且與行數據長度匹配，可以構造成字典列表if headers and len(headers) == len(row_data) and any(h for h in headers): # 確保表頭不為空current_table_data.append(dict(zip(headers, row_data)))else: # 否則，作為列表存儲current_table_data.append(row_data)if current_table_data:tables_data.append({"table_index": table_index,"data": current_table_data})if not tables_data:return None# 文件名基于URL (簡化版)parsed_url = urlparse(url)time_ = datetime.now().strftime("%Y%m%d_%H%M%S")filename = f"{parsed_url.netloc.replace('.', '_')}_{time_}_structured_tables.json"filepath = OUTPUT_DIR / filenametry:with open(filepath, 'w', encoding='utf-8') as f:json.dump(tables_data, f, ensure_ascii=False, indent=4)return str(filepath.relative_to(pathlib.Path.cwd())) # 返回相對路徑except IOError as e:print(f"Error saving structured data to {filepath}: {e}")return Noneexcept TypeError as e:print(f"Error serializing structured data for {url}: {e}")return Nonedef generate_consolidated_output_json(data_list: list):"""Generate a single JSON file consolidating metadata from all URLs."""time_ = datetime.now().strftime("%Y%m%d_%H%M%S")filename = f"{time_}_consolidated_metadata.json"filepath = OUTPUT_DIR / filenametry:with open(filepath, 'w', encoding='utf-8') as f:json.dump(data_list, f, ensure_ascii=False, indent=4)return str(filepath.relative_to(pathlib.Path.cwd()))except (IOError, TypeError) as e:print(f"Error saving consolidated metadata JSON: {e}")return None
#網頁信息提取智能體：提取網頁信息并生成元數據
# 該智能體負責提取網頁信息，包括文本、圖片、表格等，并生成元數據
def intelligent_agent(url_list: list):"""Intelligent agent function that integrates all modules to process multiple URLs:- Fetch HTML content via each URL- Extract text content (supports Chinese)- Extract image URLs- Extract structured data (tables)- Generate consolidated metadata JSON for all URLs"""consolidated_data = []for target_url in url_list:if not target_url.startswith(('http://', 'https://')):print(f"Invalid URL format: {target_url}. Skipping.")continueprint(f"Processing URL: {target_url}")html = fetch_html(target_url)if not html:print(f"Failed to fetch content for {target_url}.")continueprint("HTML content loaded successfully.")# Extract and save text contenttext_path = extract_text_content(html, target_url)print(f"Text content saved to: {text_path}" if text_path else "Text extraction failed.")# Extract image URLsimage_urls = extract_image_urls(html, target_url)print(f"Extracted {len(image_urls)} image URLs.")# Extract structured data (tables)structured_data_path = extract_structured_data_tables(html, target_url)print(f"Structured data saved to: {structured_data_path}" if structured_data_path else "Table extraction failed.")file_links = extract_file_links(html, target_url)print(f"Extracted {len(file_links)} file links.")# Append individual URL data to consolidated listconsolidated_data.append({"page_url": target_url,"text_content_file_path": text_path,"image_urls": image_urls,"structured_data_file_path": structured_data_path,"document_path": file_links,})# Generate consolidated output JSONfinal_json_path = generate_consolidated_output_json(consolidated_data)if final_json_path:print(f"Consolidated metadata JSON generated at: {final_json_path}")return final_json_pathelse:print("Failed to generate consolidated metadata JSON.")return None

Analyze Summarize Agent：內容聚合與總結

## 分析及總結智能體：根據用戶的問題進行網頁信息分析及總結
def analyze_and_summarize(webpage_content_path: str, user_request: str):"""根據用戶問題，對網頁內容進行深度分析，并生成一份包含數據、圖表等元素的深度研究報告。"""with open(webpage_content_path, 'r', encoding='utf-8') as f:webpage_content_list = json.load(f)# 獲取網頁表格內容if not isinstance(webpage_content_list, list):raise ValueError("Expected a list of webpage metadata dicts")aggregated_text = []aggregated_tables = []aggregated_images = set()aggregated_urls = []# Process each metadata dict in the listfor webpage_content in webpage_content_list:# 1. 獲取網頁文本內容text_content_info = Nonetext_content_path = webpage_content.get('text_content_file_path')if text_content_path:try:with open(text_content_path, 'r', encoding='utf-8') as f:text_content_info = f.read()aggregated_text.append(text_content_info)except Exception as e:print(f"? Failed to read text content from {text_content_path}: {e}")# 2. 獲取網頁表格數據table_data_info = Nonetable_data_path = webpage_content.get('structured_data_file_path')if table_data_path:try:with open(table_data_path, 'r', encoding='utf-8') as f:table_data_info = json.load(f)aggregated_tables.extend(table_data_info)except Exception as e:print(f"? Failed to read table data from {table_data_path}: {e}")# 3. 獲取網頁圖片鏈接image_urls = webpage_content.get('image_urls', [])if image_urls:aggregated_images.update(url for url in image_urls if url.startswith(('http://', 'https://')))web_url = webpage_content.get('page_url')if web_url:aggregated_urls.append(web_url)# 合并所有文本內容document_path = webpage_content.get('document_path', [])# web_browser_auto_page.pyimport pathlib# 修改后的文檔處理邏輯if document_path:for doc_path in document_path:try:if not pathlib.Path(doc_path).exists():print(f"文件不存在: {doc_path}")continuefile_content = FileParser.parse(doc_path)aggregated_text.append({"file_path": doc_path,"content": file_content[:10000]  # 截取前5000字符防止過大})except Exception as e:print(f"文檔解析失敗 ({doc_path}): {str(e)}")aggregated_text.append({"file_path": doc_path,"error": str(e)})time.sleep(4)  # 避免請求過快# 生成分析和總結的提示web_content_prompt = f"""網頁鏈接是: {aggregated_urls}網頁文本內容是: {aggregated_text}網頁圖片鏈接是: {aggregated_images}網頁表格數據是: {aggregated_tables}請根據以上信息，分析并總結出與用戶訴求相關的信息。"""llm = ChatOpenAI(model=model_use, temperature=0.2,max_tokens=118000)prompt = f"""你是一個網頁內容分析助手。請根據以下HTML內容，分析并總結出與用戶訴求相關的信息。用戶訴求為: "{user_request}"頁面HTML內容及圖片、表格、文本等信息如下:{web_content_prompt[:50000]} 請生成一份包含數據、圖表等元素的深度研究報告。報告要求：1.輸出內容以markdown格式呈現，包含標題、段落、列表等結構。2.輸出內容包括：研究背景、研究方法、研究結果、結論和建議等部分,其中背景部分200-500字；方法部分200-500字，方法部分需要輸出數據來源，如網址、文件等；結果部分400-700字，對于數據需要使用表格形式輸出，結論和建議部分200-500字。3.數據可視化：引用網頁提供的合適的圖表圖片，以markdown語法嵌入圖片，如：![圖片名稱](圖片路徑)。4.輸出內容應簡潔明了,但是必須足夠豐富，輸出內容有極強的洞察，同時避免冗長的描述。【注意】你需要完整根據用戶訴求和所有網頁內容進行分析和總結，輸出一份完整的深度研究報告。"""human_message = f"Analyzing and summarizing for the request: {user_request}"response = llm.invoke([("system", prompt), ("human", human_message)])return response.content.strip()

協作機制設計
-智能體信息聚合：#上述各類智能體信息集合：智能體名稱，智能體詳細描述信息（功能、角色、使用方式），參數，預期輸入，預期輸出。

agents_info = {"intelligent_agent": {"description": "網頁信息提取智能體：提取網頁信息并生成元數據","function": "負責提取網頁信息，包括文本、圖片、表格等，并生成元數據，輸出網頁元數據文件地址","parameters": {"url_list": {"type": "dict","description": "目標網頁的URL list，格式為['https://example.com/page1', 'https://example.com/page2']"}}},"webpage_system_agent": {"description": "網頁系統智能體：自動判斷頁面是否包含用戶訴求相關信息，若無則嘗試翻頁/搜索等操作，獲取新的業務系統url","function": "1.自動判斷頁面是否包含用戶訴求相關信息，若無則嘗試翻頁/搜索等操作，獲取新的業務系統url;2.可以深度研究用戶網頁系統，提取相關模塊URL","parameters": {"url_list": {"type": "list","description": "目標網頁的URL list，格式為['https://example.com/page1', 'https://example.com/page2']"},"user_request": {"type": "string","description": "用戶的請求"},"max_iterations": {"type": "integer","description": "最大迭代次數"}}},"analyze_and_summarize": {"description": "分析及總結智能體：根據用戶的問題進行網頁信息分析及總結","function": "根據用戶問題，處理網頁元數據信息，根據元信息對網頁內容進行深度分析，并生成一份包含數據、圖表等元素的深度研究報告","parameters": {"webpage_content_path": {"type": "string","description": "網頁元數據文件地址，數據的文件地址，如xxx_metadata.json"},"user_request": {"type": "string","description": "用戶的請求"}}},"web_search": {"description": "聯網搜索智能體：根據用戶的訴求獲取網址列表","function": "負責聯網搜索，根據用戶的訴求獲取相關網頁的URL列表","parameters": {"target_url": {"type": "string","description": "目標網頁的URL"}}}
}

智能體參數設計及聚合

#只抽取智能體的參數
agent_parameters_info = {"intelligent_agent": {"url_list": {"type": "list","description": "目標網頁的URL list，格式為['https://example.com/page1', 'https://example.com/page2']"},},"webpage_system_agent": {"url_list": {"type": "list","description": "目標網頁的URL list，格式為['https://example.com/page1', 'https://example.com/page2']"},"user_request": {"type": "string","description": "用戶的請求"    },      },"analyze_and_summarize": {"webpage_content_path": {"type": "string","description": "基礎網頁源信息文件地址，網頁列表元數據文件地址，如xxx_metadata.json"},"user_request": {"type": "string","description": "用戶的請求"}},"web_search": {"query": {"type": "string","description": "用戶輸入的查詢"},"max_results": {"type": "integer","description": "最大結果數"}},
}

智能體協作機制設計
– 智能體參數生成

##定義一個智能體信息共享路由:根據用戶的問題獲取智能體的輸入參數
def get_agent_parameters(user_prompt: str,last_agent_response=None,agents_info=agent_parameters_info):llm = ChatOpenAI(model=model_use, temperature=0.2)message = f"""你是一個智能體信息共享助手。根據用戶的問題，獲取智能體的輸入參數。用戶問題為: "{user_prompt}"智能體信息為: {agents_info}前一個智能體的輸出為: {last_agent_response}請根據用戶問題，獲取智能體的輸入參數。[警告]請不要杜撰智能體參數信息，根據各個智能體的輸出，準確的完整的輸出各個智能體的輸入參數。【建議】涉及到網址搜索的，最大搜索結果數至少為10個。你應該輸出一個JSON對象，包含智能體的名稱和輸入參數。輸出的信息格式如下：'''json{{"<agent_name>": {{"target_url": "https://example.com"}},.....}}'''"""human_message = f"根據用戶問題，獲取智能體的輸入參數: {user_prompt}"response = llm.invoke([("system", message), ("human", human_message)])json_str = response.content.replace("```json", "").replace("```", "").strip()try:result = json.loads(json_str)except Exception as e:print(f"? JSON parse error: {e}")result = {"agent_name": None, "input_parameters": {}}return result

– 多智能體協作機制

#定義一個多智能體協作機制設計和輸出的模塊:輸出多智能體協作的方案，包括步驟信息，各步驟選擇的智能體及其任務，各個步驟的預期輸出
def multi_agent_cooperation(user_request: str,agents_info=agents_info,feedback=None,uesr_initial_url=None):llm = ChatOpenAI(model=model_use, temperature=0.2)uesr_initial_prompt = None#解析初始網頁內容：if uesr_initial_url:uesr_initial_prompt = fetch_html(uesr_initial_url)message = f"""你是一個多智能體協作助手。根據用戶問題，選擇合適的智能體進行任務處理。用戶問題為: "{user_request}"智能體信息為: {agents_info}。【警告】你只能使用上述智能體，不可以添加新的智能體。用戶提供的初始網址信息： {uesr_initial_prompt if uesr_initial_prompt else "無"}可以提供的歷史任務輸出反饋信息為: {feedback if feedback else "無"}請判斷用戶問題與用戶提供的初始網址信息的關聯性。請根據用戶問題，選擇合適的智能體進行任務處理。首先評估初始網頁系統是否滿足用戶需求，如果不滿足，使用智能體 webpage_system_agent 進行頁面信息提取和評估。如果智能體歷史執行結果沒有輸出用戶的訴求相關信息，則需要重新設計智能體協作策略，嘗試使用 webpage_system_agent 進行頁面信息提取和評估，獲取更多的網頁地址。業務場景及智能體協作方案：[注意]如果用戶的初始訴求未提供網址信息，則需要使用智能體 web_search 進行聯網搜索，獲取相關網頁的URL列表。(一)場景一：確定的業務場景，用戶需要提取網頁信息并生成元數據。1.用戶提供的url與用戶的問題強關聯，不需要評估是否需要對頁面進行翻頁/搜索等操作。2.只需要使用智能體 intelligent_agent 進行網頁信息提取。3.根據提取的網頁信息，使用智能體 analyze_and_summarize 生成一份包含數據、圖表等元素的深度研究報告。(二)場景二：1.用戶提供的url與用戶的問題弱關聯，可能需要對頁面進行翻頁/搜索等操作。2.使用智能體 webpage_system_agent 進行頁面信息提取和評估，獲取新的網頁信息及url。3.循環迭代使用智能體 intelligent_agent 進行上述獲取的網頁地址網頁信息提取，直到獲取到完整信息。4.根據提取的網頁信息，使用智能體 analyze_and_summarize 生成一份包含數據、圖表等元素的深度研究報告。(三)場景三：1.協作方案可能會有調整，請根據上一次協作方案執行情況，動態調整智能體協作策略。2.協作方案執行歷史中有些內容可以使用，請根據用戶訴求、協作方案執行評估情況、協作歷史信息，動態調整智能體協作策略。【注意】多智能體協作過程中可能會有報錯信息，這意味著網頁信息提取不完整或無法獲取到用戶訴求相關信息，需要重新設計智能體協作策略。你應該輸出一個JSON對象，包含多智能體協作的方案，包括步驟信息，各步驟選擇的智能體及其任務，各個步驟的預期輸出。輸出的信息格式如下：'''json{{"steps": [{{"step_name": "步驟1","step_description": "描述步驟1的功能","agent_name": "intelligent_agent",}},...],"task_name":"<總結一個任務名稱，如：金融市場分析>",}}'''"""human_message = f"根據用戶問題，選擇合適的智能體進行任務處理: {user_request}"response = llm.invoke([("system", message), ("human", human_message)])json_str = response.content.replace("```json", "").replace("```", "").strip()try:result = json.loads(json_str)except Exception as e:print(f"? JSON parse error: {e}")result = {"steps": []}return result

評估與反饋機制設計

智能體輸出結果反饋及自糾錯

#結果評估及反饋模塊:根據智能體的輸出結果，評估結果是否符合預期，并給出反饋
def evaluate_and_feedback(agent_output, user_request):llm = ChatOpenAI(model=model_use, temperature=0.2)message = f"""你是一位資深的智能體輸出評估與反饋專家。你的任務是基于用戶請求和智能體的輸出來進行嚴謹、細致的評估，并提供有建設性的反饋。**評估標準：*請綜合考慮以下方面對智能體的輸出進行評估：1.  **相關性 (Relevance):** 輸出是否直接回應了用戶的請求？是否跑題？2.  **準確性 (Accuracy):** 輸出中的信息是否事實正確？是否存在誤導性內容？3.  **完整性 (Completeness):** 輸出是否完整地回答了用戶問題的所有方面？是否遺漏了關鍵信息？4.  **清晰度 (Clarity):** 輸出是否易于理解？語言是否表達清晰、簡潔？5.  **幫助性 (Helpfulness):** 輸出是否真正幫助用戶解決了問題或滿足了其需求？6.  **安全性與恰當性 (Safety & Appropriateness):** 輸出是否包含不當、冒犯、有害或偏見內容？（如果適用）7.  **遵循指令 (Instruction Following):** 如果用戶請求中包含特定格式、角色或其他指令，智能體是否遵循了這些指令？8.  **內容評估**：核心訴求滿足即可，對網址、文檔、圖表、數據等并非一定需要的，需要根據用戶的訴求進行評估。根據智能體的輸出結果，評估結果是否符合預期，并給出反饋。智能體輸出結果為: "{agent_output}"用戶問題為: "{user_request}"請根據智能體的輸出結果，評估結果是否符合預期，并給出反饋。【注意】請完整準確的評估智能體的輸出是否符合用戶訴求，對要求輸出數據、網址、文件、結果等格外關注，保證輸出的準確性。【警告】你不能杜撰智能體的輸出結果，你只能根據智能體的輸出結果進行評估和反饋。你應該輸出一個JSON對象，包含評估結果和反饋信息。輸出的信息格式如下：'''json{{"evaluation": "符合預期"、"不符合預期"或"基本符合預期","feedback": "具體的反饋信息""last_dialog_summary": "上一次對話的總結信息"}}'''"""human_message = f"根據智能體的輸出結果，評估結果是否符合預期，并給出反饋。"response = llm.invoke([("system", message), ("human", human_message)])json_str = response.content.replace("```json", "").replace("```", "").strip()result = json.loads(json_str)return result

***對話歷史總結：***多輪對話歷史

##歷史對話總結智能體
def summarize_last_dialog(user_request: str, last_agent_response: str):message = [("system","你是一個歷史對話總結智能體，你需要根據用戶的輸入和智能體的輸出，總結用戶的對話內容。""請注意，你需要根據用戶的輸入和智能體的輸出，準確的總結用戶的對話內容。""1.重點關注智能體應對用戶問題的回答情況。""2.請勿重復輸出用戶輸入。""3.對生成的網址、文件地址等重點關注，需要準確的總結。""4.整體內容輸出不超過500字。"),("human",f"用戶輸入：{user_request}，智能體輸出：{last_agent_response}")]llm = ChatOpenAI(model=model_use, temperature=0.2)response = llm.invoke(message)return response.content

***智能體輸出優化：***使其符合人類閱讀習慣

#結果輸出整理：對智能體的輸出結果進行整理，生成符合用戶需求和人類閱讀偏好的內容
def organize_agent_output(agent_output: str, user_request: str):message = [("system","你是一個結果輸出整理智能體，你需要根據智能體的輸出結果，生成符合用戶需求和人類閱讀偏好的內容。""請注意，你需要根據智能體的輸出結果，準確的生成符合用戶需求和人類閱讀偏好的內容。""1.重點關注智能體應對用戶問題的回答情況。""2.關注智能體的解決方案，輸出核心內容：網址、文件、信息等。""3.對生成的網址、文件地址等重點關注，需要準確的生成。""4.數據和核心數據需要使用表格形式展示。""5.內容使用markdown格式，核心內容加粗，圖片使用markdown語法嵌入圖片，如：![圖片名稱](圖片路徑)。""6.如果需要輸出“深度研究報告”，請參考輸出內容包括：研究背景、研究方法、研究結果、結論和建議等部分,其中背景部分200-500字；方法部分200-500字，方法部分需要輸出數據來源，如網址、文件等；結果部分1000-2000字，對于數據需要使用表格形式輸出，結論和建議部分200-500字，""7.[注意]智能體沒有輸出的內容不要生成，禁止你自己發揮。"),("human",f"用戶輸入：{user_request}，智能體輸出：{agent_output}")]llm = ChatOpenAI(model=model_use, temperature=0.2)response = llm.invoke(message)return response.content

用戶系統

def recognize_speech():recognizer = sr.Recognizer()with sr.Microphone() as source:# 使用 empty 占位符來動態更新狀態status_placeholder = st.empty()status_placeholder.info("🎤 正在聆聽，請說話...")try:audio = recognizer.listen(source, timeout=3)  # 設置超時時間text = recognizer.recognize_google(audio, language="zh-CN")  # 支持中文識別status_placeholder.success(f"? 語音識別已完成：{text}")return textexcept sr.WaitTimeoutError:status_placeholder.warning("?? 您沒有說話或麥克風未檢測到輸入。")return Noneexcept sr.UnknownValueError:status_placeholder.error("? 無法理解音頻")return Noneexcept sr.RequestError as e:status_placeholder.error(f"? 請求錯誤: {e}")return Nonedef process_input(prompt):# 初始化 session_state.messagesif "webmessages" not in st.session_state:st.session_state.webmessages = []#歷史對話信息if "webhistory" not in st.session_state:st.session_state.webhistory = Noneif "voice_input" not in st.session_state:st.session_state.voice_input = Nonefor webmessages in st.session_state.webmessages:if webmessages["role"] == "user":with st.chat_message("user"):st.markdown(webmessages["content"])elif webmessages["role"] == "assistant":with st.chat_message("assistant"):st.markdown(webmessages["content"])agents_response_history = None# 假設你的視頻文件名為 my_video.mp4 并且在與你的 Streamlit 腳本相同的目錄下if prompt :st.session_state.webmessages.append({"role": "user", "content": prompt})st.chat_message("user").markdown(prompt)prompt = f"用戶問題: {prompt}\n\n,歷史對話信息: {st.session_state.webhistory if st.session_state.webhistory else '無'}"with st.spinner('正在思考(Thinking)...'):with st.expander("智能體協作方案設計", expanded=True):multi_agent_plan = multi_agent_cooperation(prompt, agents_info=agents_info, feedback=st.session_state.webhistory)  st.write("### 智能體協作方案：")st.json(multi_agent_plan)agent_response = Nonelast_error = Nonemax_retries = 10retries = 0retry_count = 0success = Falsesuccess = Falsemax_evaluation_retries = 10  # Separate counter for evaluation retrieswhile retries < max_retries and retry_count < max_evaluation_retries and not success:try:# Reset plan for each retrycurrent_plan = multi_agent_plan["steps"].copy()for step in current_plan:agent_name = step["agent_name"]# Get parameters with history contextagent_parameters = get_agent_parameters(str(step), agent_response)# Execute agentagent_response = agent_functions[agent_name](**agent_parameters.get(agent_name, {}))# Process and display responseagent_response_ = organize_agent_output(agent_response, str(step))agents_response_history = summarize_last_dialog(str(step), agent_response_)with st.expander(f"智能體 {agent_name} 輸出結果", expanded=True):st.chat_message("assistant").markdown(f"##### 智能體{agent_name}輸出結果:\n{agent_response_}")# Evaluate resultsevaluation_result = evaluate_and_feedback(agent_response, str(step))if evaluation_result["evaluation"] == "不符合預期":st.warning(f"? 評估未通過: {evaluation_result['feedback']}")# Generate new plan with feedbackmulti_agent_plan = multi_agent_cooperation(prompt + f"\n評估反饋: {evaluation_result},最新智能體輸出: {agents_response_history}")retry_count += 1break  # Exit current plan executionelse:st.success("? 評估通過")st.session_state.webmessages.append({"role": "assistant", "content": agent_response})else:  # Only executed if all steps completed successfullysuccess = Trueexcept Exception as e:error_msg = f"執行異常: {str(e)}"st.error(error_msg)last_error = error_msg# Generate new plan with error contextmulti_agent_plan = multi_agent_cooperation(prompt + f"\n執行錯誤: {last_error}")retries += 1if not success:st.error("? 達到最大重試次數仍未完成請求")if st.session_state.webmessages:               history_summary =  summarize_last_dialog(prompt,str(st.session_state.webmessages[:-5]))##最后的對話信息st.session_state.messages[:-1]保存為markdown格式文件：地址為 output_data/deep_search/__deep_search.md#判斷文件是否存在，不存在則創建if not os.path.exists("output_data/deep_search"):os.makedirs("output_data/deep_search")##設計一個文件名稱，需要包含任務情況特色try:deep_search_file_name = f"{multi_agent_plan['task_name']}__deep_search.md" deep_search_file_path = os.path.join("output_data/deep_search", deep_search_file_name)with open(deep_search_file_path, "w") as f:f.write(st.session_state.webmessages[-1]['content'])#提供一個下載按鈕，可以下載markdown文件st.download_button(label="下載深度研究報告",data=open(deep_search_file_path, "rb").read(),file_name=deep_search_file_name,mime="text/markdown")except Exception as e:st.error(f"保存文件時出錯: {str(e)}")st.session_state.webhistory = history_summarydef web_browser_auto_page():# Initialize voice input in session stateif 'voice_input' not in st.session_state:st.session_state.voice_input = ""with st.sidebar:if st.button("🎤 語音輸入"):user_input = recognize_speech()if user_input:st.session_state.voice_input = user_inputst.rerun()st.title("🌍 Web Infomation DeepSearch")st.write("🗽該助手可以自動提取網頁信息，包括文本、圖片、表格等，并生成一份包含數據、圖表等元素的深度研究報告。")with st.sidebar:if st.button("🗑? Clear Chat History"):st.session_state.webmessages = []st.session_state.webhistory = Nonest.rerun() st.markdown("""<style>.main {background-color: #f8f9fa;padding: 20px;border-radius: 10px;}.welcome-message {font-size: 1.2em;color: #1f2937;line-height: 1.6;}.feature-bullet {color: #0066cc;font-weight: bold;}.example-box {background-color: #e0f2ff;padding: 15px;border-radius: 8px;margin: 10px 0;}.assistant-chat-message {float: left;background-color: #f1f8e9;border-radius: 20px;padding: 10px 15px;margin: 10px 0;max-width: 70%;}        </style>""", unsafe_allow_html=True)# 顯示平臺LOGO# 歡迎消息與功能指引st.markdown("""<div class="welcome-message">🌟 歡迎使用**網頁信息提取助手**！我可以幫您完成以下任務：<ul><li class="feature-bullet">🔗 聯網搜索獲取目標網頁</li><li class="feature-bullet">📄 提取網頁文本/圖片/表格</li><li class="feature-bullet">📊 生成深度數據分析報告</li><li class="feature-bullet">🔍 智能評估結果相關性</li></ul>🔍 <strong>使用建議：</strong><div class="example-box">示例問題：<br>"獲取5月生豬市場價格數據，并生成分析報告"<br>"從[中指云](https://www.cih-index.com/)，分析房地產市場趨勢，生成房市深度研究報告"<br>"提取指定網頁的表格信息：https://zhujia.zhuwang.com.cn/indexov.shtml"</div>請直接輸入您的需求，我將為您智能規劃執行方案。</div>""", unsafe_allow_html=True)# Handle voice input after normal chat inputif st.session_state.voice_input:prompt = st.session_state.voice_inputst.session_state.voice_input = ""  # Clear after useprocess_input(prompt)# Regular chat inputif prompt := st.chat_input("歡迎使用網頁信息提取助手！請問有什么可以幫助您的嗎？"):process_input(prompt)

Multi Agents Collaboration OS：Web DeepSearch System —>Demo

任務一：獲取5月生豬市場價格數據，并生成分析報告 <——無具體網址信息

協作方案生成
執行與結果
聯網搜索網址結果以及結構化信息提起
最終研究報告生成

任務二：從中指云 https://www.cih-index.com/)，分析房地產市場趨勢，生成房市深度研究報告 <——有具體網址信息

智能體協作方案
智能體執行結果
深度研究報告生成

總結：Web DeepSearch System

現存問題分析

1. 聯網搜索質量優化
當前系統通過 web_search 和 search_duckduckgo 實現多引擎搜索，但搜索結果質量受限于API返回數據的結構化程度。代碼中雖采用結果過濾機制（extract_webpage_info），但缺乏智能排序算法和可信度評估模塊。建議引入BERT重排序技術，結合網頁PageRank值和內容質量評估模型（如 evaluate_page_relevance 的增強版），在保證成本可控的前提下提升優質結果的召回率。

2. 專業報告生成質量
analyze_and_summarize 函數采用單一LLM生成模式，對專業領域知識的處理存在局限性。代碼中 prompt 設計未充分融入領域術語和結構化模版，導致生成內容學術性不足。可擴展為分層生成架構：先由 FileParser 提取結構化數據，再通過RAG模式調用領域知識庫，最后采用CoT思維鏈生成技術報告，提升專業術語使用準確性和分析深度。

3. 迭代生成成本控制
webpage_system_agent 的5輪迭代機制導致API調用次數線性增長。代碼中 max_iterations 采用固定閾值，缺乏動態終止策略。建議引入成本感知模塊，在 multi_agent_cooperation 中集成執行軌跡分析，當連續3輪評估分數（evaluate_and_feedback）提升小于10%時自動終止迭代，同時優化Playwright實例的復用率（async_playwright 上下文管理）。

系統擴展方向

1. 網頁操作自動化增強
基于現有 fetch_html_async 的Playwright能力，可擴展實現：

表單智能填充（page.type + CSS選擇器語義映射）
分頁導航（page.click + DOM樹結構分析）
動態數據捕獲（page.wait_for_selector + XPath模式庫）
建立網頁操作DSL描述語言，將 agent_parameters_info 擴展為可配置的操作指令集。

2. 智能體能力矩陣構建
在 agents_info 體系內集成：

數據分析智能體：對接 extract_structured_data_tables 輸出，擴展PandasAI交互接口
可視化智能體：基于 aggregated_images 實現Altair/Vega-Lite自動圖表生成
看板組裝智能體：通過 organize_agent_output 增強Markdown組件化輸出能力
構建智能體間數據流管道，通過 generate_consolidated_output_json 實現中間結果標準化交換。

3. 垂直場景深度適配
擴展 multi_agent_cooperation 的任務模版庫：

金融領域：集成巨潮資訊/東方財富數據規范，預置PE Ratio分析工作流
電商領域：構建價格監控模版（web_urls_process_query + 歷史價格曲線分析）
學術領域：連接CrossRef API，實現文獻綜述自動生成鏈路
通過領域特征注入（model config.json）和評估標準定制化（evaluate_and_feedback 規則引擎），提升場景化任務的完成度。