APPLICATION

DeepSeek-R1驱动的房地产AI代理

本文探讨如何使用 Smolagents 和 DeepSeek-R1 构建房地产代理，并利用工具进行网页抓取和数据导出。

admin

Feb 10, 2025 • 6 min read

AI 代理对于自动化复杂的推理任务至关重要。Smolagents 是由 Hugging Face 开发的轻量级 AI 代理框架，允许将大型语言模型 (LLM) 与现实世界的数据处理无缝集成。与其他高级模型相比，DeepSeek-R1 是一种开源 LLM，它以更低的成本增强了推理能力。我们使用 Ollama 来托管 DeepSeek-R1，从而实现高效的本地部署。

本文探讨如何使用 Smolagents 和 DeepSeek-R1 构建推理代理，并利用工具进行网页抓取和数据导出。

1、Smolagents 概述

Smolagents 提供了一个极简的 AI 代理框架，专为开发人员设计，以便高效地构建和部署智能代理。

Smolagents的主要功能：

简单：紧凑的代码库（约 1,000 行），易于开发。
代码代理：执行 Python 代码片段以提高准确性。
安全执行：在沙盒环境中运行代码。
多功能 LLM 集成：支持多个 LLM，包括 Hugging Face 模型和 OpenAI 的 GPT。
工具中心集成：允许从 Hugging Face Hub 共享和导入工具。

优点：

卓越的可组合性：嵌套函数调用增强了逻辑表示。
高效的对象处理：与 JSON 相比，简化了对象管理。
极致灵活性：执行任何计算操作。

2、DeepSeek-R1 概述

DeepSeek-R1 是由 DeepSeek AI 开发的开源 LLM。它提供：

以较低的成本提供高级推理能力。
高效处理基于文本的任务。
与 Smolagents 等代理框架集成。

3、实施：构建推理代理

我们将使用 Smolagents 和 DeepSeek-R1 开发一个推理代理，能够：

从 realtor.com 抓取房地产经纪人数据。
将抓取的数据保存到 CSV 文件中。
使用 DeepSeek-R1 执行推理任务。

3.1 导入所需库

from typing import Optional, Dict
from smolagents import CodeAgent, tool, LiteLLMModel , GradioUI, OpenAIServerModel
import requests
import os
import time
from bs4 import BeautifulSoup
import pandas as pd

Smolagents 模块：

CodeAgent：定义和管理 AI 代理。
tool：用于定义代理工具的装饰器。
LiteLLMModel：集成各种 LLM。
OpenAIServerModel：连接到外部模型。

其他库：

requests：用于发送 HTTP 请求。
BeautifulSoup：解析 HTML 以进行网页抓取。
pandas：处理结构化数据。

3.2 Web 抓取工具

@tool
def scrape_real_estate_agents(state: str, city_name: str, num_pages: Optional[int] = 2) -> Dict[str, any]:
    """Scrapes realtor.com for real estate agent information in specified city and state
    
    Args:
        state: State abbreviation (e.g., 'CA', 'NY')
        city_name: City name with hyphens instead of spaces (e.g., 'buffalo')
        num_pages: Number of pages to scrape (default: 2)
    """
    try:
        # Initialize results
        agent_names = []         # Names
        agent_phones = []        # Phone numbers
        agent_offices = []       # Office names
        pages_scraped = 0
        
        # Set up headers
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Connection": "keep-alive"
        }

        # Process pages
        for page in range(1, num_pages + 1):
            # Construct URL
            if page == 1:
                url = f'https://www.realtor.com/realestateagents/{city_name}_{state}/'
            else:
                url = f'https://www.realtor.com/realestateagents/{city_name}_{state}/pg-{page}'
            
            print(f"Scraping page {page}...")
            
            # Get page content
            response = requests.get(url, headers=headers)
            if response.status_code != 200:
                return {"error": f"Failed to access page {page}: Status code {response.status_code}"}

            soup = BeautifulSoup(response.text, features="html.parser")
            
            # Find all agent cards
            agent_cards = soup.find_all('div', class_='agent-list-card')
            
            for card in agent_cards:
                # Find name
                name_elem = card.find('div', class_='agent-name')
                if name_elem:
                    name = name_elem.text.strip()
                    if name and name not in agent_names:
                        agent_names.append(name)
                        print(f"Found agent: {name}")

                # Find phone
                phone_elem = card.find('a', {'data-testid': 'agent-phone'}) or \
                            card.find(class_='btn-contact-me-call') or \
                            card.find('a', href=lambda x: x and x.startswith('tel:'))
                
                if phone_elem:
                    phone = phone_elem.get('href', '').replace('tel:', '').strip()
                    if phone:
                        agent_phones.append(phone)
                        print(f"Found phone: {phone}")

                # Get office/company name
                office_elem = card.find('div', class_='agent-group') or \
                            card.find('div', class_='text-semibold')
                if office_elem:
                    office = office_elem.text.strip()
                    agent_offices.append(office)
                    print(f"Found office: {office}")
                else:
                    agent_offices.append("")
            
            pages_scraped += 1
            time.sleep(2)  # Rate limiting

        if not agent_names:
            return {"error": "No agents found. The website structure might have changed or no results for this location."}

        # Return structured data
        return {
            "names": agent_names,
            "phones": agent_phones,
            "offices": agent_offices,
            "total_agents": len(agent_names),
            "pages_scraped": pages_scraped,
            "city": city_name,
            "state": state
        }
        
    except Exception as e:
        return {"error": f"Scraping error: {str(e)}"}

