刚拿到这条文案时,它看起来只是一段普通的品牌直播预告:“咚咚,咚咚,凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!8月8日 20:00-21:00【凡士林官方旗舰店】抖音直播间一起「龚」享浪漫时刻”。如果只是人工阅读,信息很直白;但如果要把这段文案接入活动管理系统、自动生成日程提醒或投放报表,第一件事就是把“代言人、日期、时间、平台、直播间、话题”等字段从自然语言里抽出来,变成结构化数据。
这次我们不讨论品牌策略,只看技术实现。以一个真实的直播公告文本为输入样例,走完基于正则表达式和大模型提示词的信息抽取流程,覆盖字段设计、环境准备、批量处理、接口调用和常见问题排查。文中代码均使用通用写法,适配本地 Python 环境和 OpenAI 兼容接口,可以直接套用到品牌公告、展览通知、课程预告、社群活动等文本解析场景。
本文适合三类读者:一是做活动运营系统、需要自动解析公告文案的 Python 开发者;二是正在学习提示词工程、想拿真实文本练手的大模型应用开发者;三是想把自定义工具接到自动化流程里,但不希望纯手工整理每条活动信息的测试和运维人员。整体难度中等,不需要深厚算法基础,只要会 Python 基础语法即可跟上。
1. 活动文案信息抽取任务与数据形态
信息抽取(Information Extraction)在业务系统里非常常见。对于一条活动文案,我们希望输出稳定的 JSON 结构,减少人工录入成本。以凡士林这条直播公告为样例,目标字段可以拆成下面几类:
| 字段 | 文案中的原始表达 | 抽取难度 | 建议处理方式 |
|---|---|---|---|
| 代言人 / 嘉宾 | @龚俊Simon | 低 | 正则或大模型均可 |
| 活动日期 | 8月8日 | 中 | 正则提取,再补年份逻辑 |
| 活动时间段 | 20:00-21:00 | 低 | 正则提取 |
| 直播平台 | 抖音 | 中 | 关键词匹配或大模型识别 |
| 直播间名称 | 凡士林官方旗舰店 | 中 | 括号匹配或大模型识别 |
| 活动 slogan | 一起「龚」享浪漫时刻 | 高 | 建议用大模型抽取 |
| 品牌归属 | 凡士林 | 中 | 大模型识别更稳 |
从数据形态看,这类文案有几个特点:
- 时间信息不完整,只有“8月8日”,没有年份。
- 平台名称、直播间名称往往用括号或“在……直播间”这类句式表达。
- 社交媒体账号用 “@用户名” 表示。
- 带有营销性修辞,比如“带着花来敲门啦”“「龚」享浪漫时刻”,纯正则无法理解语义。
- 输入文本非常短,通常在 50 到 200 字之间,很适合大模型一次性处理。
最终希望得到的结构化 JSON 大概是:
{ "brand": "凡士林", "celebrity": "龚俊Simon", "date": "2025-08-08", "start_time": "20:00", "end_time": "21:00", "platform": "抖音", "live_room": "凡士林官方旗舰店", "event_slogan": "一起「龚」享浪漫时刻", "raw_text": "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!..." }后端系统拿到这份 JSON 后,可以直接判断是否需要创建日程、推送给相关人员,也可以存库后做后续统计。接下来先看最基础的实现路线。
2. 环境准备与依赖安装
整个方案不需要 GPU,CPU 即可运行。涉及两个主要内容:一是正则方案的 Python 脚本,二是大模型 API 调用。如果希望结果更稳定,可以接入本地 Ollama 服务或任意 OpenAI 兼容接口;如果不想引入外部接口,单用正则也能完成 70% 左右的字段抽取。
建议 Python 版本为 3.9 以上,并创建一个独立虚拟环境,避免依赖冲突。
python -m venv venv # Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate需要安装的库不多,主要是:
pip install requests python-dateutil pandas说明:
- requests:用于调用大模型 API。
- python-dateutil:用于把“月日”解析成完整日期,处理年份推断。
- pandas:用于批量结果聚合和本地展示。
如果没有装 jieba 或 NLP 分词库,这个场景不需要,因为信息字段大多靠标点和关键词切分,正则和大模型已经足够。
3. 正则表达式抽取:快速但有限
先上一版纯正则可以跑通的方案。它的优点是快、可控、不需要外部服务;缺点是遇到表达方式变化会漏抽。
3.1 从文案中提取用户名和时间
假设输入文本是一段短文案,用 Python 正则提取 @ 用户名、月日、时间段。
import re from datetime import datetime text = ( "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!" "8月8日 20:00-21:00【凡士林官方旗舰店】抖音直播间一起「龚」享浪漫时刻" ) results = {} # 1. 提取 @ 用户名 m = re.search(r"@([\w\u4e00-\u9fa5]+)", text) if m: results["celebrity"] = m.group(1) # 2. 提取月日 m = re.search(r"(\d{1,2})月(\d{1,2})日", text) if m: month = int(m.group(1)) day = int(m.group(2)) results["month"] = month results["day"] = day # 3. 提取时间段 m = re.search(r"(\d{1,2}:\d{2})-(\d{1,2}:\d{2})", text) if m: results["start_time"] = m.group(1) results["end_time"] = m.group(2) print(results)运行输出:
{'celebrity': '龚俊Simon', 'month': 8, 'day': 8, 'start_time': '20:00', 'end_time': '21:00'}这条文案的时间格式很规整,所以正则一步就能拿到。需要注意的是,如果时间是“晚上8点到9点”,正则就失效了,后续需要大模型兜底。
3.2 提取平台与直播间名称
平台关键词一般可以枚举,直播间名称则依赖括号或固定句式。
# 4. 提取平台 platform_keywords = ["抖音", "快手", "淘宝", "京东", "小红书", "视频号", "哔哩哔哩"] for p in platform_keywords: if p in text: results["platform"] = p break # 5. 提取直播间名称 m = re.search(r"【([^】]+)】", text) if m: results["live_room"] = m.group(1) # 6. 如果文本里出现“一起「xx」”,提取活动 slogan m = re.search(r"一起「([^」]+)」", text) if m: results["event_slogan"] = "一起「" + m.group(1) + "」" print(results)运行输出:
{ 'celebrity': '龚俊Simon', 'month': 8, 'day': 8, 'start_time': '20:00', 'end_time': '21:00', 'platform': '抖音', 'live_room': '凡士林官方旗舰店', 'event_slogan': '一起「龚」享浪漫时刻' }到这里,8 个核心字段已经拿到 6 个,只差品牌归属和完整日期。品牌归属可以简单用关键词表匹配,但更通用做法还是交给大模型。
3.3 正则方案的局限
正则方案的优点非常明显:零成本、毫秒级返回、便于调试。但实际业务里,活动文案经常出现以下变化:
- 时间写成“8月8日晚上8点”而不是“20:00”。
- 平台写成“抖爸爸”“蓝色小鸟”等网络昵称。
- 直播间名称没有用括号,而是写成“在凡士林官方旗舰店直播间等你”。
- 品牌名没有直接出现,只写了“凡士林官方旗舰店”。
- 活动 slogan 带有大量标点符号和 Emoji,正则很难精确切分。
所以更稳的做法,是把正则作为第一层兜底,遇到漏抽字段再送入大模型做二次补充。接下来看大模型提示词抽取的具体实现。
4. 大模型提示词抽取:把一句话变成 JSON
大模型擅长处理短文本语义抽取,能理解“品牌代言人”“官方旗舰店”“一起「龚」享浪漫时刻”背后的结构关系。这里采用 OpenAI 兼容接口,可以接本地 Ollama 服务,也可以接其他厂商提供的兼容接口服务。
4.1 提示词设计
提示词是决定抽取效果的关键。不要把整段文案直接丢给模型就要求返回 JSON,而应明确给出字段定义、输出格式和边界规则。
system_prompt = """ 你是一个活动公告信息抽取助手。用户会输入一段活动文案,你需要从文案中抽取以下字段: - brand:品牌名称,例如“凡士林”。如果没有明确提到品牌,可以给 null。 - celebrity:参与活动的明星、代言人或嘉宾用户名。如果没有,给 null。 - date:活动日期,格式 YYYY-MM-DD。文案通常只给“X月X日”,请推断出最近的一个日期。 - start_time:开始时间,格式 HH:MM。 - end_time:结束时间,格式 HH:MM。如果没有明确结束时间,给 null。 - platform:直播平台或活动平台,例如“抖音”“快手”。 - live_room:直播间名称或会场名称。 - event_slogan:活动口号或传播文案中的简短主题表达。 - raw_text:原样保留用户输入的文案。 约束: 1. 只能从输入文本中抽取信息,不要编造文本中不存在的字段。 2. 如果某个字段在文本中找不到,输出 null。 3. 只输出 JSON,不要输出额外解释。 """.strip() user_text = ( "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!" "8月8日 20:00-21:00【凡士林官方旗舰店】抖音直播间一起「龚」享浪漫时刻" )这里的关键点是给出“推断最近一个日期”的说明,让模型把“8月8日”补全为具体年份,而不是只输出“08-08”。
4.2 调用兼容接口
调用代码保持通用,只需要替换接口地址和模型名。
import requests import json API_URL = "http://127.0.0.1:8000/v1/chat/completions" MODEL_NAME = "your-model-name" def extract_activity_info(text, api_url=API_URL, model=MODEL_NAME): payload = { "model": model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": text} ], "temperature": 0, "response_format": {"type": "json_object"} } response = requests.post(api_url, json=payload, timeout=120) response.raise_for_status() content = response.json()["choices"][0]["message"]["content"] return content result_text = extract_activity_info(user_text) print(result_text)注意,如果接口服务不支持response_format参数,可以去掉该参数,同时在 system prompt 里明确要求“只输出 JSON,不要输出 markdown 代码块”。
4.3 解析返回结果
很多大模型接口在温度较高时,可能返回带 markdown 代码块的文本,比如:
{"brand": "凡士林"}因此解析时要做兼容处理。
import json import re def parse_model_json(content: str) -> dict: content = content.strip() # 去掉可能的 ```json 代码块包裹 if content.startswith("```"): content = re.sub(r"^```(?:json)?\s*", "", content) content = re.sub(r"\s*```$", "", content) try: return json.loads(content) except json.JSONDecodeError: # 如果 JSON 解析失败,尝试从文本中截取第一个花括号到最后一个花括号 start = content.find("{") end = content.rfind("}") if start != -1 and end != -1 and end > start: return json.loads(content[start:end + 1]) raise parsed = parse_model_json(result_text) print(json.dumps(parsed, ensure_ascii=False, indent=2))在实际场景中,用这套提示词抽出的结果大致如下:
{ "brand": "凡士林", "celebrity": "龚俊Simon", "date": "2025-08-08", "start_time": "20:00", "end_time": "21:00", "platform": "抖音", "live_room": "凡士林官方旗舰店", "event_slogan": "一起「龚」享浪漫时刻", "raw_text": "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!..." }到这里,正则和大模型的组合思路已经跑通:先用正则抽稳定字段,再把漏掉的字段交给大模型补全。
5. 批量任务:一次性处理多条活动文案
实际运营场景里,输入往往是一批活动文案,比如 50 条品牌公告、200 条课程预告。批量处理时不能一条条手动调接口,需要设计一个可重复执行的批处理脚本。
5.1 输入输出目录设计
建议目录结构如下:
project/ ├── input/ │ └── activities.jsonl ├── output/ │ └── result.jsonl ├── process.py └── requirements.txtinput/activities.jsonl每行一条待处理文案:
{"id": 1, "text": "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!8月8日 20:00-21:00【凡士林官方旗舰店】抖音直播间一起「龚」享浪漫时刻"} {"id": 2, "text": "8月10日 19:30,XX品牌开业直播,视频号见"}使用 JSONL 而不是 JSON 的好处是支持追加写入,适合增量处理和断点续跑。
5.2 批量脚本
import json import time from pathlib import Path INPUT_FILE = Path("input/activities.jsonl") OUTPUT_FILE = Path("output/result.jsonl") OUTPUT_FAILED = Path("output/failed.jsonl") def process_one(item): text = item["text"] # 这里调用前面定义的 extract_activity_info 和 parse_model_json content = extract_activity_info(text) parsed = parse_model_json(content) parsed["raw_text"] = text parsed["id"] = item["id"] return parsed def main(): OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True) success_count = 0 fail_count = 0 with open(INPUT_FILE, "r", encoding="utf-8") as fin, \ open(OUTPUT_FILE, "a", encoding="utf-8") as fout, \ open(OUTPUT_FAILED, "a", encoding="utf-8") as ferr: for line in fin: line = line.strip() if not line: continue try: item = json.loads(line) result = process_one(item) fout.write(json.dumps(result, ensure_ascii=False) + "\n") success_count += 1 print(f"[OK] id={item.get('id')}") except Exception as e: fail_count += 1 ferr.write(line + "\n") print(f"[FAIL] id={item.get('id')}, error={e}") time.sleep(0.5) # 简单限速,避免压垮本地接口 print(f"done, success={success_count}, fail={fail_count}") if __name__ == "__main__": main()这个脚本做了三件事:逐行读取输入、调用大模型接口、把结果或失败样本分别写入不同文件。失败样本单独保存,方便后续重新处理,不至于把原始数据弄丢。
5.3 失败重试策略
批量调用本地大模型接口时,最常遇到的问题是接口超时和进程卡死。建议在调用层增加一个简单的重试机制。
def call_with_retry(text, max_retry=3, base_delay=2): for attempt in range(max_retry): try: content = extract_activity_info(text) return content except Exception as e: print(f"attempt={attempt + 1}, error={e}") if attempt < max_retry - 1: time.sleep(base_delay * (attempt + 1)) raise RuntimeError(f"failed after {max_retry} retries: {text[:50]}")加上重试后,批处理脚本的稳定性会明显提升。如果单条文本较长,可以适当把超时时间从 120 秒提高到 180 秒,但不必制造过大的并发压力。
6. 结果验证与本地展示
大模型输出的结果不一定 100% 满足要求,验证非常关键。一个轻量做法是把批量结果读进 pandas DataFrame,检查字段是否完整、日期是否在合理范围。
import pandas as pd import json rows = [] with open("output/result.jsonl", "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: rows.append(json.loads(line)) df = pd.DataFrame(rows) print(df[["id", "brand", "celebrity", "date", "start_time", "end_time", "platform", "live_room"]].to_string(index=False))输出示例:
id brand celebrity date start_time end_time platform live_room 1 凡士林 龚俊Simon 2025-08-08 20:00 21:00 抖音 凡士林官方旗舰店 2 XX品牌 None 2025-08-10 19:30 None 视频号 None如果看到关键字段大量为 null,说明提示词需要调整,或者输入文案本身确实缺少这些信息。比如第二条文案没有提“代言人”,celebrity 为 null 是正常的。
也可以在脚本里增加一个简单的校验函数:
def validate_result(data: dict): required = ["date", "start_time", "platform"] for field in required: if not data.get(field): return False, f"missing field: {field}" return True, "ok"校验规则可以根据业务需求灵活定义,不一定要求所有字段都非空。
7. 接口 API 与提醒调度扩展
如果希望把抽取能力提供给其他系统,可以封装成一个轻量 HTTP API。这里用 FastAPI 做示例。
pip install fastapi uvicornfrom fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ActivityRequest(BaseModel): text: str class ActivityResponse(BaseModel): brand: str | None = None celebrity: str | None = None date: str | None = None start_time: str | None = None end_time: str | None = None platform: str | None = None live_room: str | None = None event_slogan: str | None = None raw_text: str @app.post("/api/extract", response_model=ActivityResponse) def extract(req: ActivityRequest): content = extract_activity_info(req.text) parsed = parse_model_json(content) parsed["raw_text"] = req.text return parsed启动服务:
uvicorn api:app --host 127.0.0.1 --port 8000调用时只需要一个 POST 请求:
curl -X POST http://127.0.0.1:8000/api/extract \ -H "Content-Type: application/json" \ -d '{"text": "咚咚,咚咚凡士林亚太区品牌代言人@龚俊Simon 带着花来敲门啦!8月8日 20:00-21:00【凡士林官方旗舰店】抖音直播间一起「龚」享浪漫时刻"}'接口跑通后,可以继续扩展提醒能力。比如用 cron 定时扫描未来 24 小时内的活动,生成日历提醒或推送到群机器人。这里的核心思路是:先把文本转成结构化数据,后续所有业务动作都基于系统里的date和start_time字段判断,而不是继续人工读取原始文案。
8. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 正则提取不到 @ 用户名 | 用户名包含中文、英文混合或下划线,正则字符集过窄 | 打印文本,确认编码是否正常 | 使用[\w\u4e00-\u9fa5]+,必要时允许点号和横线 |
| 日期缺少年份 | 文案只写“8月8日” | 检查生成结果 date 字段 | 在提示词中要求推断最近一个日期,或用 dateutil 兜底 |
| API 返回超时 | 模型服务负载高或文本过长 | 查看服务端日志,检查 timeout 配置 | 增加重试次数,降低并发,分批处理 |
| 返回内容不是纯 JSON | 模型把 JSON 包在 markdown 代码块里 | 打印返回原始文本 | 在解析函数中兼容 markdown 代码块 |
| 直播平台识别错误 | 关键词表覆盖不全 | 查看 platform 字段是否为空 | 增加更多枚举关键词,或改用大模型识别 |
| 批量任务中途卡住 | 某条文本触发异常,没有走异常处理 | 检查 failed.jsonl 是否有数据 | 为单条处理加 try/except 和超时重试 |
| 结果字段大量 null | 提示词字段定义不清楚,或文案本身缺少信息 | 打印原始文案,确认信息是否真实存在 | 调整提示词,增加示例输出 |
| 本地接口占用过高 | 同时发起太多请求,模型排队 | 查看 CPU/GPU 占用和请求队列 | 降低并发数,增加 sleep 间隔 |
最常踩的坑有两个:一是接口返回结果没有稳定 JSON,导致解析直接抛异常;二是日期年份推断不统一,同一批文案可能被推断到不同年份。建议在提示词中固定规则:如果日期已经过去,取下一年的同一天;如果未来一年内存在,就取最近的那个日期。这样可以减少结果波动。
9. 最佳实践与合规边界
把一条品牌直播文案转成结构化 JSON,看起来只是一件小事,但落到生产环境时还是建议注意几点。
第一,先小样本验证。不要一上来就批量处理几千条文本。先用 20 到 30 条文案跑一遍,人工核对抽取结果,确认准确率后再扩大规模。
第二,保留原始文本。在任何处理流程中,raw_text都应该跟结构化字段一起保存。原始数据是可追溯的基础,一旦发现解析错误,还能回到原文审计。
第三,区分“规则可处理”和“规则不可处理”。正则快但脆,大模型强但慢。常规状态直接用正则,遇到未命中字段再调用大模型,成本会更可控。
第四,接口服务必须限制访问范围。封装的 API 如果暴露到公网,务必加鉴权或白名单,避免被外部滥用。内部系统调用时,建议只监听127.0.0.1或内网地址。
第五,重视版权与授权边界。本文示例是公开的品牌直播公告,属于品牌主动发布的宣传内容,可以用于技术学习和内部字段抽取演示。但如果要从社交媒体平台采集大量用户生成内容,必须确认数据来源合法、处理目的合规,不能绕过平台权限限制抓取用户隐私数据,也不能把他人创作的营销文案直接用于商用数据库。涉及明星、品牌、肖像、声音等内容,在实际业务系统中要保持警觉,只处理已经获得合法授权的素材。
第六,对抽取结果建立人工复核机制。大模型输出有概率偏差,自动抽取不能完全替代人工判断。建议在关键业务动作前增加审核步骤。
10. 总结与下一步
这条凡士林直播公告本身只是一段几十字的文案,但把它变成结构化 JSON 后,整条链路已经完整覆盖“文本输入、规则抽取、大模型补全、批量处理、API 封装、结果校验”六个环节。
建议动手时先跑通正则版,确认输入输出目录和失败样本记录逻辑,再接入大模型接口。这样即使接口不稳定,基础字段依然能先落库。最容易踩的坑是大模型返回 JSON 格式不稳定,记得在解析层做兼容。
后续可以继续扩展的方向包括:把抽取结果自动写入日历系统;用定时任务每 10 分钟扫描一批新增公告;在接口层加入多模型路由策略;给不同活动类型配置不同的提示词模板。对大多数品牌运营系统来说,这一步完成后,后续的排期、通知和报表工作就不需要再靠人工盯原文了。