使用 FastAPI 和 Pydantic 用 Python 构建 AI 领导力评估工具
将原始潜在客户转化为合格客户：结构化大语言模型提取、Pydantic验证、基于规则的评分、PostgreSQL存储、队列处理以及人工审核环节。
来源: I Built an AI Lead Filtering Agent in Python — From Raw Leads to Qualified Customers
大多数企业并不缺乏潜在客户，他们的难题在于如何筛选这些客户。通过网页表单、电子邮件营销活动以及各种营销工具，可能会收到数百条咨询信息，仍需有人逐一查看、阅读内容、了解客户需求、核实预算，再判断是否值得重视。随着咨询量不断增加，这种人工筛选的工作成本也会随之上升。
实际的问题是：人工智能代理能否处理这些潜在客户的信息，理解其意图，根据企业规则进行筛选，然后将合适的客户转交给销售团队？虽然可以做到，但不能简单地将所有客户信息输入大型语言模型后就依赖其回复。一个可靠的系统需要结构化的输出、明确的规则、验证机制、数据存储以及完善的流程。以下内容将展示如何用 Python 构建这样的系统。
系统的功能
流程很简短：有潜在客户来电，系统对其进行解析、评估，保存结果后触发下一步操作。
以这位潜在客户发来的消息为例：
“我们的销售团队需要一款定制化的CRM系统，大约30个用户席位，要求在两个月内交付。大概需要多少费用？”
人工处理时需要立即提取若干关键信息：
- 存在明确的业务需求
- 该项目属于定制软件
- 团队用户数量约为30人
- 有明确的交付时间要求
- 核心问题是成本
客服人员也应提取这些关键信息，协助销售团队决定后续行动。不能认为每条详细消息都是优质潜在客户，必须根据公司的实际筛选标准来评估这些信息。
系统架构
规划中的工作流程如下所示：
┌──────────────────┐
│ Incoming Lead │
│ Website / Email │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Lead Ingestion │
│ Validate & Store │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ AI Extraction │
│ Intent, Need, │
│ Budget, Timeline │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Business Rules │
│ Qualification │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Lead Scoring │
│ Qualified / │
│ Review / Reject │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ PostgreSQL │
│ Store Result │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Automation │
│ CRM / Email / │
│ Sales Notification│
└──────────────────┘
其中有一个设计原则最为重要：将人工智能的理解功能与业务决策分开。大语言模型负责从潜在客户信息中提取数据，而由您的代码来决定这些数据如何影响客户是否符合资格。通过这种分离，系统更易于测试、调试和修改。
第一步：确定什么条件可使潜在客户符合资格
业务规则应先于代码确定。以一家软件开发机构为例，它可以将满足以下条件的潜在客户视为符合资格：
- 客户有真实的软件需求
- 该需求与机构提供的服务相匹配
- 消息包含足够详细的信息以便了解项目情况
- 客户的预算或购买意向达到机构的标准
- 消息不是垃圾信息或明显无关的咨询
这些规则仅作为示例，实际企业应自行设定阈值。一份填好的记录可能显示：服务内容为定制化CRM开发，预算为500,000卢比，时间期限为两个月，企业用户数为30人，需求紧迫程度高，且该请求符合条件。
人工智能应负责提取数据，而业务规则则用于做出最终决策。
步骤2：创建Python项目
得益于其API、大语言模型集成、数据处理及自动化方面的生态系统，Python非常适合这项工作。项目的结构如下：
ai-lead-agent/
│
├── app/
│ ├── main.py
│ ├── config.py
│ │
│ ├── agents/
│ │ └── lead_agent.py
│ │
│ ├── schemas/
│ │ └── lead.py
│ │
│ ├── services/
│ │ ├── lead_service.py
│ │ └── scoring_service.py
│ │
│ └── db/
│ └── database.py
│
├── tests/
│ └── test_scoring.py
│
├── requirements.txt
└── .env
安装所需的软件包：
pip install openai pydantic fastapi uvicorn python-dotenv
所需组件包括：
- 用于后端 API 的 FastAPI
- 用于结构化验证的 Pydantic
- 用于大语言模型示例的 OpenAI API
- 作为生产环境的数据库 PostgreSQL
- 用于潜在客户评分的 普通 Python
日后可以将大语言模型提供商更换为其他服务。
步骤 3：定义结构化的潜在客户架构
在人工智能应用中，一个常见的错误是将自由文本作为最终输出。这样的回复很难被可靠地处理：
This looks like a good lead.
The customer wants a CRM and seems interested.
目标是生成可预测的 JSON。创建 app/schemas/lead.py 文件，首先编写导入语句：
from typing import Optional, Literal
from pydantic import BaseModel, Field
接着定义模型，从输入的潜在客户开始，再确定模型必须返回的分析结果：
class LeadInput(BaseModel):
name: str
email: str
company: Optional[str] = None
message: str
class LeadAnalysis(BaseModel):
summary: str service_type: Optional[str] = None intent: Literal[
"high",
"medium",
"low",
"unknown"
] budget: Optional[float] = None currency: Optional[str] = None timeline: Optional[str] = None is_spam: bool = False missing_information: list[str] = Field(
default_factory=list
) qualification_notes: str recommended_action: Literal[
"contact_sales",
"request_more_information",
"reject",
"manual_review"
]
现在，AI的响应有了明确的格式。如果模型返回无效值，应用程序可以拒绝它或对其进行处理，而不会默默地存储错误数据。
第4步：构建潜在客户分析函数
接下来是app/agents/lead_agent.py。模型需要理解潜在客户所说的话，且不得编造不存在的内容。首先从导入语句开始：
import json
import os
接下来是分析功能本身。它将输入数据发送给模型，要求模型输出符合规范的结果，并对返回的结果进行验证：
from openai import OpenAIfrom app.schemas.lead import (
LeadInput,
LeadAnalysis
)
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY")
)
SYSTEM_PROMPT = """
You are a lead analysis assistant for a software
development agency.Your task is to analyze incoming customer leads.Extract:
- Customer requirement
- Service type
- Buying intent
- Budget if explicitly mentioned
- Timeline if explicitly mentioned
- Spam indicators
- Missing informationRules:
1. Never invent a budget.
2. Never invent a timeline.
3. If information is missing, return null.
4. Return valid structured JSON.
5. Do not make a final business decision based
only on assumptions.
"""
def analyze_lead(lead: LeadInput) -> LeadAnalysis: user_prompt = f"""
Analyze this incoming lead: Name: {lead.name}
Email: {lead.email}
Company: {lead.company}
Message: {lead.message}
""" response = client.chat.completions.create(
model="gpt-4o-mini",
temperature=0,
response_format={
"type": "json_object"
},
messages=[
{
"role": "system",
"content": SYSTEM_PROMPT
},
{
"role": "user",
"content": user_prompt
}
]
) result = json.loads(
response.choices[0].message.content
) return LeadAnalysis.model_validate(result)
为何温度要设为零
在提取结果时，需要的是一致性而非富有创意的文字表达。较低的温度能减少不同响应之间的差异，但无法保证结果的正确性。正因如此，仍需进行模式验证和业务规则检查。
模型名称及API功能也会随时间变化。在将其投入实际应用时，应选择服务商当前支持的模型及结构化输出方式。
第5步：添加资格判定规则
这部分工作不应完全交给大语言模型处理。假设该机构的示例标准如下：
- 项目属于相关软件开发工作
- 若指定了预算，金额至少为200,000卢比
- 消息不属于垃圾信息
- 包含足够的信息以便理解需求
创建 app/services/scoring_service.py 文件，首先写入导入语句：
from app.schemas.lead import LeadAnalysis
评分函数初始值为零，它会收集每次调整背后的原因，对于垃圾信息则直接返回零分：
def score_lead(analysis: LeadAnalysis) -> dict: score = 0
reasons = [] if analysis.is_spam:
return {
"score": 0,
"status": "rejected",
"reasons": ["Spam detected"]
} if analysis.service_type:
score += 25
reasons.append(
"Service requirement identified"
) if analysis.intent == "high":
score += 30
reasons.append(
"High buying intent"
) elif analysis.intent == "medium":
score += 15
reasons.append(
"Medium buying intent"
) if analysis.budget is not None:
if analysis.budget >= 200000:
score += 25
reasons.append(
"Budget meets example threshold"
)
else:
reasons.append(
"Budget below example threshold"
) if analysis.timeline:
score += 10
reasons.append(
"Timeline provided"
) if analysis.missing_information:
score -= 10
reasons.append(
"Some information is missing"
) score = max(0, min(score, 100)) if score >= 70:
status = "qualified" elif score >= 40:
status = "review" else:
status = "unqualified" return {
"score": score,
"status": status,
"reasons": reasons
}
这是一个简单的示例评分方案。实际的代理机构可能会考虑其他因素：
- 项目规模
- 行业领域
- 地理位置
- 该服务是否可用
- 客户的紧急程度
- 以往的互动记录
- 对所提供预算的信任度
- 重复的潜在客户
在让系统自动拒绝潜在客户之前，应先用实际案例测试评分功能。
第6步：将代理与FastAPI连接
创建 app/main.py 并导入FastAPI：
from fastapi import FastAPI
文件的其余部分通过一个端点将模式、分析函数和评分服务整合在一起：
from app.schemas.lead import LeadInput
from app.agents.lead_agent import analyze_lead
from app.services.scoring_service import score_lead
app = FastAPI(
title="AI Lead Filtering Agent"
)
@app.get("/health")
def health_check():
return {
"status": "ok"
}
@app.post("/analyze-lead")
def analyze_incoming_lead(
lead: LeadInput
): analysis = analyze_lead(lead) score = score_lead(analysis) return {
"lead": lead,
"analysis": analysis,
"qualification": score
}
启动服务器：
uvicorn app.main:app --reload
然后打开交互式 API 文档：
http://127.0.0.1:8000/docs
从那里您可以提交潜在客户信息，并查看结构化的回复。
第7步：尝试使用真实的潜在客户信息
发送类似以下的示例请求：
{
"name": "Rahul Sharma",
"email": "rahul@example.com",
"company": "Example Technologies",
"message": "We need a custom CRM for around 30 users. Our budget is approximately 500000 INR and we want to launch within 2 months."
}
一个合理的提取结果如下：
{
"summary": "Customer needs a custom CRM.",
"service_type": "CRM development",
"intent": "high",
"budget": 500000,
"currency": "INR",
"timeline": "2 months",
"is_spam": false,
"missing_information": [],
"qualification_notes": "Clear requirement, budget, and timeline.",
"recommended_action": "contact_sales"
}
评分服务随后会对提取的数据进行评估，输出结果如下：
{
"score": 90,
"status": "qualified",
"reasons": [
"Service requirement identified",
"High buying intent",
"Budget meets example threshold",
"Timeline provided"
]
}
请记住，此输出仅用于展示预期结果，并非来自实际 API 调用的测量数据。
第 8 步：将潜在客户信息存储在 PostgreSQL 中
在真正的业务应用中，仅将结果保存在内存中是不够的，因此需要实现数据持久化。一个简单的表结构如下所示：
CREATE TABLE leads (
id BIGSERIAL PRIMARY KEY,
其余列显示了领先者的详细信息及分析结果：
name VARCHAR(150) NOT NULL, email VARCHAR(255) NOT NULL, company VARCHAR(255), message TEXT NOT NULL, service_type VARCHAR(255), intent VARCHAR(20), budget NUMERIC(14, 2), currency VARCHAR(10), timeline VARCHAR(100), is_spam BOOLEAN DEFAULT FALSE, ai_score INTEGER, qualification_status VARCHAR(30), analysis JSONB, created_at TIMESTAMPTZ DEFAULT NOW()
);
analysis列保留了原始的结构化AI输出，这有助于你在需要排查为何某个潜在客户会被赋予特定分类时进行调试。
在正式部署时，还需考虑以下事项：
- 为每个潜在客户分配唯一标识符
- 设置机制以检测重复数据
- 为状态和创建时间建立索引
- 由于潜在客户数据属于敏感信息，需进行加密并实施严格的访问控制
- 处理模式迁移
- 保留操作审计记录
第9步：添加自动化功能
至此，该代理已不再仅仅是简单的API。可以这样理解其工作流程：
New Lead
│
▼
Save in PostgreSQL
│
▼
Analyze with AI
│
▼
Calculate Qualification
│
├── Qualified ────────► Notify Sales
│
├── Review ───────────► Human Review Queue
│
└── Unqualified ─────► Store for Reporting
当有符合条件的潜在客户出现时，系统可以：
- 将该潜在客户信息存储在数据库中
- 通知销售团队
- 创建CRM记录
- 指派销售代表
- 安排跟进时间
- 更新该潜在客户的状态
自动化层可以是n8n、后台工作进程或基于队列的系统。在实际应用中，让整个流程依赖于单个同步HTTP请求是不明智的。更稳健的设计会在中间加入队列：
Lead API
│
▼
Message Queue
│
▼
Background Worker
│
▼
AI Analysis
│
▼
Database
│
▼
Automation
潜在客户信息会以异步方式进行处理，失败的任务可以重新尝试。
第10步：添加人工审核
这可能是最重要的一步。人工智能可能会误判潜在客户。请看这条消息：
“目前我们正在比较不同的CRM系统，尚未确定预算，但希望了解大致价格。”
模型可能会将该潜在客户的意向程度标记为中等或高，但企业更希望将此类潜在客户转交给人工处理，而非自动判定为合格。这就是设置三种状态的原因：
- 合格：潜在客户符合规则，将交由销售团队处理。
- 审核中：潜在客户信息有用，但需由人工决定。
- 不合格：潜在客户未达到预设标准。
在以下情况下，人工审核最为有用：
- 未明确说明预算情况
- 需求表述模糊
- 模型似乎对自己生成的输出缺乏确定性
- 消息中的细节相互矛盾
- 发送者所在行业并非该公司服务的领域
优秀的AI系统不会隐藏不确定性，而是将其展现出来。
为何这能被视为智能体
普通的LLM调用仅涉及输入、模型和输出：
Input → LLM → Output
代理工作流的功能更为丰富。它能理解传入的任务，提取相关信息，应用相应的工具或业务规则，确定下一步操作，保存状态，触发相应动作；当出现异常时还会进行重试或升级处理。对于负责人筛选功能，其完整路径如下：
Incoming Lead
│
▼
AI Analysis
│
▼
Business Rules
│
▼
Qualification Decision
│
▼
Action / Human Review
人工智能只是其中一个组成部分。后端应用程序才能确保整个流程的可靠性，这一点非常重要。
上线前的改进
第一个版本足以用来了解工作流程，但不应直接投入使用。
后台处理。通过基于Redis的Celery、RQ、云托管队列或数据库支持的作业系统等任务队列来执行潜在客户分析。这有助于处理重试和大量数据的情况。
重复检测。同一个潜在客户可能会通过网站和电子邮件营销两种途径被录入两次。可通过电子邮件、公司名称及其他标识符进行匹配。
提示词版本管理。将每个AI结果的对应提示词版本保存下来。当规则发生变化时，便可确定是哪个提示词和规则产生了之前的结果。
可观测性。需监控人工智能的响应延迟、API错误、令牌使用情况、每条潜在客户的成本、资格评估结果的分布，以及人类干预智能体的频率。
评估数据集。应收集经过人工审核标注的真实潜在客户数据，并用这些数据来测试系统。这远比仅通过一个良好的示例来评判智能体更有意义。
安全性。必须将API密钥完全置于前端代码之外，检查每一条传入的请求，保护个人数据，并限制能够查看潜在客户信息的人员范围。
架构带来的启示
构建人工智能代理并不仅仅是调用大型语言模型而已。真正的工程工作在于模型本身：明确的业务问题、结构化数据、可靠的验证机制、确定的规则、数据持久化、自动化处理、人工审核以及监控机制。如果忽略了这些要素，最终得到的只会是看似聪明实则无法解决业务问题的聊天机器人。
潜在客户筛选工具不应仅仅输出答案，而应生成有用且可追踪的工作流程。模型负责理解杂乱无章的客户消息，Python负责执行相应流程，业务规则用于确定候选资格，PostgreSQL负责存储状态信息，而自动化机制则将处理结果传递给销售团队。通过这些环节的协同，人工智能演示便能转化为可实际使用的应用程序。
接下来的自然步骤是将该 API 与真实的潜在客户来源相连，添加后台任务，并创建一个控制面板供销售人员查看哪些潜在客户符合条件。其目的并非用人工智能取代所有的销售决策，而是减少重复性工作，让销售人员能够专注于那些真正值得关注的潜在客户。
相关阅读
- 用 Python 从零构建 AI 智能体：逐步讲解 GAME 循环 — 用纯 Python 构建具备目标、动作、内存和环境功能的工具调用智能体，随后添加工具注册表、隐藏的依赖注入机制以及类型化输出功能。