import os from dotenv import load_dotenv from crewai import Agent, Task, Crew, Process from langchain_openai import ChatOpenAI # 加载环境变量 (OPENAI_API_KEY) load_dotenv() # 初始化 LLM，使用 Deepseek API # Please install OpenAI SDK first: `pip3 install openai` llm = ChatOpenAI( api_key=os.getenv('DEEPSEEK_API_KEY'), base_url=os.getenv('DEEPSEEK_API_BASE'), model_name="deepseek/deepseek-chat", temperature=0.7, ) # --- 全局工具和上下文加载 --- def load_jargon_dictionary(file_path='livestream_jargon.txt'): """加载直播间黑话字典作为上下文""" try: with open(file_path, 'r', encoding='utf-8') as f: return f.read() except FileNotFoundError: print(f"警告: 黑话字典文件 {file_path} 未找到。") return "" # 加载字典内容 jargon_context = load_jargon_dictionary() # --- 定义智能体 (Agents) --- # 1. 文本预处理专员 cleaner_agent = Agent( role='文本预处理专员', goal='修正语音转录稿中的错别字，移除无效语气词，并将因多人抢话导致的断裂句子进行逻辑合并，使文本流畅易读。', backstory=( '你是一位经验丰富的转录稿校对员，对中文口语表达习惯了如指掌。' '你特别擅长处理直播领域的文本，能够精准识别并修正行业黑话相关的错别字，' '同时去除“嗯、啊、是吧、你知道吧”等冗余信息，让混乱的原文变得清晰、专业。' ), verbose=True, allow_delegation=False, llm=llm ) # 2. 主题内容筛选师 analyst_agent = Agent( role='互联网娱乐直播主题分析师', goal='从清洗后的文本中，精准筛选并提取出所有与“互联网娱乐直播运营策略、技巧、建议、分析”相关的核心内容。', backstory=( '你是一位深耕直播行业多年的分析师，能够迅速从海量对话中识别出真正有价值的“干货”。' '你的任务是过滤掉日常闲聊、跑题内容、纯粹的情绪表达以及与直播运营无关的对话，' '只保留那些能为直播从业者提供具体指导和启发的内容。' ), verbose=True, allow_delegation=False, llm=llm ) # 3. 知识结构化专家 architect_agent = Agent( role='知识结构化专家', goal='将提炼出的核心内容，识别出其中的问答（QA）对，并将其转换成指定的JSON格式的知识条目。', backstory=( '你是一名顶级的知识管理专家，擅长将非结构化的信息转化为条理清晰、易于检索的知识库。' '你将接收一份关于直播运营的纯文本“干货”，你的任务是将其中的每一个独立的知识点或问答，' '都 meticulously 地构造成一个标准的JSON对象，确保所有字段都得到恰当、准确的填充。' ), verbose=True, allow_delegation=False, llm=llm ) # --- 定义任务 (Tasks) --- def create_processing_tasks(jargon_context): """ 为单个MD文件创建处理任务流 """ # 任务1: 清洗文本 task_clean = Task( description=( '1. 读取原始Markdown文本：`{markdown_text}`\n' '2. 参考这份《直播间黑话字典》进行校对：`{jargon_dictionary}`\n' '3. **核心要求**:\n' ' a. **修正错别字**: 特别是字典中提到的黑话，例如将“场馆”修正为“场观”。\n' ' b. **移除无效词**: 删除如“嗯”、“啊”、“哼”、“是吧”、“你知道吧？”、“完了”等大量出现的口水词。\n' ' c. **合并断句**: 将不同发言人之间因快速抢话而打断的句子，根据逻辑重新组合成完整、通顺的段落。\n' ' d. **保留发言人标识**: 在处理后，依然保留“发言人X”的标识，但可以合并连续的、属于同一发言人的段落。\n' '4. **输出**: 返回处理后的、干净流畅的完整文本。' ), expected_output='一份经过校对、去冗、合并断句后的，但仍保留发言人信息的纯净文本。', agent=cleaner_agent, llm=llm ) # 任务2: 筛选核心内容 task_filter = Task( description=( '1. 分析由上一步任务输出的纯净文本。\n' '2. **核心要求**: 识别并抽取出所有直接讨论“互联网娱乐直播”的策略、技巧、数据分析、观众互动、变现方法、内容创作、平台规则等核心“干货”内容。\n' '3. **需要剔除的内容**: \n' ' - 与主题无关的闲聊（例如讨论超市老板娘的例子，除非它被明确用来解释一个直播策略）。\n' ' - 直播间的日常互动和喊话（例如“帮老王点点赞”、“加个灯牌”），除非它被用作一个具体策略的示例。\n' ' - 无法形成完整知识点的碎片化对话。\n' '4. **输出**: 返回一个只包含高价值、与主题强相关内容的文本块。' ), expected_output='一个浓缩了所有关于直播运营策略和技巧的精华文本。', agent=analyst_agent, context=[task_clean], # 明确任务依赖 llm=llm ) # 任务3: 结构化为JSON task_structure = Task( description=( '1. 分析上一步筛选出的精华文本。\n' '2. **核心要求**: 识别文本中每一个独立的知识点或Q&A对（包括自问自答形式）。对于每一个识别出的“答案”部分，严格按照以下JSON结构进行封装。一个答案封装成一个JSON对象。如果文本包含多个知识点，则输出一个JSON列表。\n' '3. **JSON模板**:\n' '```json\n' '{{\n' ' "answer_id": "一个独特的、基于内容的英文小写下划线ID，例如: pk_strategy_001",\n' ' "summary": "用一句话精准概括这个知识点的核心思想。",\n' ' "core_strategy": [\n' ' {{\n' ' "step": 1,\n' ' "title": "策略步骤的标题",\n' ' "content": "对该步骤的具体描述。"\n' ' }}\n' ' ],\n' ' "script_examples": ["从原文中提炼或总结出的可直接使用的话术示例。"],\n' ' "common_pitfalls": ["根据原文内容，总结出的常见错误做法或思维误区。"],\n' ' "case_study": {{\n' ' "streamer_name": "如果原文提到，则填写主播名",\n' ' "description": "如果原文提到相关案例，则进行描述。若无，则明确说明。"\n' ' }},\n' ' "source_reference": {{\n' ' "video_title": "可以基于文件名或内容推断，例如 livestream_part_1",\n' ' "timestamp": "从原文中找到这个知识点开始的时间戳，例如 08:41"\n' ' }}\n' '}}\n' '```\n' '4. **填充指南**:\n' ' - `answer_id`: 必须创造一个有意义的ID。\n' ' - 如果某项内容（如`case_study`）在原文中完全没有提及，请在该字段的description中明确注明“原文未提及相关案例”。\n' ' - `core_strategy`可以包含多个步骤，如果知识点本身就是一个整体，则只包含一个step。\n' ' - `source_reference`中的时间戳要尽可能精确。\n' '5. **输出**: 一个包含一个或多个JSON对象的字符串，格式为JSON数组（即 `[ { ... }, { ... } ]`）。' ), expected_output='一个严格符合模板的JSON字符串数组，其中每个对象都是一个从原文提炼出的结构化知识点。', agent=architect_agent, context=[task_filter], # 明确任务依赖 llm=llm ) return [task_clean, task_filter, task_structure] # --- 主执行流程 --- def main(): input_dir = 'raw_md_files' output_dir = 'processed_json_outputs' if not os.path.exists(output_dir): os.makedirs(output_dir) md_files = [f for f in os.listdir(input_dir) if f.endswith('.md')] for md_file in md_files: print(f"\n{'='*20} 正在处理文件: {md_file} {'='*20}") input_file_path = os.path.join(input_dir, md_file) with open(input_file_path, 'r', encoding='utf-8') as f: markdown_content = f.read() # 为当前文件创建任务和Crew tasks = create_processing_tasks(jargon_context) livestream_crew = Crew( agents=[cleaner_agent, analyst_agent, architect_agent], tasks=tasks, process=Process.sequential, verbose=True # verbose=2会打印出思考过程，便于调试 ) # 启动Crew，传入动态数据 result = livestream_crew.kickoff(inputs={ 'markdown_text': markdown_content, 'jargon_dictionary': jargon_context }) print(f"\n{'*'*20} 文件 {md_file} 处理完成 {'*'*20}") print("最终输出结果:") print(result) # 保存结果 output_filename = os.path.splitext(md_file)[0] + '.json' output_file_path = os.path.join(output_dir, output_filename) with open(output_file_path, 'w', encoding='utf-8') as f: # 从CrewOutput对象中提取原始字符串输出 final_json_string = result.raw # (可选但推荐) 清理可能存在的多余标记，确保文件是纯净的JSON if final_json_string.startswith("```json"): final_json_string = final_json_string[7:] if final_json_string.endswith("```"): final_json_string = final_json_string[:-3] final_json_string = final_json_string.strip() f.write(final_json_string) # <--- 使用 .raw 属性 print(f"结果已保存至: {output_file_path}") if __name__ == '__main__': main()