基于大模型API的活动策划辅助系统设计与实现

1. 背景与问题

儿童节活动策划涉及创意生成、内容编排、物料设计、执行跟踪等多个环节。传统方式依赖人工经验，存在三个核心痛点。

效率层面，策划一份完整方案通常需要三到五人协作两到三天。创意层面，受限于个人经验积累，方案容易陷入重复模式。执行层面，方案转化为可执行任务的过程依赖手工拆解，易遗漏细节。

大模型API的出现为这些问题的解决提供了新的技术路径。本文将从工程实现角度，分享一套基于大模型API构建的活动策划辅助系统，涵盖提示词工程、结构化输出解析、批量任务调度、检索增强生成等关键技术。系统已投入实际使用，累计处理活动方案五十二份。

2. 系统总体架构

系统采用分层解耦设计，核心模块分为四层。

接入层负责接收用户请求，参数校验与路由分发。业务层包含提示词管理、输出解析、任务调度三个核心组件。数据层存储活动模板、历史方案、执行记录。基础设施层封装大模型API调用、向量数据库、缓存等通用能力。

class ActivityPlannerSystem: def __init__(self): self.prompt_manager = PromptTemplateManager() self.output_parser = StructuredOutputParser() self.task_scheduler = TaskScheduler() self.rag_engine = RetrievalAugmentedGenerator() def generate_plan(self, requirements): template = self.prompt_manager.select_template(requirements) raw_output = self.call_llm(template.fill(requirements)) structured_plan = self.output_parser.parse(raw_output) tasks = self.task_scheduler.decompose(structured_plan) return structured_plan, tasks

各模块通过接口契约通信，支持独立升级替换。提示词模板纳入版本控制系统，所有变更可追溯。

3. 提示词模板的工程化设计

提示词模板的质量直接决定输出效果。经过多次迭代，形成了一套分层模板结构。

class PromptTemplateManager: def __init__(self): self.templates = { 'creative_generation': { 'system': '你是一名儿童活动策划专家，拥有十年以上经验。', 'user': ''' 【任务】为{age_range}儿童设计一场{theme}主题的活动方案。 【约束条件】 - 参与人数：{participant_count} - 场地限制：{venue_constraints} - 时长限制：{duration_minutes} - 预算等级：{budget_level} 【输出格式】 请按以下JSON结构输出，确保所有字段完整： { "plan_name": "活动名称", "core_idea": "核心创意说明(不超过200字)", "timeline": [{"time": "时间点", "activity": "活动内容", "duration": 时长, "materials": ["物料"]}], "materials": [{"category": "类别", "items": [{"name": "名称", "quantity": "数量"}]}], "staff": [{"role": "角色", "count": 人数, "responsibilities": ["职责"]}], "risks": [{"scenario": "场景", "response": "应对措施"}] } ''' }, 'host_script': {...}, 'quiz_questions': {...} } def fill_template(self, template_name, variables): template = self.templates[template_name] filled_user = template['user'].format(**variables) return ChatPromptTemplate.from_messages([ SystemMessage(content=template['system']), HumanMessage(content=filled_user) ])

模板设计遵循三个原则。约束前置，将所有限制条件放在提示词开头。格式锁定，明确指定JSON schema。示例引导，在复杂场景中提供一到两个输出样例。

4. 结构化输出解析与容错机制

大模型生成的JSON输出存在格式不稳定、字段缺失、数据类型错误等问题。需要建立专门的解析与验证层。

class StructuredOutputParser: def __init__(self): self.schemas = { 'activity_plan': { 'type': 'object', 'required': ['plan_name', 'core_idea', 'timeline', 'materials', 'staff', 'risks'], 'properties': { 'plan_name': {'type': 'string', 'maxLength': 50}, 'timeline': { 'type': 'array', 'items': { 'type': 'object', 'required': ['time', 'activity', 'duration'], 'properties': { 'time': {'type': 'string', 'pattern': '^\\d{2}:\\d{2}$'}, 'duration': {'type': 'integer', 'minimum': 1, 'maximum': 60} } } } } } } def parse(self, raw_output, schema_name): json_text = self._extract_json(raw_output) try: data = json.loads(json_text) validate(data, self.schemas[schema_name]) return data except json.JSONDecodeError: return self._repair_json(json_text, schema_name) except ValidationError: return self._fill_missing_fields(data, schema_name) def _extract_json(self, text): import re match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', text, re.DOTALL) return match.group(0) if match else '{}'

解析层实现了三级容错。提取失败时重试，最多三次。JSON格式错误时尝试修复，如补齐缺失的括号。字段缺失时根据schema补全默认值。

5. 批量任务调度与并发控制

