🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

如果你在外贸行业，特别是像电梯这样的工业品领域，一定经历过这样的痛苦：每天花几个小时在各种B2B平台、海关数据、企业黄页里大海捞针，手动筛选出的客户名单要么不精准，要么信息残缺，效率极低。更头疼的是，好不容易找到一些潜在客户，却不知道他们的采购偏好、公司规模、业务匹配度，销售邮件发出去石沉大海。

今天要讨论的“Codex”，并不是一个现成的、开箱即用的“外贸客户开发神器”。从网络热词和搜索趋势来看，很多人正在寻找“codex安装教程”、“codex使用”、“codex接入deepseek”，这反映出市场对一种新型、高效的AI驱动的信息处理和自动化工具的强烈需求。结合项目标题“以 Elevator 电梯为例，直接跑出 85 家目标客户 + 22 列深度分析”，我们可以清晰地判断：本文的核心，是拆解如何利用类似Codex理念的AI智能体（Agent）工作流，自动化地完成从线索挖掘、数据清洗到深度分析的全过程，从而将外贸业务开发从“体力劳动”升级为“脑力决策”。

简单说，我们不再手动搜索“电梯公司”，而是教会AI一套方法，让它自动去全网寻找电梯制造商、经销商、维保公司，并自动整理出他们的官网、联系方式、主营产品、技术特点、甚至社交媒体动态，最终生成一份可直接用于销售触达的深度分析报表。本文将为你还原这个过程的完整技术实现路径。

1. 这篇文章真正要解决的问题：从“找客户”到“懂客户”的范式转变

传统外贸开发信之所以转化率低，核心问题在于“广撒网”和“信息不对称”。你并不真正了解目标客户，发的都是千篇一律的模板邮件。

本文要解决的，正是这个痛点。我们通过构建一个智能的“客户开发AI工作流”，实现以下目标：

1. 效率提升：将原本需要数天手动完成的客户搜寻和初步筛选工作，压缩到几小时甚至几分钟内由AI自动完成。
2. 深度洞察：超越简单的公司名称和邮箱，获取包括业务范围、技术路线、近期动态、供应链位置等在内的22个维度的深度信息，为个性化营销提供弹药。
3. 流程标准化：将优秀的业务员的开发思路固化为可重复、可迭代的AI指令（Prompt）和流程，降低对个人经验的依赖。
4. 持续监控：一旦工作流建立，可以定期运行，自动发现新客户或跟踪老客户的动态变化。

以“电梯”行业为例，我们最终要得到的不是一份简单的名单，而是一张包含如下信息的智能表格：

• 基础信息：公司名、官网、国家、城市。
• 业务信息：主要产品（如乘客电梯、载货电梯、自动扶梯）、品牌、是否OEM/ODM。
• 技术信息：电梯速度、载重、驱动方式（曳引/液压）、是否具备物联网（IoT）功能。
• 市场信息：目标市场（本地/国际）、认证情况（CE, ISO）。
• 动态信息：近期新闻、招聘信息、社交媒体活跃度。
• 联系信息：关键联系人邮箱、电话、LinkedIn资料。

接下来，我们将从概念、工具选择、实战流程到代码实现，完整拆解如何构建这样一个系统。

2. 核心概念与工具选型：什么是“Codex”式工作流？

首先需要澄清概念。根据有限的网络信息，“Codex app”被描述为一个支持并行处理线程、内置工作树和Git功能的桌面体验工具。结合“codex接入deepseek”等热词，我们可以推断，在当前语境下，“Codex”更可能指的是一个以DeepSeek等大型语言模型（LLM）为核心，能够处理复杂、多步骤任务（如研究、分析、汇总）的AI智能体（Agent）或自动化工作流平台。

它不同于简单的ChatGPT对话，而是强调：

• 任务分解：将“找电梯客户”这个大任务，拆解为“搜索-筛选-提取-分析-格式化”等多个子任务。
• 工具调用：AI可以自主调用搜索引擎、访问网页、解析文档、操作表格等工具。
• 持续迭代：基于上一步的结果，规划下一步的行动，直到达成目标。
• 结果结构化：最终输出是规整的JSON或CSV，而非杂乱文本。

为了实现这样的工作流，我们需要以下几类工具：

本教程将选择最具性价比和可行性的技术栈：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，请提前申请。

1. DeepSeek API：

   • 访问 DeepSeek 官方平台注册账号。
   • 在控制台创建API Key，并记录备用。注意保管，不要泄露。

2. 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_KEY=your_deepseek_api_key_here SERPER_API_KEY=your_serper_api_key_here

4. 核心流程拆解：四步构建智能客户开发工作流

整个工作流可以抽象为四个核心阶段，如下图所示（概念图）：

[启动] -> [阶段一：智能搜索与发现] -> [阶段二：精准抓取与解析] -> [阶段三：多维度信息提取] -> [阶段四：结构化分析与输出] -> [结果报表]

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页面）。挑战：网站结构各异，反爬策略不同。策略：

1. 使用requests库模拟浏览器访问，设置合理的User-Agent和延迟，遵守robots.txt。
2. 使用BeautifulSoup解析HTML，提取主内容区域的文本。优先抓取<main>,<article>标签或包含大量文本的<div>。
3. 对抓取失败或内容过少的网站进行过滤。

4.3 阶段三：多维度信息提取

