
觉得Python + AI就是万能钥匙。但踩的坑,比我过去总和踩的还多。
今天就来分享这半年的实战经验——不是那种“AI改变世界”的鸡汤,而是实实在在的坑、真实的痛、有效的解决方案。
=
初期幻想:
我:用Python写一个销售数据分析脚本,要能自动生成日报
ChatGPT:好的,这是一个销售数据分析脚本...
# 生成的代码看起来专业,但:
1. 用了过时的pandas API(pd.read_csv的某些参数已弃用)
2. 假设数据格式完全规整(实际数据千奇百怪)
3. 没有处理中文编码问题(我们的数据有GBK、UTF-8混用)
4. 性能极差(全量读取2GB的CSV文件)实际运行结果:
# ChatGPT生成的"完美"代码
importpandasaspd
defanalyze_sales_data():
df = pd.read_csv('sales_data.csv') # 问题1:2GB文件直接读,内存爆炸
daily_sales = df.groupby('date')['amount'].sum() # 问题2:date列格式不一致
# 生成报告
report = f"日销售总额: {daily_sales.sum()}"
returnreport
# 实际运行报错:
# MemoryError: Unable to allocate 5.3 GiB
# 或者:KeyError: 'date'(列名实际是'sale_date')
# 或者:UnicodeDecodeError: 'utf-8' codec can't decode...解决方案:分步骤、给示例、要解释
改进后的提问方式:
我:我需要一个销售数据分析脚本,分步骤实现:
第一步:数据加载
- 文件:sales_data.csv,约2GB
- 编码:可能是UTF-8或GBK,需要自动检测
- 内存限制:不能一次性加载,需要分块处理
- 实际列名:sale_date(日期),product_id(产品ID),amount(金额),region(区域)
请先写数据加载部分,用chunksize参数,并处理编码问题。ChatGPT改进后的代码:
importpandasaspd
importchardet
frompathlibimportPath
defdetect_encoding(file_path):
"""检测文件编码"""
withopen(file_path, 'rb') asf:
raw_data = f.read(10000) # 读取前10000字节检测
result = chardet.detect(raw_data)
returnresult['encoding']
defload_sales_data_chunked(file_path, chunksize=100000):
"""分块加载大文件"""
encoding = detect_encoding(file_path)
# 列名映射(实际列名 -> 标准列名)
column_mapping = {
'sale_date': 'date',
'sale_date_str': 'date',
'销售日期': 'date',
'product_id': 'product_id',
'产品编号': 'product_id',
'amount': 'amount',
'金额': 'amount',
'region': 'region',
'区域': 'region'
}
chunks = []
forchunkinpd.read_csv(file_path,
encoding=encoding,
chunksize=chunksize,
low_memory=False):
# 重命名列
chunk = chunk.rename(columns=lambdax: column_mapping.get(str(x).strip(), x))
# 确保必要的列存在
required_cols = ['date', 'amount']
missing = [colforcolinrequired_colsifcolnotinchunk.columns]
ifmissing:
raiseValueError(f"缺少必要列: {missing}")
chunks.append(chunk)
# 合并(如果内存允许)
iflen(chunks) *chunksize<1_000_000: # 小于100万行
returnpd.concat(chunks, ignore_index=True)
else:
# 返回生成器,流式处理
return (chunkforchunkinchunks)
# 测试代码
try:
data = load_sales_data_chunked('sales_data.csv', chunksize=50000)
print("数据加载成功,开始处理...")
exceptExceptionase:
print(f"数据加载失败: {e}")经验总结:
问题现象:
# 我的项目逐渐变成了这样:
# utils.py(第1次让ChatGPT写的)
defprocess_data_v1(data):
# 风格1:喜欢用lambda和map
returnlist(map(lambdax: x*2, filter(lambdax: x>0, data)))
# analytics.py(第2次让ChatGPT写的)
defprocess_data_v2(data):
# 风格2:喜欢用列表推导式
return [x*2forxindataifx>0]
# report_generator.py(第3次让ChatGPT写的)
defprocess_data_v3(data):
# 风格3:喜欢用pandas
importpandasaspd
s = pd.Series(data)
returns[s>0].mul(2).tolist()
# 同一个功能,三种写法,维护起来要疯更严重的是依赖冲突:
# 脚本A(让ChatGPT写的)
importpandasaspd
print(pd.__version__) # 2.0.0,用了新API
# 脚本B(让ChatGPT写的)
importpandasaspd
# 用了1.x的API,在2.0中已弃用
df = pd.DataFrame()
df.append(other_df) # 在pandas 2.0中已弃用!
# 脚本C(我自己写的)
# 到底该用哪个版本的pandas?!解决方案:建立项目规范
第一步:创建项目模板
# project_template/
# ├── pyproject.toml # 依赖管理
# ├── .pre-commit-config.yaml # 代码检查
# ├── .python-version # Python版本
# ├── src/
# │ ├── __init__.py
# │ ├── config.py # 配置管理
# │ └── utils/ # 工具函数
# ├── tests/
# └── scripts/ # 可执行脚本
# pyproject.toml
[project]
name = "ai-automation-project"
version = "1.0.0"
requires-python = ">=3.9"
dependencies = [
"pandas>=2.0.0,<3.0.0", # 明确版本范围
"requests>=2.28.0,<3.0.0",
"openpyxl>=3.1.0,<4.0.0",
"pydantic>=2.0.0,<3.0.0",
]
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[tool.black]
line-length = 88
target-version = ['py39']
[tool.ruff]
select = ["E", "F", "I", "B", "C4", "N", "A", "S", "T", "Q", "RUF"]
ignore = []
line-length = 88第二步:给ChatGPT的提示模板
请按照以下规范编写Python代码:
项目规范:
1. Python版本:3.9+
2. 代码风格:Black格式化,Ruff检查
3. 类型提示:必须添加
4. 异常处理:必须包含
5. 日志记录:使用logging模块
6. 依赖:只能使用以下库(及指定版本):
- pandas>=2.0.0,<3.0.0
- requests>=2.28.0,<3.0.0
- pydantic>=2.0.0,<3.0.0
代码要求:
1. 函数必须有类型提示和文档字符串
2. 使用with语句管理资源
3. 使用pathlib处理路径
4. 主要逻辑要有单元测试
请写一个函数,功能是...第三步:代码审查清单每次让ChatGPT生成代码后,检查:
典型案例:客户分群算法
我:用Python实现一个客户RFM分群算法
ChatGPT:生成了一套完整的RFM分析代码,有聚类、可视化、报告...
看起来很专业,但:
问题1:我们的客户数据只有购买记录,没有"最近一次购买时间"
问题2:业务上不需要聚类,只需要简单规则分群
问题3:生成的报告格式不符合公司要求
问题4:没有考虑数据质量问题(大量测试数据混入)实际需要的 vs AI生成的:
# AI生成的"专业"版本(复杂但无用)
fromsklearn.clusterimportKMeans
fromsklearn.preprocessingimportStandardScaler
importmatplotlib.pyplotasplt
defrfm_clustering(df):
# 复杂的RFM计算
df['Recency'] = (pd.Timestamp.now() -df['last_purchase']).dt.days
df['Frequency'] = df.groupby('customer_id')['purchase_id'].transform('count')
df['Monetary'] = df.groupby('customer_id')['amount'].transform('sum')
# 标准化
scaler = StandardScaler()
rfm_scaled = scaler.fit_transform(df[['Recency', 'Frequency', 'Monetary']])
# KMeans聚类
kmeans = KMeans(n_clusters=5, random_state=42)
df['Cluster'] = kmeans.fit_predict(rfm_scaled)
# 可视化
fig = plt.figure(figsize=(12, 8))
ax = fig.add_subplot(111, projection='3d')
# ... 一堆可视化代码
returndf
# 实际需要的版本(简单但实用)
defsimple_customer_segmentation(orders_df):
"""
简单客户分群(基于业务规则)
业务规则:
1. 重要客户:近30天有购买且累计消费>10000
2. 活跃客户:近90天有购买
3. 流失风险:近90天无购买但历史有消费
4. 流失客户:近180天无购买
5. 新客户:首次购买在30天内
"""
today = pd.Timestamp.now()
# 计算客户指标
customer_stats = orders_df.groupby('customer_id').agg({
'order_date': ['max', 'min', 'count'], # 最近购买、首次购买、购买次数
'amount': 'sum' # 累计消费
})
customer_stats.columns = ['last_purchase', 'first_purchase', 'purchase_count', 'total_amount']
# 应用业务规则
conditions = [
# 重要客户
(customer_stats['last_purchase'] >= today-pd.Timedelta(days=30)) &
(customer_stats['total_amount'] >10000),
# 活跃客户
(customer_stats['last_purchase'] >= today-pd.Timedelta(days=90)) &
(customer_stats['total_amount'] <= 10000),
# 流失风险
(customer_stats['last_purchase'] <today-pd.Timedelta(days=90)) &
(customer_stats['last_purchase'] >= today-pd.Timedelta(days=180)) &
(customer_stats['total_amount'] >0),
# 流失客户
(customer_stats['last_purchase'] <today-pd.Timedelta(days=180)),
# 新客户
(customer_stats['first_purchase'] >= today-pd.Timedelta(days=30))
]
segments = ['重要客户', '活跃客户', '流失风险', '流失客户', '新客户']
customer_stats['segment'] = np.select(conditions, segments, default='其他')
# 过滤测试数据(业务特定规则)
customer_stats = customer_stats[~customer_stats.index.str.startswith('TEST_')]
returncustomer_stats解决方案:业务逻辑自己写,AI只做工具部分
分工原则:
协作流程:
1. 我:先写业务逻辑伪代码
2. 我:让AI实现其中的工具函数
3. 我:组装工具函数,添加业务逻辑
4. 我:测试和优化示例:
# 第一步:我写业务逻辑框架
defprocess_orders(orders_data):
"""
处理订单数据的主要流程
"""
# 1. 数据清洗(让AI实现)
cleaned_data = clean_order_data(orders_data)
# 2. 业务验证(我自己写)
ifnotvalidate_business_rules(cleaned_data):
raiseBusinessRuleError("数据不符合业务规则")
# 3. 计算指标(部分让AI实现)
metrics = calculate_business_metrics(cleaned_data)
# 4. 生成报告(让AI实现基础,我添加业务格式)
report = generate_business_report(metrics)
returnreport
# 第二步:让AI实现clean_order_data
"""
请实现clean_order_data函数:
输入:订单DataFrame,包含列:order_id, customer_id, order_date, amount, status
清洗要求:
1. 删除status为'cancelled'的订单
2. 将order_date转换为datetime,处理多种格式
3. 删除amount为负数或大于100万的异常值
4. 处理customer_id为空的记录
5. 返回清洗后的DataFrame
"""
# 第三步:我自己写业务验证
defvalidate_business_rules(data):
"""业务特定验证"""
# 规则1:周末不允许有大额订单(业务规定)
weekend_big_orders = data[
(data['order_date'].dt.dayofweek>= 5) & # 周末
(data['amount'] >50000)
]
iflen(weekend_big_orders) >0:
logger.warning(f"发现{len(weekend_big_orders)}笔周末大额订单")
# 业务决策:需要人工审核
returnFalse
# 规则2:同一客户一天内订单不能超过10笔
daily_orders = data.groupby(['customer_id', data['order_date'].dt.date]).size()
if (daily_orders>10).any():
logger.warning("有客户一天内订单超过10笔,可能异常")
returnFalse
returnTrue惨痛教训:
# AI生成的邮件发送函数
defsend_email_report(recipients, report_data):
"""发送邮件报告"""
importsmtplib
fromemail.mime.textimportMIMEText
msg = MIMEText(report_data, 'html')
msg['Subject'] = '每日报告'
msg['From'] = 'reports@company.com'
msg['To'] = ', '.join(recipients) # 问题在这里!
withsmtplib.SMTP('smtp.company.com') asserver:
server.send_message(msg)
returnTrue
# 看起来没问题,直到:
# 1. recipients超过100人,SMTP服务器拒绝
# 2. 某个邮箱地址无效,整个发送失败
# 3. HTML报告有特殊字符,编码错误
# 4. SMTP服务器偶尔超时,没有重试机制解决方案:测试驱动 + 防御性编程
第一步:让AI先写测试
请先为send_email_report函数编写测试用例,包括:
1. 正常情况测试
2. 边界情况:空收件人列表、超长收件人列表
3. 异常情况:无效邮箱格式、SMTP连接失败
4. 性能测试:发送1000封邮件的耗时第二步:防御性编程要求
# 改进后的邮件发送函数
fromtypingimportList, Optional
importlogging
fromtenacityimportretry, stop_after_attempt, wait_exponential
logger = logging.getLogger(__name__)
classEmailError(Exception):
"""邮件发送异常"""
pass
defvalidate_email(email: str) ->bool:
"""简单的邮箱验证"""
importre
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
returnbool(re.match(pattern, email))
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
defsend_email_safe(
recipients: List[str],
subject: str,
content: str,
max_recipients: int = 50
) ->dict:
"""
安全发送邮件
Args:
recipients: 收件人列表
subject: 邮件主题
content: 邮件内容
max_recipients: 单次发送最大收件人数
Returns:
发送结果统计
"""
ifnotrecipients:
logger.warning("收件人列表为空")
return {"sent": 0, "failed": 0, "invalid": 0}
# 验证邮箱格式
valid_emails = []
invalid_emails = []
foremailinrecipients:
ifvalidate_email(email):
valid_emails.append(email)
else:
invalid_emails.append(email)
logger.warning(f"无效邮箱地址: {email}")
ifnotvalid_emails:
raiseEmailError("没有有效的收件人邮箱")
# 分批发送(避免SMTP限制)
batches = [
valid_emails[i:i+max_recipients]
foriinrange(0, len(valid_emails), max_recipients)
]
results = {"sent": 0, "failed": 0, "invalid": len(invalid_emails)}
forbatch_num, batchinenumerate(batches, 1):
logger.info(f"发送第{batch_num}批,共{len(batch)}个收件人")
try:
# 实际发送逻辑
send_email_batch(batch, subject, content)
results["sent"] += len(batch)
logger.info(f"第{batch_num}批发送成功")
exceptExceptionase:
results["failed"] += len(batch)
logger.error(f"第{batch_num}批发送失败: {e}")
# 记录失败的邮箱以便重试
log_failed_emails(batch, str(e))
returnresults
defsend_email_batch(recipients: List[str], subject: str, content: str):
"""实际发送邮件的函数(由AI实现基础,我增强)"""
# AI生成的发送逻辑,但包装了更多检查
iflen(recipients) >50:
raiseValueError("单批收件人不能超过50人")
# 检查内容长度
iflen(content) >1024*1024: # 1MB
raiseValueError("邮件内容过大")
# 实际发送代码...
pass第三步:完整的测试套件
importpytest
fromunittest.mockimportMock, patch
classTestEmailSender:
"""邮件发送测试"""
deftest_valid_emails(self):
"""测试有效邮箱"""
recipients = ["test@example.com", "user@domain.com"]
result = send_email_safe(recipients, "测试", "内容")
assertresult["sent"] == 2
assertresult["invalid"] == 0
deftest_invalid_emails(self):
"""测试无效邮箱"""
recipients = ["invalid-email", "test@com"]
result = send_email_safe(recipients, "测试", "内容")
assertresult["invalid"] == 2
assertresult["sent"] == 0
deftest_mixed_emails(self):
"""测试混合邮箱"""
recipients = ["valid@example.com", "invalid", "another@test.com"]
result = send_email_safe(recipients, "测试", "内容")
assertresult["sent"] == 2
assertresult["invalid"] == 1
@patch('smtplib.SMTP')
deftest_smtp_failure(self, mock_smtp):
"""测试SMTP失败"""
mock_smtp.return_value.send_message.side_effect = Exception("SMTP错误")
recipients = ["test@example.com"]
# 应该重试3次
withpytest.raises(EmailError):
send_email_safe(recipients, "测试", "内容")
assertmock_smtp.return_value.send_message.call_count == 3
deftest_large_recipient_list(self):
"""测试大量收件人"""
recipients = [f"user{i}@example.com"foriinrange(120)]
result = send_email_safe(recipients, "测试", "内容", max_recipients=50)
# 应该分成3批发送
assertresult["sent"] == 120
# 这里需要mock实际的发送来验证分批逻辑典型案例:数据去重
# AI生成的去重方法(O(n²)复杂度)
defremove_duplicates(data):
"""删除列表中的重复项"""
unique = []
foritemindata:
ifitemnotinunique: # 每次都要遍历unique列表
unique.append(item)
returnunique
# 10万条数据时,运行时间:约10秒
# 优化后(O(n)复杂度)
defremove_duplicates_fast(data):
"""使用集合去重(保持顺序)"""
seen = set()
unique = []
foritemindata:
ifitemnotinseen: # 集合查找是O(1)
seen.add(item)
unique.append(item)
returnunique
# 10万条数据时,运行时间:约0.02秒性能优化检查清单:
# 每次让AI生成代码后,问这些问题:
# 1. 时间复杂度是多少?
# AI:这个函数是O(n²),因为嵌套循环
# 我:能否优化到O(n)或O(n log n)?
# 2. 内存使用如何?
# AI:这个函数需要复制整个数据集
# 我:能否流式处理或分块处理?
# 3. 有无不必要的重复计算?
# AI:这个函数每次调用都重新计算
# 我:能否缓存结果?
# 4. 是否使用了合适的数据结构?
# AI:这个函数用列表查找
# 我:能否用集合或字典?
# 5. 是否可以利用并行计算?
# AI:这个函数是单线程的
# 我:能否用多线程或多进程?实际优化案例:
# 优化前:AI生成的报告生成函数
defgenerate_daily_reports(start_date, end_date):
"""生成每日报告(性能极差)"""
reports = []
current = start_date
whilecurrent<= end_date:
# 每次都要查询数据库
daily_data = query_database(current)
# 复杂的计算
report = calculate_report(daily_data)
# 格式化
formatted = format_report(report)
reports.append(formatted)
current += timedelta(days=1)
returnreports
# 问题:
# 1. 频繁的数据库查询(N+1问题)
# 2. 没有缓存
# 3. 没有并行处理
# 优化后:
fromconcurrent.futuresimportThreadPoolExecutor
fromfunctoolsimportlru_cache
@lru_cache(maxsize=30) # 缓存30天的数据
defget_daily_data_cached(date):
"""带缓存的数据库查询"""
returnquery_database(date)
defgenerate_daily_reports_optimized(start_date, end_date):
"""优化后的报告生成"""
# 1. 批量预取数据
all_dates = []
current = start_date
whilecurrent<= end_date:
all_dates.append(current)
current += timedelta(days=1)
# 2. 并行查询(如果数据库支持)
withThreadPoolExecutor(max_workers=5) asexecutor:
all_data = list(executor.map(get_daily_data_cached, all_dates))
# 3. 批量计算
reports = []
fordate, datainzip(all_dates, all_data):
report = calculate_report(data)
formatted = format_report(report)
reports.append(formatted)
returnreports
# 性能对比:
# 优化前:30天报告需要180秒(每天6秒)
# 优化后:30天报告需要30秒(并行+缓存)1. AI是放大器,不是替代品
2. 分而治之的协作模式
我的工作:
- 业务逻辑设计
- 系统架构规划
- 异常处理设计
- 性能优化策略
- 测试用例设计
AI的工作:
- 工具函数实现
- 数据处理代码
- 通用算法实现
- 文档生成
-mock数据生成3. 质量保证体系
# 必须有的检查点
CHECKPOINTS = {
"code_style": "black + ruff",
"type_hints": "mypy检查",
"tests": "pytest覆盖率>80%",
"performance": "基准测试",
"security": "安全扫描",
"documentation": "API文档生成"
}
# 自动化流水线
defai_assisted_development(task):
"""AI辅助开发流程"""
# 1. 需求分析(我)
requirements = analyze_requirements(task)
# 2. 架构设计(我)
architecture = design_architecture(requirements)
# 3. 代码生成(AI + 我)
code = generate_code_with_ai(architecture)
# 4. 测试编写(AI生成,我补充)
tests = generate_tests_with_ai(code)
# 5. 代码审查(我 + 工具)
review_results = code_review(code, tests)
# 6. 性能优化(我指导,AI实现)
optimized = optimize_performance(code)
# 7. 文档生成(AI生成,我校对)
documentation = generate_documentation(optimized)
return {
"code": optimized,
"tests": tests,
"docs": documentation,
"review": review_results
}4. 持续学习和调整
技术永远在变,但好的工程实践不会变。 AI不会让糟糕的开发者变好,但会让好的开发者变得更好。
现在,当我面对新任务时,我的思考流程是:
这不是放弃思考,而是更高效的思考。就像有了IDE,我们依然要理解代码;有了AI,我们依然要理解问题。
“无他,惟手熟尔”!有需要的用起来!
本文分享自 Nicholas与Pypi 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!