多场活动同时策划时，需要批量调用大模型API。并发控制是保证系统稳定的关键。

class BatchTaskProcessor: def __init__(self, max_concurrent=5): self.semaphore = asyncio.Semaphore(max_concurrent) self.timeout = ClientTimeout(total=120) async def process_batch(self, requests): async with ClientSession(timeout=self.timeout) as session: tasks = [self._process_one(session, req) for req in requests] results = await asyncio.gather(*tasks, return_exceptions=True) return results async def _process_one(self, session, request): async with self.semaphore: await asyncio.sleep(0.1) # 请求间隔 async with session.post(API_URL, json=request) as response: return await response.json()

并发数设置为五时，一百个方案的批量处理时间约为八分钟，单方案平均处理时间约四秒。信号量控制确保不超过API限流阈值，请求间隔避免瞬时峰值。

6. 检索增强生成在应急预案中的应用

活动现场的应急预案需要根据场地实际情况动态生成。RAG技术将场地信息与通用预案模板结合，生成针对性方案。

class RAGEnhancedGenerator: def __init__(self, vector_store, embedding_model): self.vector_store = vector_store self.embedding = embedding_model def generate_contingency_plan(self, venue_description, activity_type): query_vector = self.embedding.encode(venue_description) similar_cases = self.vector_store.search(query_vector, top_k=3) context = '\n\n'.join([case.content for case in similar_cases]) prompt = f''' 参考以下类似场地的应急预案案例： {context} 请为以下活动生成应急预案： - 活动类型：{activity_type} - 场地情况：{venue_description} 输出格式：按风险等级从高到低排列，每个风险包含触发条件、应对措施、负责人、备用方案。 ''' return self.call_llm(prompt)

向量数据库存储了历史活动的场地特征与应急预案。经过二十场活动的数据积累，预案生成的准确率从百分之六十提升到了百分之八十五。RAG方案相比直接生成，准确率提升约百分之四十。

7. 提示词模板的迭代优化体系

提示词模板需要持续优化。建立了离线评估和在线监控两条路径。

离线评估使用历史成功案例作为测试集，每次模板变更后运行回归测试。

class TemplateEvaluator: def __init__(self, test_cases): self.test_cases = test_cases def evaluate(self, template): results = [] for case in self.test_cases: output = template.fill_and_call(case['input']) score = self._calculate_score(output, case['expected']) results.append(score) return { 'avg_score': sum(results) / len(results), 'pass_rate': len([s for s in results if s > 0.8]) / len(results) }

在线监控收集实际使用中的输出质量。当连续十个方案的解析失败率超过百分之十时，触发告警并自动回滚到上一个稳定模板。这个闭环机制保证了系统在模型版本更新时的稳定性。

8. 成本控制策略

大模型API调用成本是系统规模化应用的关键制约因素。

缓存策略方面，相同或相似的请求命中缓存后直接返回。缓存键由请求参数的哈希值生成，有效期七天。缓存命中率约为百分之三十。

请求合并方面，多个独立的小任务合并为一次大请求。三个不同环节的主持词生成合并为一次调用，Token消耗减少约百分之四十。

模型分级方面，简单任务如物料清单整理使用轻量级模型，复杂任务如创意生成使用高性能模型。动态路由后，总成本降低约百分之三十五。

三项策略叠加，单次活动策划的API成本控制在两元以内。

9. 部署与可观测性

系统采用容器化部署，通过Kubernetes编排。

version: '3.8' services: api_gateway: build: . ports: - "8080:8080" environment: - LLM_API_KEY=${LLM_API_KEY} - REDIS_URL=redis://cache:6379 - VECTOR_STORE_URL=qdrant://vector_db:6333 healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/health"] interval: 30s timeout: 10s retries: 3

监控指标包括API调用延迟、Token消耗、解析成功率、缓存命中率。接入Prometheus和Grafana，设置告警规则。解析失败率连续十分钟超过百分之十时发送告警通知。部署至今，系统可用性保持在百分之九十九点五以上。

10. 效果评估与总结

系统已在三个儿童节活动中投入实际使用。累计处理活动方案五十二份，生成主持词一百六十余篇，物料提示词两百余组。人工复核后采纳率约为百分之七十。

核心经验总结如下。提示词模板需要结构化设计，输出格式必须明确约束。解析层承担容错职责，不能直接依赖模型输出的稳定性。批量处理需要控制并发，避免触发限流。RAG技术可以显著提升场景化内容的准确性。成本控制需要从缓存、合并、分级多个维度综合施策。

以上方案中的代码和配置均来自实际项目，已做脱敏处理。完整代码仓库及部署文档已开源，欢迎在评论区留言获取访问地址。