阿里QwQ-32b或者deepseek-r1模型+browser-use进行数据采集步骤

摘要：import asyncioimport osimport CSVfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_ollama import

bash

复制

bash

复制

OPENAI_API_KEY=DEEPSEEK_API_KEY=《你的api-key》 DASHSCOPE_API_KEY=《你的api-key》ANTHROPIC_API_KEY=GEMINI_API_KEY=《你的api-key》OLLAMA_HOST=https://127.0.0.1:11434 #本地ollama模型

复制

import asyncioimport osimport CSVfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_ollama import ChatOllamafrom pydantic import SecretStrfrom browser_use import Agent, Controllerfrom browser_use.agent.views import ActionResultfrom browser_use import BrowserConfig, Browserfrom openai import OpenAI# dotenvload_dotenvapi_key = os.getenv('DASHSCOPE_API_KEY', '')if not api_key:raise ValueError('DEEPSEEK_API_KEY is not set')# 公共参数commons={"qds":"1100048","qdn":"重庆机场集团有限公司","data_source_type":"招投标","page_index":"首页>采购公告",}controller = Controller@controller.registry.action('保存结果到指定文件')def save_to_file(text: str, file_path: str):with open(file_path, 'w', encoding='utf-8') as f:f.write(text)return ActionResult(extracted_content=f"已保存内容到文件: {file_path}")@controller.registry.action('保存CSV数据')def save_csv_data(data: list, file_path: str):"""保存数据到CSV文件Args:data: 包含字典的列表，每个字典代表一行数据file_path: 保存的文件路径"""if not data:return ActionResult(extracted_content="没有数据可保存")# 确保data是列表格式if not isinstance(data, list):data = [data]# 获取所有可能的字段名fieldnames = setfor item in data:if isinstance(item, dict):fieldnames.update(item.keys)fieldnames = list(fieldnames)with open(file_path, 'w', encoding='utf-8', newline='') as f:writer = csv.DictWriter(f, fieldnames=fieldnames)writer.writeheaderfor item in data:if isinstance(item, dict):writer.writerow(item)return ActionResult(extracted_content=f"已成功保存{len(data)}条记录到CSV文件: {file_path}")@controller.registry.action('追加CSV数据')def append_csv_data(data: list, file_path: str):"""追加数据到CSV文件，如果文件不存在则创建Args:data: 包含字典的列表，每个字典代表一行数据file_path: 保存的文件路径"""if not data:return ActionResult(extracted_content="没有数据可保存")# 增加调试信息print(f"接收到的数据类型: {type(data)}")print(f"数据内容: {data}")# 确保data是列表格式if not isinstance(data, list):try:# 尝试将字符串转换为列表import jsonif isinstance(data, str):data = json.loads(data)except Exception as e:print(f"数据转换失败: {str(e)}")data = [data]# 获取所有可能的字段名fieldnames = setfor item in data:if isinstance(item, dict):fieldnames.update(item.keys)fieldnames = list(fieldnames)# 检查文件是否存在file_exists = os.path.isfile(file_path)try:with open(file_path, 'a', encoding='utf-8', newline='') as f:writer = csv.DictWriter(f, fieldnames=fieldnames)# 如果文件不存在，写入表头if not file_exists:writer.writeheader# 写入数据for item in data:if isinstance(item, dict):writer.writerow(item)return ActionResult(extracted_content=f"已成功追加{len(data)}条记录到CSV文件: {file_path}")except Exception as e:error_msg = f"写入CSV文件时出错: {str(e)}"print(error_msg)return ActionResult(extracted_content=error_msg)async def run_search:# Basic configurationconfig = BrowserConfig(headless=False,disable_security=True,)browser = Browser(config=config)# 简化任务描述，使其更加明确task_description = ("打开 https://zc.cqa.cn/eip/websit/index.do ""获取前3页的采购公告列表中的所有公告，获取每条公告的详情页链接,公告标题，公告时间。""对于每条公告，提取以下字段：'标题'、'发布时间'、'详情链接'。""将该页数据整理成列表格式，列表中每个公告是一个字典，包含上述三个字段。""重要：每采集完一页数据后，立即使用'追加CSV数据'工具将该页数据追加到文件'采购公告-2025-03-07.csv'中，然后再继续采集下一页。""具体步骤如下：""1. 打开网站后，找到采购公告列表""2. 获取当前页的所有公告数据（标题、发布时间、详情链接）""3. 使用'追加CSV数据'工具将当前页的数据追加到CSV文件""4. 点击'下一页'按钮，进入下一页""5. 重复步骤2-4，直到采集完前3页的数据")agent = Agent(browser=browser,task=task_description,llm=ChatOpenAI(base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",model='qwq-32b',streaming=True,api_key=SecretStr(api_key),temperature=0.1, # 降低温度，使输出更确定性max_tokens=4096,model_kwargs={"response_format": {"type": "text"}},),max_actions_per_step=1, # 每步只执行一个动作controller=controller,use_vision=True, # 启用视觉能力，帮助更好地理解页面tool_calling_method="raw",)try:result = await agent.runprint("任务完成，结果：", result)return resultexcept Exception as e:error_msg = f"任务执行出错: {str(e)}"print(error_msg)return error_msgif __name__ == '__main__':asyncio.run(run_search)