目标：从抓取到的网页文本中，提取出我们关心的22个维度的信息。这是AI的核心价值所在。我们设计一个详细的“信息提取Prompt”，指导模型像一名经验丰富的市场分析师一样阅读网页，并按要求填写信息。

例如，Prompt会这样要求模型：

请从以下公司官网文本中，提取以下信息。如果信息不存在，请填写“N/A”。 公司名称： 国家/地区： 城市： 主要产品（分类列出）： 是否提供OEM/ODM服务： 核心技术特点（如曳引机品牌、物联网功能）： ...

技术实现：将清洗后的网页文本分批送入DeepSeek模型，调用其函数调用（Function Calling）或结构化输出（JSON Mode）能力，直接返回格式化的JSON数据。

4.4 阶段四：结构化分析与输出

目标：将提取出的所有公司信息汇总、清洗、去重，并保存为结构化的CSV文件。操作：

1. 将每个公司返回的JSON数据合并成一个列表。
2. 使用pandas库进行数据清洗（处理空值、统一格式）。
3. 根据“国家”、“产品类型”等字段进行简单的统计分析。
4. 导出最终的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, headers=headers, data=payload) 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, headers=headers, timeout=10) 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=' ', strip=True) # 简化空白字符 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( model="deepseek-chat", # 根据DeepSeek最新模型名调整 openai_api_key=DEEPSEEK_API_KEY, openai_api_base=DEEPSEEK_API_BASE, temperature=0.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(industry="elevator"): 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_results=5) 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" 处理 ({idx+1}/{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", inplace=True) filename = f"{industry}_companies_analysis.csv" df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"成功！已保存 {len(df)} 家公司信息到 '{filename}'。") print("\n数据预览：") print(df[['company_name', 'country', 'main_products']].head()) else: print("未提取到有效的公司信息。") if __name__ == "__main__": # 运行主流程，可以替换行业关键词 main(industry="elevator 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_workers=5): """并行抓取和提取信息""" results = [] with ThreadPoolExecutor(max_workers=max_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), total=len(urls), desc="Processing 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 如何运行

1. 确保你的.env文件已正确配置API密钥。
2. 在项目根目录下，运行主程序：

python main_workflow.py

6.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文件被创建，且包含多条记录，关键字段（如公司名、国家、产品）非空率较高。
• 初级排查：如果结果为空或很少，请按顺序检查：
  1. API密钥：确认.env文件配置正确，且API服务无欠费。
  2. 网络连接：确认程序可以访问外网（用于搜索和抓取）。
  3. 搜索词质量：查看search_queries变量输出的搜索词是否具体、相关。可以手动调整Prompt让AI生成更精准的词。
  4. 网页抓取：检查fetch_webpage_text函数是否因网站反爬而返回空文本。可尝试增加请求头、使用Session或代理（需合法合规）。
  5. 信息提取：检查extract_company_info函数中模型返回的原始JSON字符串，看是否解析失败。可能是模型未按格式输出，需要优化Prompt。

7. 常见问题与排查思路

在实际运行中，你可能会遇到以下问题：

8. 最佳实践与工程建议

要将这个脚本转化为一个稳定、可用的生产级工具，你需要考虑以下几点：

1. 合规与伦理第一：

   • 遵守robots.txt：在抓取任何网站前，检查其robots.txt文件，尊重Disallow规则。
   • 设置礼貌延迟：在请求间添加随机延迟（如time.sleep(random.uniform(1, 3))），避免对目标网站造成压力。
   • 明确数据用途：此方法仅用于收集公开的商业信息。切勿用于爬取个人隐私数据、受版权保护内容或任何非法用途。

2. 提升数据质量：

   • 多轮验证：对AI提取的关键信息（如邮箱、电话），可通过正则表达式进行二次验证。
   • 信息融合：对同一公司的多个信息来源（如官网、LinkedIn、行业目录）进行信息融合和去重，补充单一来源的不足。
   • 人工审核闭环：将AI生成的列表导入CRM或表格后，应设计简单的人工审核标记流程（如“有效/无效”），这些反馈数据可用于微调AI的Prompt，形成闭环优化。

3. 系统健壮性：

   • 错误重试与熔断：为网络请求和API调用实现重试逻辑（如tenacity库）和熔断机制，避免因个别失败导致整个任务崩溃。
   • 状态持久化：使用SQLite或小文件记录已处理URL的状态（成功/失败），支持任务中断后续跑，避免重复工作。
   • 日志与监控：使用Python的logging模块记录详细日志，便于排查问题。监控API调用成本和成功率。

4. 扩展性设计：

   • 模块化：将搜索、抓取、解析、保存等模块彻底解耦，方便单独替换或升级（例如，将Serper换成其他搜索API）。
   • 支持多行业：将行业关键词和对应的提取字段模板化，通过配置文件管理，快速适配到“光伏面板”、“医疗器械”等其他行业。
   • 集成化：将最终输出的CSV自动同步到Google Sheets，或通过Zapier/Make.com触发后续的邮件营销流程，实现全自动化。

通过以上步骤，你构建的不仅仅是一个脚本，而是一个可扩展、可维护的“AI外贸业务开发助手”。它将你从繁琐的信息搜集工作中解放出来，让你能更专注于需要人类判断和情感连接的高价值环节——客户沟通和关系维护。

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度