30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度如果你在外贸行业特别是像电梯这样的工业品领域一定经历过这样的痛苦每天花几个小时在各种B2B平台、海关数据、企业黄页里大海捞针手动筛选出的客户名单要么不精准要么信息残缺效率极低。更头疼的是好不容易找到一些潜在客户却不知道他们的采购偏好、公司规模、业务匹配度销售邮件发出去石沉大海。今天要讨论的“Codex”并不是一个现成的、开箱即用的“外贸客户开发神器”。从网络热词和搜索趋势来看很多人正在寻找“codex安装教程”、“codex使用”、“codex接入deepseek”这反映出市场对一种新型、高效的AI驱动的信息处理和自动化工具的强烈需求。结合项目标题“以 Elevator 电梯为例直接跑出 85 家目标客户 22 列深度分析”我们可以清晰地判断本文的核心是拆解如何利用类似Codex理念的AI智能体Agent工作流自动化地完成从线索挖掘、数据清洗到深度分析的全过程从而将外贸业务开发从“体力劳动”升级为“脑力决策”。简单说我们不再手动搜索“电梯公司”而是教会AI一套方法让它自动去全网寻找电梯制造商、经销商、维保公司并自动整理出他们的官网、联系方式、主营产品、技术特点、甚至社交媒体动态最终生成一份可直接用于销售触达的深度分析报表。本文将为你还原这个过程的完整技术实现路径。1. 这篇文章真正要解决的问题从“找客户”到“懂客户”的范式转变传统外贸开发信之所以转化率低核心问题在于“广撒网”和“信息不对称”。你并不真正了解目标客户发的都是千篇一律的模板邮件。本文要解决的正是这个痛点。我们通过构建一个智能的“客户开发AI工作流”实现以下目标效率提升将原本需要数天手动完成的客户搜寻和初步筛选工作压缩到几小时甚至几分钟内由AI自动完成。深度洞察超越简单的公司名称和邮箱获取包括业务范围、技术路线、近期动态、供应链位置等在内的22个维度的深度信息为个性化营销提供弹药。流程标准化将优秀的业务员的开发思路固化为可重复、可迭代的AI指令Prompt和流程降低对个人经验的依赖。持续监控一旦工作流建立可以定期运行自动发现新客户或跟踪老客户的动态变化。以“电梯”行业为例我们最终要得到的不是一份简单的名单而是一张包含如下信息的智能表格基础信息公司名、官网、国家、城市。业务信息主要产品如乘客电梯、载货电梯、自动扶梯、品牌、是否OEM/ODM。技术信息电梯速度、载重、驱动方式曳引/液压、是否具备物联网IoT功能。市场信息目标市场本地/国际、认证情况CE, ISO。动态信息近期新闻、招聘信息、社交媒体活跃度。联系信息关键联系人邮箱、电话、LinkedIn资料。接下来我们将从概念、工具选择、实战流程到代码实现完整拆解如何构建这样一个系统。2. 核心概念与工具选型什么是“Codex”式工作流首先需要澄清概念。根据有限的网络信息“Codex app”被描述为一个支持并行处理线程、内置工作树和Git功能的桌面体验工具。结合“codex接入deepseek”等热词我们可以推断在当前语境下“Codex”更可能指的是一个以DeepSeek等大型语言模型LLM为核心能够处理复杂、多步骤任务如研究、分析、汇总的AI智能体Agent或自动化工作流平台。它不同于简单的ChatGPT对话而是强调任务分解将“找电梯客户”这个大任务拆解为“搜索-筛选-提取-分析-格式化”等多个子任务。工具调用AI可以自主调用搜索引擎、访问网页、解析文档、操作表格等工具。持续迭代基于上一步的结果规划下一步的行动直到达成目标。结果结构化最终输出是规整的JSON或CSV而非杂乱文本。为了实现这样的工作流我们需要以下几类工具工具类型推荐选项作用核心AI模型DeepSeek API, OpenAI GPT-4, Claude 3负责理解任务、规划步骤、解析信息、生成总结。自动化框架LangChain, AutoGen, CrewAI提供构建多智能体Multi-Agent工作流的框架方便管理任务流和工具调用。搜索与爬取Serper API, Google Search API, BeautifulSoup, Scrapy获取公开网络信息。注意必须遵守robots.txt合法合规使用。数据处理Pandas, Jupyter Notebook清洗、分析、保存最终数据。环境与部署Python 3.9, Jupyter Lab, Docker (可选)运行代码的环境。本教程将选择最具性价比和可行性的技术栈Python LangChain DeepSeek API Serper API。这套组合灵活、可控且适合大多数开发者。3. 环境准备与前置条件在开始编写代码前请确保你的开发环境已就绪。3.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以macOS/Linux为例。Python版本 3.9 或以上。推荐使用 Miniconda 或 Pyenv 管理多版本Python。包管理工具pip或conda。代码编辑器VS Code (推荐) 或 PyCharm。3.2 关键API申请本工作流需要调用外部服务的API请提前申请。DeepSeek API访问 DeepSeek 官方平台注册账号。在控制台创建API Key并记录备用。注意保管不要泄露。Serper API(用于搜索)访问 Serper.dev 注册。其Google搜索API免费额度充足非常适合开发阶段。获取你的API Key。3.3 项目初始化与依赖安装打开终端创建一个新的项目目录并安装依赖。# 创建项目目录 mkdir ai-sales-leads cd ai-sales-leads # 创建虚拟环境 (推荐) python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain langchain-community langchain-core pip install beautifulsoup4 requests pandas jupyter pip install langchain-openai # 虽然叫openai但通常兼容DeepSeek等遵循OpenAI格式的API # 安装用于搜索的库我们将使用requests和自定义工具也可以使用langchain的serper工具 pip install google-search-results # 备用方案Serper有官方LangChain集成创建环境变量文件.env用于安全存储API密钥。切勿将此文件提交到Git仓库# .env 文件内容 DEEPSEEK_API_KEYyour_deepseek_api_key_here SERPER_API_KEYyour_serper_api_key_here4. 核心流程拆解四步构建智能客户开发工作流整个工作流可以抽象为四个核心阶段如下图所示概念图[启动] - [阶段一智能搜索与发现] - [阶段二精准抓取与解析] - [阶段三多维度信息提取] - [阶段四结构化分析与输出] - [结果报表]4.1 阶段一智能搜索与发现目标根据“电梯”这个关键词让AI生成更具体、更多元的搜索查询扩大线索来源。传统做法业务员手动在Google搜索“elevator manufacturer”。AI做法让LLM基于行业知识生成一批长尾搜索词例如“passenger elevator OEM company Europe”“escalator maintenance service provider in Middle East”“home elevator supplier Thailand”“elevator IoT solution startup”技术实现编写一个Prompt让DeepSeek模型生成搜索词列表然后通过Serper API并发执行这些搜索去重后得到初步的官网链接池。4.2 阶段二精准抓取与解析目标从收集到的官网链接中抓取关键页面的HTML内容如首页、About Us、Products页面。挑战网站结构各异反爬策略不同。策略使用requests库模拟浏览器访问设置合理的User-Agent和延迟遵守robots.txt。使用BeautifulSoup解析HTML提取主内容区域的文本。优先抓取main,article标签或包含大量文本的div。对抓取失败或内容过少的网站进行过滤。4.3 阶段三多维度信息提取目标从抓取到的网页文本中提取出我们关心的22个维度的信息。这是AI的核心价值所在。我们设计一个详细的“信息提取Prompt”指导模型像一名经验丰富的市场分析师一样阅读网页并按要求填写信息。例如Prompt会这样要求模型请从以下公司官网文本中提取以下信息。如果信息不存在请填写“N/A”。 公司名称 国家/地区 城市 主要产品分类列出 是否提供OEM/ODM服务 核心技术特点如曳引机品牌、物联网功能 ...技术实现将清洗后的网页文本分批送入DeepSeek模型调用其函数调用Function Calling或结构化输出JSON Mode能力直接返回格式化的JSON数据。4.4 阶段四结构化分析与输出目标将提取出的所有公司信息汇总、清洗、去重并保存为结构化的CSV文件。操作将每个公司返回的JSON数据合并成一个列表。使用pandas库进行数据清洗处理空值、统一格式。根据“国家”、“产品类型”等字段进行简单的统计分析。导出最终的elevator_companies_analysis.csv文件。5. 完整示例与代码实现下面我们分模块实现上述工作流。请将以下代码保存到不同的Python文件中。5.1 配置与工具模块 (config.py和tools.py)首先创建配置文件加载环境变量。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) SERPER_API_KEY os.getenv(SERPER_API_KEY) DEEPSEEK_API_BASE https://api.deepseek.com # 请根据DeepSeek官方文档确认最新地址 # 搜索配置 SEARCH_DELAY 1 # 每次搜索之间的延迟秒避免触发风控 USER_AGENT Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36然后创建基础的网络工具。# tools.py import requests import time from bs4 import BeautifulSoup from config import USER_AGENT, SEARCH_DELAY import json def search_serper(query: str, api_key: str, num_results: int 10): 使用Serper API进行Google搜索 url https://google.serper.dev/search payload json.dumps({ q: query, num: num_results }) headers { X-API-KEY: api_key, Content-Type: application/json } try: response requests.request(POST, url, headersheaders, datapayload) response.raise_for_status() data response.json() # 提取有机搜索结果的链接 links [] if organic in data: for result in data[organic]: link result.get(link) if link and link.startswith(http): links.append(link) return links[:num_results] # 返回指定数量的链接 except Exception as e: print(f搜索出错 {query}: {e}) return [] finally: time.sleep(SEARCH_DELAY) # 礼貌延迟 def fetch_webpage_text(url: str): 抓取网页并提取主要文本内容 headers {User-Agent: USER_AGENT} try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() soup BeautifulSoup(resp.content, html.parser) # 移除脚本、样式等标签 for script in soup([script, style, nav, footer, header]): script.decompose() # 尝试获取主内容区域如果没有则获取body main_content soup.find(main) or soup.find(article) or soup.find(body) if main_content: text main_content.get_text(separator , stripTrue) # 简化空白字符 text .join(text.split()) return text[:15000] # 限制文本长度避免超出模型上下文 else: return except Exception as e: print(f抓取网页失败 {url}: {e}) return 5.2 AI智能体工作流主模块 (main_workflow.py)这是核心逻辑所在我们使用LangChain来编排任务。# main_workflow.py import json from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough from config import DEEPSEEK_API_KEY, DEEPSEEK_API_BASE, SERPER_API_KEY from tools import search_serper, fetch_webpage_text import pandas as pd import time # 1. 初始化DeepSeek模型通过LangChain的OpenAI兼容接口 llm ChatOpenAI( modeldeepseek-chat, # 根据DeepSeek最新模型名调整 openai_api_keyDEEPSEEK_API_KEY, openai_api_baseDEEPSEEK_API_BASE, temperature0.1, # 低温度保证输出稳定性 ) # 2. 定义Prompt模板生成多样化搜索词 search_keywords_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深外贸市场研究员。请针对{industry}行业生成20个用于寻找潜在客户制造商、供应商、服务商的英文长尾Google搜索关键词。关键词应覆盖不同产品类型、服务模式、地域市场。以JSON列表格式返回例如[\keyword1\, \keyword2\]), (user, 行业是{industry}) ]) # 3. 定义信息提取Prompt模板 info_extraction_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的商业情报分析员。请仔细阅读以下公司官网的文本内容并提取出指定的信息。 请严格按照给定的JSON格式输出只输出JSON对象不要有任何额外解释。 如果某项信息无法从文本中推断请将对应字段的值设为“N/A”。 ), (user, 需要提取的信息字段如下 {{ company_name: 公司全称, website: 官网网址, country: 总部所在国家, city: 城市, main_products: [产品1, 产品2], // 数组列出主要电梯/扶梯产品类型 business_model: OEM/ODM/自有品牌/经销商, // 选择最符合的一项或填写N/A key_technologies: [技术1, 技术2], // 如物联网、节能、高速梯等 target_markets: [市场1, 市场2], // 提及的国际或本地市场 certifications: [CE, ISO9001], // 提到的认证 company_profile_summary: 一段50字以内的公司简介摘要 }} 官网文本内容 {page_text} ) ]) # 4. 构建工作流链条 def generate_search_queries(industry: str): 生成搜索关键词链 chain search_keywords_prompt | llm | StrOutputParser() result chain.invoke({industry: industry}) try: # 尝试解析返回的JSON queries json.loads(result) return queries if isinstance(queries, list) else [] except json.JSONDecodeError: # 如果模型返回的不是标准JSON尝试提取引号内的内容 import re queries re.findall(r\([^\])\, result) return queries[:20] def extract_company_info(page_text: str, url: str): 提取单公司信息链 if len(page_text) 100: # 文本太短跳过 return None chain info_extraction_prompt | llm | StrOutputParser() try: info_json_str chain.invoke({page_text: page_text}) info_dict json.loads(info_json_str) info_dict[website] url # 确保网址被记录 return info_dict except Exception as e: print(f信息提取失败: {e}) return None def main(industryelevator): print(f开始为 {industry} 行业生成潜在客户列表...) # 步骤1: 生成搜索词 print(1. 生成多样化搜索关键词...) search_queries generate_search_queries(industry) print(f生成 {len(search_queries)} 个搜索词: {search_queries[:5]}...) all_links set() # 步骤2: 并发执行搜索这里简化为循环生产环境可用线程池 print(2. 执行网络搜索收集公司官网链接...) for query in search_queries: links search_serper(query, SERPER_API_KEY, num_results5) all_links.update(links) print(f 搜索 {query} 找到 {len(links)} 个链接) if len(all_links) 50: # 限制总链接数防止过量请求 break print(f去重后共收集到 {len(all_links)} 个潜在官网链接。) # 步骤3: 抓取网页并提取信息 print(3. 抓取网页并提取结构化信息...) companies_data [] for idx, url in enumerate(list(all_links)[:30]): # 限制抓取数量避免超时 print(f 处理 ({idx1}/{min(30, len(all_links))}): {url}) text fetch_webpage_text(url) if text: company_info extract_company_info(text, url) if company_info: companies_data.append(company_info) time.sleep(0.5) # 抓取间隔 # 步骤4: 保存结果 print(4. 保存结果到CSV文件...) if companies_data: df pd.DataFrame(companies_data) # 数据清洗处理空值 df.fillna(N/A, inplaceTrue) filename f{industry}_companies_analysis.csv df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f成功已保存 {len(df)} 家公司信息到 {filename}。) print(\n数据预览) print(df[[company_name, country, main_products]].head()) else: print(未提取到有效的公司信息。) if __name__ __main__: # 运行主流程可以替换行业关键词 main(industryelevator and escalator)5.3 增强版并行处理与错误处理 (advanced_workflow.py)为了提高效率我们可以引入并行处理。这里使用concurrent.futures线程池。# advanced_workflow.py (部分关键函数改写) from concurrent.futures import ThreadPoolExecutor, as_completed from main_workflow import fetch_webpage_text, extract_company_info import tqdm # 进度条库需安装pip install tqdm def process_single_url(url): 处理单个URL的任务函数用于并行化 text fetch_webpage_text(url) if text and len(text) 100: info extract_company_info(text, url) return info return None def parallel_info_extraction(urls, max_workers5): 并行抓取和提取信息 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_url {executor.submit(process_single_url, url): url for url in urls} # 使用tqdm显示进度 for future in tqdm.tqdm(as_completed(future_to_url), totallen(urls), descProcessing URLs): url future_to_url[future] try: data future.result() if data: results.append(data) except Exception as exc: print(f{url} generated an exception: {exc}) return results # 在主函数中替换步骤3的循环 # companies_data parallel_info_extraction(list(all_links)[:30])6. 运行结果与效果验证6.1 如何运行确保你的.env文件已正确配置API密钥。在项目根目录下运行主程序python main_workflow.py6.2 预期输出与日志程序运行后你将在控制台看到类似以下的日志清晰地展示了工作流的每个阶段开始为 elevator and escalator 行业生成潜在客户列表... 1. 生成多样化搜索关键词... 生成 15 个搜索词: [elevator manufacturer Europe, escalator supplier China, OEM elevator company, ...]... 2. 执行网络搜索收集公司官网链接... 搜索 elevator manufacturer Europe 找到 8 个链接 搜索 escalator supplier China 找到 10 个链接 ... 去重后共收集到 78 个潜在官网链接。 3. 抓取网页并提取结构化信息... 处理 (1/30): https://www.example-elevator.com 处理 (2/30): https://www.anotherelevator.co.uk ... 4. 保存结果到CSV文件... 成功已保存 22 家公司信息到 elevator_and_escalator_companies_analysis.csv。 数据预览 company_name country main_products 0 Example Elevator Co. Germany [Passenger Elevator, Freight Elevator] 1 UK Lift Solutions Ltd. UK [Escalator, Moving Walkway] ...6.3 结果文件分析生成的CSV文件将包含我们在Prompt中定义的所有字段。你可以用Excel或Numbers打开它。一个理想的结果应该包含公司名称准确的全称。地理位置国家和城市。产品阵列以列表形式清晰列出。业务模式明确标注OEM/ODM等。技术关键词提取出的技术特性。简介摘要AI生成的凝练描述。这份报表就是你的“85家目标客户22列深度分析”的雏形。数字“85”和“22”是目标实际数量取决于搜索词的广度、网页可访问性以及信息提取的准确性。6.4 验证成功与否成功标志CSV文件被创建且包含多条记录关键字段如公司名、国家、产品非空率较高。初级排查如果结果为空或很少请按顺序检查API密钥确认.env文件配置正确且API服务无欠费。网络连接确认程序可以访问外网用于搜索和抓取。搜索词质量查看search_queries变量输出的搜索词是否具体、相关。可以手动调整Prompt让AI生成更精准的词。网页抓取检查fetch_webpage_text函数是否因网站反爬而返回空文本。可尝试增加请求头、使用Session或代理需合法合规。信息提取检查extract_company_info函数中模型返回的原始JSON字符串看是否解析失败。可能是模型未按格式输出需要优化Prompt。7. 常见问题与排查思路在实际运行中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行报错ModuleNotFoundError依赖包未安装或虚拟环境未激活。在终端执行pip list检查langchain,requests等包是否存在。激活虚拟环境并运行pip install -r requirements.txt(需先创建此文件)。程序卡住或无响应网络请求超时或某个网站响应慢。查看日志卡在哪一步。为requests.get设置timeout参数。在fetch_webpage_text函数中增加超时和重试机制。使用ThreadPoolExecutor时设置timeout。搜索返回结果为空Serper API 密钥无效或额度用尽搜索词过于宽泛或冷门。手动在浏览器用相同关键词搜索确认是否有结果。检查Serper控制台使用量和报错。更换搜索词在Prompt中要求生成更具体、包含地域/产品型号的词。确认API密钥正确。抓取的网页文本杂乱或为空网站使用JavaScript渲染或设置了强反爬。打印抓取的URL和响应状态码。用浏览器开发者工具查看页面真实结构。对于JS渲染站可考虑使用Selenium或Playwright。务必遵守robots.txt并显著降低请求频率。AI提取的信息不准确或格式错误Prompt指令不够清晰网页文本噪声太多。打印出送入模型的原始文本前500字符检查质量。查看模型返回的原始字符串。优化Prompt提供更明确的例子和格式要求。在提取前对网页文本进行更严格的清洗如移除导航栏、版权文字。最终CSV文件字段混乱模型返回的JSON结构不一致或json.loads失败。在extract_company_info函数中捕获异常并打印出导致错误的原始文本和模型输出。使用LangChain的StructuredOutputParser或PydanticOutputParser来强制结构化输出比单纯依赖JSON模式更稳定。达到“85家”目标很困难初始搜索范围窄或去重后有效链接少。分析all_links集合的数量和质量。增加搜索词生成数量如从20到50。让AI生成更细分的长尾词如“hospital elevator supplier”、“residential lift manufacturer Italy”。在搜索后引入简单的域名分类和过滤。8. 最佳实践与工程建议要将这个脚本转化为一个稳定、可用的生产级工具你需要考虑以下几点合规与伦理第一遵守robots.txt在抓取任何网站前检查其robots.txt文件尊重Disallow规则。设置礼貌延迟在请求间添加随机延迟如time.sleep(random.uniform(1, 3))避免对目标网站造成压力。明确数据用途此方法仅用于收集公开的商业信息。切勿用于爬取个人隐私数据、受版权保护内容或任何非法用途。提升数据质量多轮验证对AI提取的关键信息如邮箱、电话可通过正则表达式进行二次验证。信息融合对同一公司的多个信息来源如官网、LinkedIn、行业目录进行信息融合和去重补充单一来源的不足。人工审核闭环将AI生成的列表导入CRM或表格后应设计简单的人工审核标记流程如“有效/无效”这些反馈数据可用于微调AI的Prompt形成闭环优化。系统健壮性错误重试与熔断为网络请求和API调用实现重试逻辑如tenacity库和熔断机制避免因个别失败导致整个任务崩溃。状态持久化使用SQLite或小文件记录已处理URL的状态成功/失败支持任务中断后续跑避免重复工作。日志与监控使用Python的logging模块记录详细日志便于排查问题。监控API调用成本和成功率。扩展性设计模块化将搜索、抓取、解析、保存等模块彻底解耦方便单独替换或升级例如将Serper换成其他搜索API。支持多行业将行业关键词和对应的提取字段模板化通过配置文件管理快速适配到“光伏面板”、“医疗器械”等其他行业。集成化将最终输出的CSV自动同步到Google Sheets或通过Zapier/Make.com触发后续的邮件营销流程实现全自动化。通过以上步骤你构建的不仅仅是一个脚本而是一个可扩展、可维护的“AI外贸业务开发助手”。它将你从繁琐的信息搜集工作中解放出来让你能更专注于需要人类判断和情感连接的高价值环节——客户沟通和关系维护。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度如果你在外贸行业特别是像电梯这样的工业品领域一定经历过这样的痛苦每天花几个小时在各种B2B平台、海关数据、企业黄页里大海捞针手动筛选出的客户名单要么不精准要么信息残缺效率极低。更头疼的是好不容易找到一些潜在客户却不知道他们的采购偏好、公司规模、业务匹配度销售邮件发出去石沉大海。今天要讨论的“Codex”并不是一个现成的、开箱即用的“外贸客户开发神器”。从网络热词和搜索趋势来看很多人正在寻找“codex安装教程”、“codex使用”、“codex接入deepseek”这反映出市场对一种新型、高效的AI驱动的信息处理和自动化工具的强烈需求。结合项目标题“以 Elevator 电梯为例直接跑出 85 家目标客户 22 列深度分析”我们可以清晰地判断本文的核心是拆解如何利用类似Codex理念的AI智能体Agent工作流自动化地完成从线索挖掘、数据清洗到深度分析的全过程从而将外贸业务开发从“体力劳动”升级为“脑力决策”。简单说我们不再手动搜索“电梯公司”而是教会AI一套方法让它自动去全网寻找电梯制造商、经销商、维保公司并自动整理出他们的官网、联系方式、主营产品、技术特点、甚至社交媒体动态最终生成一份可直接用于销售触达的深度分析报表。本文将为你还原这个过程的完整技术实现路径。1. 这篇文章真正要解决的问题从“找客户”到“懂客户”的范式转变传统外贸开发信之所以转化率低核心问题在于“广撒网”和“信息不对称”。你并不真正了解目标客户发的都是千篇一律的模板邮件。本文要解决的正是这个痛点。我们通过构建一个智能的“客户开发AI工作流”实现以下目标效率提升将原本需要数天手动完成的客户搜寻和初步筛选工作压缩到几小时甚至几分钟内由AI自动完成。深度洞察超越简单的公司名称和邮箱获取包括业务范围、技术路线、近期动态、供应链位置等在内的22个维度的深度信息为个性化营销提供弹药。流程标准化将优秀的业务员的开发思路固化为可重复、可迭代的AI指令Prompt和流程降低对个人经验的依赖。持续监控一旦工作流建立可以定期运行自动发现新客户或跟踪老客户的动态变化。以“电梯”行业为例我们最终要得到的不是一份简单的名单而是一张包含如下信息的智能表格基础信息公司名、官网、国家、城市。业务信息主要产品如乘客电梯、载货电梯、自动扶梯、品牌、是否OEM/ODM。技术信息电梯速度、载重、驱动方式曳引/液压、是否具备物联网IoT功能。市场信息目标市场本地/国际、认证情况CE, ISO。动态信息近期新闻、招聘信息、社交媒体活跃度。联系信息关键联系人邮箱、电话、LinkedIn资料。接下来我们将从概念、工具选择、实战流程到代码实现完整拆解如何构建这样一个系统。2. 核心概念与工具选型什么是“Codex”式工作流首先需要澄清概念。根据有限的网络信息“Codex app”被描述为一个支持并行处理线程、内置工作树和Git功能的桌面体验工具。结合“codex接入deepseek”等热词我们可以推断在当前语境下“Codex”更可能指的是一个以DeepSeek等大型语言模型LLM为核心能够处理复杂、多步骤任务如研究、分析、汇总的AI智能体Agent或自动化工作流平台。它不同于简单的ChatGPT对话而是强调任务分解将“找电梯客户”这个大任务拆解为“搜索-筛选-提取-分析-格式化”等多个子任务。工具调用AI可以自主调用搜索引擎、访问网页、解析文档、操作表格等工具。持续迭代基于上一步的结果规划下一步的行动直到达成目标。结果结构化最终输出是规整的JSON或CSV而非杂乱文本。为了实现这样的工作流我们需要以下几类工具工具类型推荐选项作用核心AI模型DeepSeek API, OpenAI GPT-4, Claude 3负责理解任务、规划步骤、解析信息、生成总结。自动化框架LangChain, AutoGen, CrewAI提供构建多智能体Multi-Agent工作流的框架方便管理任务流和工具调用。搜索与爬取Serper API, Google Search API, BeautifulSoup, Scrapy获取公开网络信息。注意必须遵守robots.txt合法合规使用。数据处理Pandas, Jupyter Notebook清洗、分析、保存最终数据。环境与部署Python 3.9, Jupyter Lab, Docker (可选)运行代码的环境。本教程将选择最具性价比和可行性的技术栈Python LangChain DeepSeek API Serper API。这套组合灵活、可控且适合大多数开发者。3. 环境准备与前置条件在开始编写代码前请确保你的开发环境已就绪。3.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以macOS/Linux为例。Python版本 3.9 或以上。推荐使用 Miniconda 或 Pyenv 管理多版本Python。包管理工具pip或conda。代码编辑器VS Code (推荐) 或 PyCharm。3.2 关键API申请本工作流需要调用外部服务的API请提前申请。DeepSeek API访问 DeepSeek 官方平台注册账号。在控制台创建API Key并记录备用。注意保管不要泄露。Serper API(用于搜索)访问 Serper.dev 注册。其Google搜索API免费额度充足非常适合开发阶段。获取你的API Key。3.3 项目初始化与依赖安装打开终端创建一个新的项目目录并安装依赖。# 创建项目目录 mkdir ai-sales-leads cd ai-sales-leads # 创建虚拟环境 (推荐) python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain langchain-community langchain-core pip install beautifulsoup4 requests pandas jupyter pip install langchain-openai # 虽然叫openai但通常兼容DeepSeek等遵循OpenAI格式的API # 安装用于搜索的库我们将使用requests和自定义工具也可以使用langchain的serper工具 pip install google-search-results # 备用方案Serper有官方LangChain集成创建环境变量文件.env用于安全存储API密钥。切勿将此文件提交到Git仓库# .env 文件内容 DEEPSEEK_API_KEYyour_deepseek_api_key_here SERPER_API_KEYyour_serper_api_key_here4. 核心流程拆解四步构建智能客户开发工作流整个工作流可以抽象为四个核心阶段如下图所示概念图[启动] - [阶段一智能搜索与发现] - [阶段二精准抓取与解析] - [阶段三多维度信息提取] - [阶段四结构化分析与输出] - [结果报表]4.1 阶段一智能搜索与发现目标根据“电梯”这个关键词让AI生成更具体、更多元的搜索查询扩大线索来源。传统做法业务员手动在Google搜索“elevator manufacturer”。AI做法让LLM基于行业知识生成一批长尾搜索词例如“passenger elevator OEM company Europe”“escalator maintenance service provider in Middle East”“home elevator supplier Thailand”“elevator IoT solution startup”技术实现编写一个Prompt让DeepSeek模型生成搜索词列表然后通过Serper API并发执行这些搜索去重后得到初步的官网链接池。4.2 阶段二精准抓取与解析目标从收集到的官网链接中抓取关键页面的HTML内容如首页、About Us、Products页面。挑战网站结构各异反爬策略不同。策略使用requests库模拟浏览器访问设置合理的User-Agent和延迟遵守robots.txt。使用BeautifulSoup解析HTML提取主内容区域的文本。优先抓取main,article标签或包含大量文本的div。对抓取失败或内容过少的网站进行过滤。4.3 阶段三多维度信息提取目标从抓取到的网页文本中提取出我们关心的22个维度的信息。这是AI的核心价值所在。我们设计一个详细的“信息提取Prompt”指导模型像一名经验丰富的市场分析师一样阅读网页并按要求填写信息。例如Prompt会这样要求模型请从以下公司官网文本中提取以下信息。如果信息不存在请填写“N/A”。 公司名称 国家/地区 城市 主要产品分类列出 是否提供OEM/ODM服务 核心技术特点如曳引机品牌、物联网功能 ...技术实现将清洗后的网页文本分批送入DeepSeek模型调用其函数调用Function Calling或结构化输出JSON Mode能力直接返回格式化的JSON数据。4.4 阶段四结构化分析与输出目标将提取出的所有公司信息汇总、清洗、去重并保存为结构化的CSV文件。操作将每个公司返回的JSON数据合并成一个列表。使用pandas库进行数据清洗处理空值、统一格式。根据“国家”、“产品类型”等字段进行简单的统计分析。导出最终的elevator_companies_analysis.csv文件。5. 完整示例与代码实现下面我们分模块实现上述工作流。请将以下代码保存到不同的Python文件中。5.1 配置与工具模块 (config.py和tools.py)首先创建配置文件加载环境变量。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) SERPER_API_KEY os.getenv(SERPER_API_KEY) DEEPSEEK_API_BASE https://api.deepseek.com # 请根据DeepSeek官方文档确认最新地址 # 搜索配置 SEARCH_DELAY 1 # 每次搜索之间的延迟秒避免触发风控 USER_AGENT Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36然后创建基础的网络工具。# tools.py import requests import time from bs4 import BeautifulSoup from config import USER_AGENT, SEARCH_DELAY import json def search_serper(query: str, api_key: str, num_results: int 10): 使用Serper API进行Google搜索 url https://google.serper.dev/search payload json.dumps({ q: query, num: num_results }) headers { X-API-KEY: api_key, Content-Type: application/json } try: response requests.request(POST, url, headersheaders, datapayload) response.raise_for_status() data response.json() # 提取有机搜索结果的链接 links [] if organic in data: for result in data[organic]: link result.get(link) if link and link.startswith(http): links.append(link) return links[:num_results] # 返回指定数量的链接 except Exception as e: print(f搜索出错 {query}: {e}) return [] finally: time.sleep(SEARCH_DELAY) # 礼貌延迟 def fetch_webpage_text(url: str): 抓取网页并提取主要文本内容 headers {User-Agent: USER_AGENT} try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() soup BeautifulSoup(resp.content, html.parser) # 移除脚本、样式等标签 for script in soup([script, style, nav, footer, header]): script.decompose() # 尝试获取主内容区域如果没有则获取body main_content soup.find(main) or soup.find(article) or soup.find(body) if main_content: text main_content.get_text(separator , stripTrue) # 简化空白字符 text .join(text.split()) return text[:15000] # 限制文本长度避免超出模型上下文 else: return except Exception as e: print(f抓取网页失败 {url}: {e}) return 5.2 AI智能体工作流主模块 (main_workflow.py)这是核心逻辑所在我们使用LangChain来编排任务。# main_workflow.py import json from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough from config import DEEPSEEK_API_KEY, DEEPSEEK_API_BASE, SERPER_API_KEY from tools import search_serper, fetch_webpage_text import pandas as pd import time # 1. 初始化DeepSeek模型通过LangChain的OpenAI兼容接口 llm ChatOpenAI( modeldeepseek-chat, # 根据DeepSeek最新模型名调整 openai_api_keyDEEPSEEK_API_KEY, openai_api_baseDEEPSEEK_API_BASE, temperature0.1, # 低温度保证输出稳定性 ) # 2. 定义Prompt模板生成多样化搜索词 search_keywords_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深外贸市场研究员。请针对{industry}行业生成20个用于寻找潜在客户制造商、供应商、服务商的英文长尾Google搜索关键词。关键词应覆盖不同产品类型、服务模式、地域市场。以JSON列表格式返回例如[\keyword1\, \keyword2\]), (user, 行业是{industry}) ]) # 3. 定义信息提取Prompt模板 info_extraction_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的商业情报分析员。请仔细阅读以下公司官网的文本内容并提取出指定的信息。 请严格按照给定的JSON格式输出只输出JSON对象不要有任何额外解释。 如果某项信息无法从文本中推断请将对应字段的值设为“N/A”。 ), (user, 需要提取的信息字段如下 {{ company_name: 公司全称, website: 官网网址, country: 总部所在国家, city: 城市, main_products: [产品1, 产品2], // 数组列出主要电梯/扶梯产品类型 business_model: OEM/ODM/自有品牌/经销商, // 选择最符合的一项或填写N/A key_technologies: [技术1, 技术2], // 如物联网、节能、高速梯等 target_markets: [市场1, 市场2], // 提及的国际或本地市场 certifications: [CE, ISO9001], // 提到的认证 company_profile_summary: 一段50字以内的公司简介摘要 }} 官网文本内容 {page_text} ) ]) # 4. 构建工作流链条 def generate_search_queries(industry: str): 生成搜索关键词链 chain search_keywords_prompt | llm | StrOutputParser() result chain.invoke({industry: industry}) try: # 尝试解析返回的JSON queries json.loads(result) return queries if isinstance(queries, list) else [] except json.JSONDecodeError: # 如果模型返回的不是标准JSON尝试提取引号内的内容 import re queries re.findall(r\([^\])\, result) return queries[:20] def extract_company_info(page_text: str, url: str): 提取单公司信息链 if len(page_text) 100: # 文本太短跳过 return None chain info_extraction_prompt | llm | StrOutputParser() try: info_json_str chain.invoke({page_text: page_text}) info_dict json.loads(info_json_str) info_dict[website] url # 确保网址被记录 return info_dict except Exception as e: print(f信息提取失败: {e}) return None def main(industryelevator): print(f开始为 {industry} 行业生成潜在客户列表...) # 步骤1: 生成搜索词 print(1. 生成多样化搜索关键词...) search_queries generate_search_queries(industry) print(f生成 {len(search_queries)} 个搜索词: {search_queries[:5]}...) all_links set() # 步骤2: 并发执行搜索这里简化为循环生产环境可用线程池 print(2. 执行网络搜索收集公司官网链接...) for query in search_queries: links search_serper(query, SERPER_API_KEY, num_results5) all_links.update(links) print(f 搜索 {query} 找到 {len(links)} 个链接) if len(all_links) 50: # 限制总链接数防止过量请求 break print(f去重后共收集到 {len(all_links)} 个潜在官网链接。) # 步骤3: 抓取网页并提取信息 print(3. 抓取网页并提取结构化信息...) companies_data [] for idx, url in enumerate(list(all_links)[:30]): # 限制抓取数量避免超时 print(f 处理 ({idx1}/{min(30, len(all_links))}): {url}) text fetch_webpage_text(url) if text: company_info extract_company_info(text, url) if company_info: companies_data.append(company_info) time.sleep(0.5) # 抓取间隔 # 步骤4: 保存结果 print(4. 保存结果到CSV文件...) if companies_data: df pd.DataFrame(companies_data) # 数据清洗处理空值 df.fillna(N/A, inplaceTrue) filename f{industry}_companies_analysis.csv df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f成功已保存 {len(df)} 家公司信息到 {filename}。) print(\n数据预览) print(df[[company_name, country, main_products]].head()) else: print(未提取到有效的公司信息。) if __name__ __main__: # 运行主流程可以替换行业关键词 main(industryelevator and escalator)5.3 增强版并行处理与错误处理 (advanced_workflow.py)为了提高效率我们可以引入并行处理。这里使用concurrent.futures线程池。# advanced_workflow.py (部分关键函数改写) from concurrent.futures import ThreadPoolExecutor, as_completed from main_workflow import fetch_webpage_text, extract_company_info import tqdm # 进度条库需安装pip install tqdm def process_single_url(url): 处理单个URL的任务函数用于并行化 text fetch_webpage_text(url) if text and len(text) 100: info extract_company_info(text, url) return info return None def parallel_info_extraction(urls, max_workers5): 并行抓取和提取信息 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_url {executor.submit(process_single_url, url): url for url in urls} # 使用tqdm显示进度 for future in tqdm.tqdm(as_completed(future_to_url), totallen(urls), descProcessing URLs): url future_to_url[future] try: data future.result() if data: results.append(data) except Exception as exc: print(f{url} generated an exception: {exc}) return results # 在主函数中替换步骤3的循环 # companies_data parallel_info_extraction(list(all_links)[:30])6. 运行结果与效果验证6.1 如何运行确保你的.env文件已正确配置API密钥。在项目根目录下运行主程序python main_workflow.py6.2 预期输出与日志程序运行后你将在控制台看到类似以下的日志清晰地展示了工作流的每个阶段开始为 elevator and escalator 行业生成潜在客户列表... 1. 生成多样化搜索关键词... 生成 15 个搜索词: [elevator manufacturer Europe, escalator supplier China, OEM elevator company, ...]... 2. 执行网络搜索收集公司官网链接... 搜索 elevator manufacturer Europe 找到 8 个链接 搜索 escalator supplier China 找到 10 个链接 ... 去重后共收集到 78 个潜在官网链接。 3. 抓取网页并提取结构化信息... 处理 (1/30): https://www.example-elevator.com 处理 (2/30): https://www.anotherelevator.co.uk ... 4. 保存结果到CSV文件... 成功已保存 22 家公司信息到 elevator_and_escalator_companies_analysis.csv。 数据预览 company_name country main_products 0 Example Elevator Co. Germany [Passenger Elevator, Freight Elevator] 1 UK Lift Solutions Ltd. UK [Escalator, Moving Walkway] ...6.3 结果文件分析生成的CSV文件将包含我们在Prompt中定义的所有字段。你可以用Excel或Numbers打开它。一个理想的结果应该包含公司名称准确的全称。地理位置国家和城市。产品阵列以列表形式清晰列出。业务模式明确标注OEM/ODM等。技术关键词提取出的技术特性。简介摘要AI生成的凝练描述。这份报表就是你的“85家目标客户22列深度分析”的雏形。数字“85”和“22”是目标实际数量取决于搜索词的广度、网页可访问性以及信息提取的准确性。6.4 验证成功与否成功标志CSV文件被创建且包含多条记录关键字段如公司名、国家、产品非空率较高。初级排查如果结果为空或很少请按顺序检查API密钥确认.env文件配置正确且API服务无欠费。网络连接确认程序可以访问外网用于搜索和抓取。搜索词质量查看search_queries变量输出的搜索词是否具体、相关。可以手动调整Prompt让AI生成更精准的词。网页抓取检查fetch_webpage_text函数是否因网站反爬而返回空文本。可尝试增加请求头、使用Session或代理需合法合规。信息提取检查extract_company_info函数中模型返回的原始JSON字符串看是否解析失败。可能是模型未按格式输出需要优化Prompt。7. 常见问题与排查思路在实际运行中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行报错ModuleNotFoundError依赖包未安装或虚拟环境未激活。在终端执行pip list检查langchain,requests等包是否存在。激活虚拟环境并运行pip install -r requirements.txt(需先创建此文件)。程序卡住或无响应网络请求超时或某个网站响应慢。查看日志卡在哪一步。为requests.get设置timeout参数。在fetch_webpage_text函数中增加超时和重试机制。使用ThreadPoolExecutor时设置timeout。搜索返回结果为空Serper API 密钥无效或额度用尽搜索词过于宽泛或冷门。手动在浏览器用相同关键词搜索确认是否有结果。检查Serper控制台使用量和报错。更换搜索词在Prompt中要求生成更具体、包含地域/产品型号的词。确认API密钥正确。抓取的网页文本杂乱或为空网站使用JavaScript渲染或设置了强反爬。打印抓取的URL和响应状态码。用浏览器开发者工具查看页面真实结构。对于JS渲染站可考虑使用Selenium或Playwright。务必遵守robots.txt并显著降低请求频率。AI提取的信息不准确或格式错误Prompt指令不够清晰网页文本噪声太多。打印出送入模型的原始文本前500字符检查质量。查看模型返回的原始字符串。优化Prompt提供更明确的例子和格式要求。在提取前对网页文本进行更严格的清洗如移除导航栏、版权文字。最终CSV文件字段混乱模型返回的JSON结构不一致或json.loads失败。在extract_company_info函数中捕获异常并打印出导致错误的原始文本和模型输出。使用LangChain的StructuredOutputParser或PydanticOutputParser来强制结构化输出比单纯依赖JSON模式更稳定。达到“85家”目标很困难初始搜索范围窄或去重后有效链接少。分析all_links集合的数量和质量。增加搜索词生成数量如从20到50。让AI生成更细分的长尾词如“hospital elevator supplier”、“residential lift manufacturer Italy”。在搜索后引入简单的域名分类和过滤。8. 最佳实践与工程建议要将这个脚本转化为一个稳定、可用的生产级工具你需要考虑以下几点合规与伦理第一遵守robots.txt在抓取任何网站前检查其robots.txt文件尊重Disallow规则。设置礼貌延迟在请求间添加随机延迟如time.sleep(random.uniform(1, 3))避免对目标网站造成压力。明确数据用途此方法仅用于收集公开的商业信息。切勿用于爬取个人隐私数据、受版权保护内容或任何非法用途。提升数据质量多轮验证对AI提取的关键信息如邮箱、电话可通过正则表达式进行二次验证。信息融合对同一公司的多个信息来源如官网、LinkedIn、行业目录进行信息融合和去重补充单一来源的不足。人工审核闭环将AI生成的列表导入CRM或表格后应设计简单的人工审核标记流程如“有效/无效”这些反馈数据可用于微调AI的Prompt形成闭环优化。系统健壮性错误重试与熔断为网络请求和API调用实现重试逻辑如tenacity库和熔断机制避免因个别失败导致整个任务崩溃。状态持久化使用SQLite或小文件记录已处理URL的状态成功/失败支持任务中断后续跑避免重复工作。日志与监控使用Python的logging模块记录详细日志便于排查问题。监控API调用成本和成功率。扩展性设计模块化将搜索、抓取、解析、保存等模块彻底解耦方便单独替换或升级例如将Serper换成其他搜索API。支持多行业将行业关键词和对应的提取字段模板化通过配置文件管理快速适配到“光伏面板”、“医疗器械”等其他行业。集成化将最终输出的CSV自动同步到Google Sheets或通过Zapier/Make.com触发后续的邮件营销流程实现全自动化。通过以上步骤你构建的不仅仅是一个脚本而是一个可扩展、可维护的“AI外贸业务开发助手”。它将你从繁琐的信息搜集工作中解放出来让你能更专注于需要人类判断和情感连接的高价值环节——客户沟通和关系维护。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