代码说明：

设置自定义标头以避免被阻止。
获取 HTML 内容并对其进行解析。
查找代理姓名、电话号码和办公室详细信息。
将它们存储在列表中以进行结构化存储。

3.3 将数据导出到 CSV

@tool
def export_to_csv(scraped_data: Dict[str, any], output_filename: Optional[str] = None) -> str:
    """Exports scraped real estate agent data to a CSV file
    
    Args:
        scraped_data: Dictionary containing the results of the scraping
        output_filename: Optional filename for the CSV file (default: cityname.csv)
    """
    try:
        if "error" in scraped_data:
            return f"Error: {scraped_data['error']}"
            
        if not output_filename:
            output_filename = f"{scraped_data['city'].replace('-', '')}.csv"
            
        # Ensure all lists are of equal length
        max_length = max(len(scraped_data['names']), len(scraped_data['phones']), len(scraped_data['offices']))
        
        # Pad shorter lists with empty strings
        scraped_data['names'].extend([""] * (max_length - len(scraped_data['names'])))
        scraped_data['phones'].extend([""] * (max_length - len(scraped_data['phones'])))
        scraped_data['offices'].extend([""] * (max_length - len(scraped_data['offices'])))
        
        # Create DataFrame with just names, phones, and offices
        df = pd.DataFrame({
            'Names': scraped_data['names'],
            'Phone': scraped_data['phones'],
            'Office': scraped_data['offices']
        })
        
        df.to_csv(output_filename, index=False, encoding='utf-8')
        return f"Data saved to {output_filename}. Total entries: {len(df)}"
        
    except Exception as e:
        return f"Error saving CSV: {str(e)}"

代码说明如下：

目的：将抓取的数据保存到 CSV 文件中。
默认文件名：如果未提供文件名，则使用城市名称。
将抓取的数据转换为 DataFrame。
将其保存为 CSV 文件。
连接到本地托管的 DeepSeek-R1。
使用 OpenAIServerModel 执行。

3.4 集成 DeepSeek-R1

deepseek_model = OpenAIServerModel(
    model_id="deepseek-r1:7b",
    api_base="http://localhost:11434/v1",
    api_key="ollama"
)

代码说明如下：

连接到本地托管的 DeepSeek-R1。
使用 OpenAIServerModel 执行。

3.5 定义 AI 代理

agent = CodeAgent(
    tools=[scrape_real_estate_agents, export_to_csv],
    model=deepseek_model,
    additional_authorized_imports=["pandas", "bs4", "time"]
)

使用以下项定义代理：

抓取和导出工具。
DeepSeek-R1 用于推理。
授权导入以防止安全风险。

3.6 运行代理

result = agent.run("""
Thought: Let's scrape realtor data
Code:
```python
# Scrape realtor data
data = scrape_real_estate_agents(state="NY", city_name="buffalo", num_pages=2)

# Save to CSV
if "error" not in data:
    result = export_to_csv(data)
    print(result)
else:
    print(f"Error: {data['error']}")
""")


- Uses **DeepSeek-R1** to generate reasoning steps.
- Calls **scraping and saving functions dynamically**.

#### 7. Launching the Interface with Gradio
```python
GradioUI(agent).launch()

3.7 使用 Gradio 制作UI

4、结束语

通过集成 Smolagents 和 DeepSeek-R1，我们创建了一个推理代理，这展示了轻量级 AI 代理如何以最小的努力处理复杂的推理任务。

原文链接：Reasoning Agent Using Smolagents and DeepSeek-R1

汇智网翻译整理，转载请标明出处