用 CocoIndex 把 YouTube 播客自动变成可查询的知识图谱转录、两步 LLM 抽取、实体消歧与 SurrealDB 落地【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex播客是互联网上最丰富的专家知识来源之一一集 Lex Fridman 或 Dwarkesh Patel 的节目往往包含几十条关于人物、技术与组织的实质性论断——但它们全部被封存在数小时的音频里无法查询也无法交叉比对不同嘉宾对同一话题的说法。本文基于 CocoIndex 仓库中的完整可运行示例 examples/conversation_to_knowledge带你用普通asyncPython 与自定义类型构建一条把 YouTube 播客剧集转化为可查询知识图谱的增量流水线下载音频、说话人分离转写、LLM 结构化抽取、跨剧集实体消歧最后以图结构落地到 SurrealDB。读完本文你将掌握 CocoIndex 的声明式目标状态target state、记忆化memoization、组件component与多态关系目标等核心用法并能在自己的播客、会议纪要等语料上复刻这套方案。要构建什么一张五节点四关系的知识图谱整个知识图谱的 Schema 包含五类节点、由四类关系连接节点类型含义说明session一集播客剧集存 YouTube ID、剧集名、描述、完整转写文本、日期statement从对话中抽取的主题性论断例如 Scaling laws suggest that larger models will continue to improve.person人物实体说话人及被提及的人tech技术/工具/框架/概念实体如 Large language modelorg组织/公司实体如 OpenAI关系类型FROM → TO说明session_statementsession→statement剧集包含哪些论断person_sessionperson→session某人参与了哪集person_statementperson→statement某人说了什么论断statement_mentionsstatement→person/tech/org论断提到了哪些实体多态目标每条statement都关联谁说的与提到了什么。最棘手的问题在于同一个真实实体在不同剧集中会出现不同拼写GPT-4、GPT4、OpenAIs GPT-4因此必须引入**实体消歧entity resolution**环节将这些名字折叠为唯一的规范节点具体方案见下文 Phase 2。流水线总览三个阶段流水线分为三个阶段与示例主文件 conv_knowledge/app.py 中app_main的组织一一对应每集会话处理Per-session processing——对每个剧集下载、转写并用 LLM 抽取元数据、说话人与论断。会话与论断立即写入 SurrealDB因为它们不需要跨剧集去重。实体消歧Entity resolution——收集所有剧集中的原始实体名用嵌入相似度 LLM 确认的方式去重。知识库创建Knowledge base creation——写入规范实体节点与全部关系。这个示例体现了 CocoIndex 的核心编程模型你用原生 Python 声明转换target_state transformation(source_state)CocoIndex 负责推导出应该插入、更新和删除什么。README 与官方文档 编程指南/核心概念 都明确指出增量处理、变更追踪、受管图目标这些重活由底层的 Rust 引擎承担因此重跑流水线只会处理新增或变更的剧集而不会重放整个语料。Phase 1每集会话处理每一集都要经历一条多步流水线起点只是一个 YouTube URL。抓取音频并转写yt-dlp AssemblyAI下载与转写封装在 conv_knowledge/fetch.py 的fetch_transcript中用yt-dlp下载音频用 AssemblyAI 做带说话人分离的转写返回带 Speaker A、Speaker B 等标签的 utterances以及 YouTube 元数据coco.fn(memoTrue) async def fetch_transcript(youtube_id: str) - SessionTranscript: url fhttps://www.youtube.com/watch?v{youtube_id} with tempfile.TemporaryDirectory() as tmpdir: # Download audio via yt-dlp, convert to mp3 (FFmpeg) ydl_opts {format: bestaudio/best, outtmpl: f{tmpdir}/audio.mp3, ...} with yt_dlp.YoutubeDL(ydl_opts) as ydl: info ydl.extract_info(url, downloadTrue) # Transcribe with AssemblyAI speaker diarization transcript aai.Transcriber().transcribe( f{tmpdir}/audio.mp3, aai.TranscriptionConfig(speaker_labelsTrue) ) utterances [Utterance(speakeru.speaker, textu.text) for u in transcript.utterances] return SessionTranscript(utterancesutterances, yt_titleinfo[title], ...)实际源码中yt-dlp的下载选项更完整format: bestaudio/best配合FFmpegExtractAudio后处理器把音频转成 64kbps 的 mp3AssemblyAI 侧开启speaker_labelsTrue并显式指定speech_models[universal-3-pro]同时把yt-dlp返回的upload_dateYYYYMMDD规范化为 ISO 格式供后续抽取使用。若转写结果没有 utterances无分离信息代码会退化为单条未知说话人 utterance——这个回退分支保证了流水线不会因数据形态差异而中断。coco.fn(memoTrue)是这里的点睛之笔它把函数记忆化详见 高级主题/记忆化键。如果某个视频已经抓取并转写过重跑会直接跳过——当你在迭代下游抽取逻辑时不必反复下载数小时的音频。两步 LLM 抽取先认人再抽主张这里存在一个自举bootstrapping问题要让论断归属正确LLM 必须先知道说话人是谁——但原始转写只有 Speaker A 这样的通用标签。所以抽取分两遍进行两者共用同一个format_transcript()定义于 conv_knowledge/extract.py第一遍把所有标签保持为 (Speaker A) 形式第二遍把已识别出的说话人替换为真实姓名未识别者仍保留 (Speaker X) 形式。第一步——识别说话人并抽取元数据。用通用标签格式化转写把 YouTube 元数据作为上下文交给 LLM返回类型化的说话人识别结果。输出是 Pydantic 模型通过 instructor 强制约束代码中为instructor.from_litellm(litellm.acompletion, modeinstructor.Mode.JSON)coco.fn(memoTrue) async def extract_metadata(reformatted_transcript: str, transcript: SessionTranscript) - SessionMetadata: client instructor.from_litellm(litellm.acompletion, modeinstructor.Mode.JSON) return await client.chat.completions.create( modelcoco.use_context(LLM_MODEL), response_modelSessionMetadata, messages[{role: system, content: METADATA_PROMPT}, {role: user, content: ...}], )class SpeakerIdentification(pydantic.BaseModel): label: str # A, B name: str # Lex Fridman — unidentifiable speakers are excluded class SessionMetadata(pydantic.BaseModel): name: str description: str | None date: str | None speakers: list[SpeakerIdentification]源码在返回前还有一层防御性后处理conv_knowledge/extract.py对结果重新model_validate以恢复类身份便于 pickling并过滤掉不像是真实人名的输出_is_plausible_person_name要求名字至少包含两个词防止 LLM 输出 null、unknown 之类的占位。提示词METADATA_PROMPT要求使用维基百科风格的规范全名Lex Fridman、Sam Altman并且只收录有把握识别的人不确定就不猜。第二步——用真实姓名抽取论断。现在 Speaker A 已经变成 Lex Fridman用真实姓名重新格式化转写抽取主题性论断及其说话人、提及实体class RawStatement(pydantic.BaseModel): statement: str # Scaling laws suggest larger models will improve speakers: list[str] # [Lex Fridman] mentioned_person: list[str] # [Ilya Sutskever] mentioned_tech: list[str] # [Large language model] mentioned_org: list[str] # [OpenAI]这里有一条必须强制遵守的规则每个名字都必须是自包含self-contained的。提示词明确禁止代词he、she、说话人标签Speaker A或上下文指代the host、the interviewer——因为来自不同剧集的论断会被放在一起交叉引用脱离原始转写后he 或 the host 没有任何意义。实体的实体类型描述与规范名示例统一配置在 conv_knowledge/models.py 的ENTITY_TYPES中person/tech/org 三种类型各带llm_description与llm_examples抽取提示词会动态拼接这些配置这也是后文新增 Product 实体类型只改配置即可生效的基础。第二步同样有后处理用正则is_speaker_label剥离 LLM 漏出来的 (Speaker B) 之类标签。声明会话与论断节点抽取完成后把会话及其论断声明为 SurrealDB 中的记录。ID 来自 CocoIndex 的IdGenerator用法见 公共资源/ID 生成它是稳定的——相同输入总是产生相同 ID因此重跑绝不会产生重复记录。next_id(content)把内容折进 ID即使论断顺序变化ID 也保持稳定id_gen IdGenerator(youtube_id) session_id await id_gen.next_id() session_table.declare_record(rowSession(idsession_id, youtube_idyoutube_id, namemetadata.name, transcriptstep2_text, ...)) for stmt in stmt_extraction.statements: stmt_id await id_gen.next_id(stmt.statement) statement_table.declare_record(rowStatement(idstmt_id, statementstmt.statement)) session_statement_rel.declare_relation(from_idsession_id, to_idstmt_id)IdGenerator在process_session内部以 YouTube 视频 ID 为依赖创建保证跨运行稳定Session节点还做了字段回退name回退到yt_title、date回退到yt_upload_date。每集会话作为独立的处理组件运行概念见 编程指南/处理组件通过coco.use_mount挂载、以 YouTube ID 为键——所以新增一集只处理那一集raw await coco.use_mount( coco.component_subpath(session, youtube_id), process_session, youtube_id, session_table, statement_table, session_statement_rel, )process_session返回原始实体名与论断关联信息SessionRawEntities供 Phase 2 和 Phase 3 使用会话与论断此刻已经写入 SurrealDB原始实体则被携带到下一步做去重。Phase 2实体消歧embedding 相似度 LLM 确认现在手里是从每集收集来的一堆原始名字同一实体有各种拼写GPT-4 vs GPT4、Sam Altman vs Samuel Altman。CocoIndex 内置了 entity_resolution 工具其策略是先给每个名字算嵌入向量用向量相似度找出近似匹配再只对相近的候选对请 LLM 确认——廉价的嵌入负责粗筛昂贵的 LLM 调用只发生在真正模糊的地方coco.fn(memoTrue) async def _resolve_entities(all_raw_entities: set[str]) - dict[str, str | None]: result await resolve_entities( entitiesall_raw_entities, embeddercoco.use_context(EMBEDDER), # Snowflake/snowflake-arctic-embed-xs resolve_pairLlmPairResolver(modelcoco.use_context(RESOLUTION_LLM_MODEL)), ) return result.to_dict() # {Apple Inc.: None, Apple: Apple Inc., AAPL: Apple Inc.}resolve_entities的实现位于 python/cocoindex/ops/entity_resolution/init.py采用冒泡式处理每个实体计算记忆化的嵌入向量并 L2 归一化查询已建索引的最近邻IndexFlatIP对归一化向量做内积等价于余弦相似度把超过阈值的近邻收集为候选再调用LlmPairResolver见 python/cocoindex/ops/entity_resolution/llm_resolver.py让 LLM 决定这些候选是否与当前实体是同一个。默认阈值参数在源码签名中明确给出max_distance: float 0.3即余弦相似度 0.7 才进入候选、top_n: int 5候选数上限。嵌入器在 lifespan 中提供为SentenceTransformerEmbedder(Snowflake/snowflake-arctic-embed-xs)详见 conv_knowledge/app.py。消歧按实体类型独立进行因此 CocoIndex 可以让 person、tech、org 三种类型并发处理entity_dedup dict(zip( [cfg.name for cfg in ENTITY_TYPES], await asyncio.gather(*( coco.use_mount(coco.component_subpath(resolve, cfg.name), _resolve_entities, _collect_all_raw(all_session_raw, cfg.name)) for cfg in ENTITY_TYPES )), ))用于确认的 LLM 通常用更小的模型即可通过RESOLUTION_LLM_MODEL配置进一步压低成本。最终输出的dedup字典形如{Apple Inc.: None, Apple: Apple Inc., AAPL: Apple Inc.}——None表示该名字自身就是规范名字符串表示它折叠到的规范名。Phase 3知识库创建拿到去重映射后写入最终图。规范实体成为节点每条关系都使用解析后的规范名。resolve_canonical(name, dedup)会沿去重链一路追到根——resolve_canonical(AAPL, dedup)→Apple Inc.helper 定义在 conv_knowledge/models.pycoco.fn async def create_knowledge_base(all_session_raw, entity_dedup, entity_tables, ...): # Canonical entity nodes (name is the id) for cfg in ENTITY_TYPES: for name, upstream in entity_dedup[cfg.name].items(): if upstream is None: # this name is canonical entity_tables[cfg.name].declare_record(rowEntity(idname, namename)) # Relationships, using canonical names for session_raw in all_session_raw: for stmt in session_raw.statements: for cfg in ENTITY_TYPES: dedup entity_dedup[cfg.name] for canonical in {resolve_canonical(e, dedup) for e in getattr(stmt.raw, fmentioned_{cfg.name})}: statement_mentions_rel.declare_relation( from_idstmt.id, to_idcanonical, to_tableentity_tables[cfg.name]) # polymorphic target完整实现conv_knowledge/app.py 的create_knowledge_base还会写出person_session识别出的说话人参加剧集与person_statement说话人作出论断两类关系并对同一声明人去重后再建边。statement_mentions关系是**多态polymorphic**的——它的目标可以是 person、tech 或 org 表——to_table告诉 CocoIndex 目标 ID 属于哪张表。目标侧在app_main中一次性挂载statement_mentions_rel await surrealdb.mount_relation_target( SURREAL_DB, statement_mentions, statement_table, [entity_tables[cfg.name] for cfg in ENTITY_TYPES], # polymorphic TO )从 SurrealDB 连接器的实现看python/cocoindex/connectors/surrealdb/_target.py 中的mount_relation_target/relation_targetTO 侧接受单表或表集合_normalize_table_names会把集合归一化为有序的表名列表——这正是多态关系目标得以表达的地方。所有表目标由mount_table_target挂载session/statement/person/tech/org 五张表共用TableSchema.from_class(...)从 dataclass 推导的 SchemaEntity节点统一以规范名为 ID可读且无需哈希。增量更新添加剧集与演进 Schema这不是一次性任务——你会在持续时间里不断添加剧集、演进 Schema。CocoIndex 的记忆化与组件模型让两者都高效。添加剧集。新 URL 会重跑流水线但只有新剧集被处理fetch_transcript与两步抽取对已有剧集全部命中记忆化缓存实体消歧复用缓存的嵌入与决策只为真正的新名字发起新的 LLM 调用声明式目标对其余部分做 diff。移除剧集则删除其组件——对应的 session、statement 与关系会自动从 SurrealDB 中清掉无需手写迁移脚本。演进 Schema。假设你要新增一个Product实体类型流水线步骤会发生什么原因抓取转写Fetch transcript复用已记忆化输入未变第一步说话人识别复用提示词未变第二步论断抽取重跑抽取提示词变了实体消歧person/tech/org复用原始实体未变实体消歧product全新执行新类型知识库创建重新声明新节点 新关系昂贵的操作——下载、转写、说话人识别——被完全复用。给 50 集播客新增一个实体类型你只需要重跑论断抽取调用以及为新类型做消歧。值得一提的细节是LLM_MODEL、RESOLUTION_LLM_MODEL、EMBEDDER等上下文键都声明了detect_changeTrue见 conv_knowledge/models.py 与 conv_knowledge/app.py意味着模型或嵌入器一换依赖它们的记忆化调用会自动失效重算——Schema 演进之外模型升级同样可以被增量机制感知。运行整个流水线前置依赖Python 3.11、FFmpegcocoindex[surrealdb,sentence_transformers]1.0.7、yt-dlp、assemblyai、instructor、litellm、faiss-cpu等。第一步启动 SurrealDBDockerdocker run -d --name surrealdb --user root -p 8787:8000 \ -v surrealdb-data:/data surrealdb/surrealdb:latest \ start --user root --pass root surrealkv:/data/database第二步设置环境变量。必填两个 Key其余均有默认值默认值取自 conv_knowledge/app.py 的 lifespan 实现变量默认值说明ASSEMBLYAI_API_KEY必填AssemblyAI 转录带说话人分离OPENAI_API_KEY必填LLM 抽取经 LiteLLMSURREALDB_URLws://localhost:8787/rpcSurrealDB RPC 端点SURREALDB_NScocoindexSurrealDB 命名空间SURREALDB_DByt_conversationsSurrealDB 数据库SURREALDB_USER/SURREALDB_PASSroot/root认证凭据INPUT_DIR./input输入目录LLM_MODELopenai/gpt-5-mini抽取模型RESOLUTION_LLM_MODELopenai/gpt-5-mini消歧确认模型export ASSEMBLYAI_API_KEY... export OPENAI_API_KEYsk-... pip install -e .第三步准备输入。在input/sample.txt中按行添加 YouTube URL#开头为注释仓库自带的 input/sample.txt 已含 10 条示例链接。URL 的剧集 ID 由正则(?:youtube\.com/watch\?v|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})提取见 conv_knowledge/app.py支持watch?v、短链、embed 三种形态输入文件通过localfs.walk_dir按**/*.txt模式读取。第四步构建图谱——增量执行重跑会跳过已处理的剧集cocoindex update conv_knowledge.app在 Surrealist 中探索与查询SurrealDB 自带可视化工具 Surrealist。连接到ws://localhost:8787命名空间cocoindex数据库yt_conversations图视图即可展示人物蓝色与 TA 说过的论断粉色之间的person_statement边。也可以直接跑分析型查询——例如哪些技术被最多不同的说话人提及覆盖所有剧集SELECT name, array::len(array::distinct( -statement_mentions-statement-person_statement-person.id )) AS person_count FROM tech ORDER BY person_count DESC LIMIT 15;再如-- All statements a person made SELECT -person_statement-person.name AS speaker, statement FROM statement; -- Everything involved in each statement SELECT statement, -statement_mentions-person.name AS persons, -statement_mentions-tech.name AS techs, -statement_mentions-org.name AS orgs FROM statement;注意最后一条查询正是statement_mentions多态关系的直接受益者一条查询即可沿同一条边类型访问三种不同的目标表。源码导览与延伸阅读完整可运行示例位于 examples/conversation_to_knowledge/conv_knowledge/app.py——应用入口、coco_lifespan、app_main、三个阶段编排、知识库创建conv_knowledge/fetch.py——yt-dlp 下载 AssemblyAI 说话人分离转写conv_knowledge/extract.py——instructor LiteLLM 的两步抽取与提示词、转写格式化conv_knowledge/models.py——Pydantic 抽取模型、dataclass 实体模型、实体类型配置、resolve_canonicaldesign.md——完整设计文档技术选型、Schema、流水线伪代码README.md——快速上手与特性说明底层机制可继续深入仓库实体消歧实现见 python/cocoindex/ops/entity_resolution/init.py 与 llm_resolver.pySurrealDB 连接器的表/关系目标挂载见 python/cocoindex/connectors/surrealdb/_target.py相关官方文档包括 核心概念、目标状态、处理组件、App 与 use_mount、记忆化键、ID 生成、LiteLLM 封装 与 实体消歧。这套转写 → 两步 LLM 抽取 → 嵌入消歧 → 声明式图落地的骨架不限于播客换成会议纪要、访谈档案或任何对话类语料只要替换数据源与抽取提示词即可复用它把不可查询的音频文本沉淀为结构化、可增量维护的知识图谱。【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 CocoIndex 把 YouTube 播客自动变成可查询的知识图谱：转录、两步 LLM 抽取、实体消歧与 SurrealDB 落地
用 CocoIndex 把 YouTube 播客自动变成可查询的知识图谱转录、两步 LLM 抽取、实体消歧与 SurrealDB 落地【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex播客是互联网上最丰富的专家知识来源之一一集 Lex Fridman 或 Dwarkesh Patel 的节目往往包含几十条关于人物、技术与组织的实质性论断——但它们全部被封存在数小时的音频里无法查询也无法交叉比对不同嘉宾对同一话题的说法。本文基于 CocoIndex 仓库中的完整可运行示例 examples/conversation_to_knowledge带你用普通asyncPython 与自定义类型构建一条把 YouTube 播客剧集转化为可查询知识图谱的增量流水线下载音频、说话人分离转写、LLM 结构化抽取、跨剧集实体消歧最后以图结构落地到 SurrealDB。读完本文你将掌握 CocoIndex 的声明式目标状态target state、记忆化memoization、组件component与多态关系目标等核心用法并能在自己的播客、会议纪要等语料上复刻这套方案。要构建什么一张五节点四关系的知识图谱整个知识图谱的 Schema 包含五类节点、由四类关系连接节点类型含义说明session一集播客剧集存 YouTube ID、剧集名、描述、完整转写文本、日期statement从对话中抽取的主题性论断例如 Scaling laws suggest that larger models will continue to improve.person人物实体说话人及被提及的人tech技术/工具/框架/概念实体如 Large language modelorg组织/公司实体如 OpenAI关系类型FROM → TO说明session_statementsession→statement剧集包含哪些论断person_sessionperson→session某人参与了哪集person_statementperson→statement某人说了什么论断statement_mentionsstatement→person/tech/org论断提到了哪些实体多态目标每条statement都关联谁说的与提到了什么。最棘手的问题在于同一个真实实体在不同剧集中会出现不同拼写GPT-4、GPT4、OpenAIs GPT-4因此必须引入**实体消歧entity resolution**环节将这些名字折叠为唯一的规范节点具体方案见下文 Phase 2。流水线总览三个阶段流水线分为三个阶段与示例主文件 conv_knowledge/app.py 中app_main的组织一一对应每集会话处理Per-session processing——对每个剧集下载、转写并用 LLM 抽取元数据、说话人与论断。会话与论断立即写入 SurrealDB因为它们不需要跨剧集去重。实体消歧Entity resolution——收集所有剧集中的原始实体名用嵌入相似度 LLM 确认的方式去重。知识库创建Knowledge base creation——写入规范实体节点与全部关系。这个示例体现了 CocoIndex 的核心编程模型你用原生 Python 声明转换target_state transformation(source_state)CocoIndex 负责推导出应该插入、更新和删除什么。README 与官方文档 编程指南/核心概念 都明确指出增量处理、变更追踪、受管图目标这些重活由底层的 Rust 引擎承担因此重跑流水线只会处理新增或变更的剧集而不会重放整个语料。Phase 1每集会话处理每一集都要经历一条多步流水线起点只是一个 YouTube URL。抓取音频并转写yt-dlp AssemblyAI下载与转写封装在 conv_knowledge/fetch.py 的fetch_transcript中用yt-dlp下载音频用 AssemblyAI 做带说话人分离的转写返回带 Speaker A、Speaker B 等标签的 utterances以及 YouTube 元数据coco.fn(memoTrue) async def fetch_transcript(youtube_id: str) - SessionTranscript: url fhttps://www.youtube.com/watch?v{youtube_id} with tempfile.TemporaryDirectory() as tmpdir: # Download audio via yt-dlp, convert to mp3 (FFmpeg) ydl_opts {format: bestaudio/best, outtmpl: f{tmpdir}/audio.mp3, ...} with yt_dlp.YoutubeDL(ydl_opts) as ydl: info ydl.extract_info(url, downloadTrue) # Transcribe with AssemblyAI speaker diarization transcript aai.Transcriber().transcribe( f{tmpdir}/audio.mp3, aai.TranscriptionConfig(speaker_labelsTrue) ) utterances [Utterance(speakeru.speaker, textu.text) for u in transcript.utterances] return SessionTranscript(utterancesutterances, yt_titleinfo[title], ...)实际源码中yt-dlp的下载选项更完整format: bestaudio/best配合FFmpegExtractAudio后处理器把音频转成 64kbps 的 mp3AssemblyAI 侧开启speaker_labelsTrue并显式指定speech_models[universal-3-pro]同时把yt-dlp返回的upload_dateYYYYMMDD规范化为 ISO 格式供后续抽取使用。若转写结果没有 utterances无分离信息代码会退化为单条未知说话人 utterance——这个回退分支保证了流水线不会因数据形态差异而中断。coco.fn(memoTrue)是这里的点睛之笔它把函数记忆化详见 高级主题/记忆化键。如果某个视频已经抓取并转写过重跑会直接跳过——当你在迭代下游抽取逻辑时不必反复下载数小时的音频。两步 LLM 抽取先认人再抽主张这里存在一个自举bootstrapping问题要让论断归属正确LLM 必须先知道说话人是谁——但原始转写只有 Speaker A 这样的通用标签。所以抽取分两遍进行两者共用同一个format_transcript()定义于 conv_knowledge/extract.py第一遍把所有标签保持为 (Speaker A) 形式第二遍把已识别出的说话人替换为真实姓名未识别者仍保留 (Speaker X) 形式。第一步——识别说话人并抽取元数据。用通用标签格式化转写把 YouTube 元数据作为上下文交给 LLM返回类型化的说话人识别结果。输出是 Pydantic 模型通过 instructor 强制约束代码中为instructor.from_litellm(litellm.acompletion, modeinstructor.Mode.JSON)coco.fn(memoTrue) async def extract_metadata(reformatted_transcript: str, transcript: SessionTranscript) - SessionMetadata: client instructor.from_litellm(litellm.acompletion, modeinstructor.Mode.JSON) return await client.chat.completions.create( modelcoco.use_context(LLM_MODEL), response_modelSessionMetadata, messages[{role: system, content: METADATA_PROMPT}, {role: user, content: ...}], )class SpeakerIdentification(pydantic.BaseModel): label: str # A, B name: str # Lex Fridman — unidentifiable speakers are excluded class SessionMetadata(pydantic.BaseModel): name: str description: str | None date: str | None speakers: list[SpeakerIdentification]源码在返回前还有一层防御性后处理conv_knowledge/extract.py对结果重新model_validate以恢复类身份便于 pickling并过滤掉不像是真实人名的输出_is_plausible_person_name要求名字至少包含两个词防止 LLM 输出 null、unknown 之类的占位。提示词METADATA_PROMPT要求使用维基百科风格的规范全名Lex Fridman、Sam Altman并且只收录有把握识别的人不确定就不猜。第二步——用真实姓名抽取论断。现在 Speaker A 已经变成 Lex Fridman用真实姓名重新格式化转写抽取主题性论断及其说话人、提及实体class RawStatement(pydantic.BaseModel): statement: str # Scaling laws suggest larger models will improve speakers: list[str] # [Lex Fridman] mentioned_person: list[str] # [Ilya Sutskever] mentioned_tech: list[str] # [Large language model] mentioned_org: list[str] # [OpenAI]这里有一条必须强制遵守的规则每个名字都必须是自包含self-contained的。提示词明确禁止代词he、she、说话人标签Speaker A或上下文指代the host、the interviewer——因为来自不同剧集的论断会被放在一起交叉引用脱离原始转写后he 或 the host 没有任何意义。实体的实体类型描述与规范名示例统一配置在 conv_knowledge/models.py 的ENTITY_TYPES中person/tech/org 三种类型各带llm_description与llm_examples抽取提示词会动态拼接这些配置这也是后文新增 Product 实体类型只改配置即可生效的基础。第二步同样有后处理用正则is_speaker_label剥离 LLM 漏出来的 (Speaker B) 之类标签。声明会话与论断节点抽取完成后把会话及其论断声明为 SurrealDB 中的记录。ID 来自 CocoIndex 的IdGenerator用法见 公共资源/ID 生成它是稳定的——相同输入总是产生相同 ID因此重跑绝不会产生重复记录。next_id(content)把内容折进 ID即使论断顺序变化ID 也保持稳定id_gen IdGenerator(youtube_id) session_id await id_gen.next_id() session_table.declare_record(rowSession(idsession_id, youtube_idyoutube_id, namemetadata.name, transcriptstep2_text, ...)) for stmt in stmt_extraction.statements: stmt_id await id_gen.next_id(stmt.statement) statement_table.declare_record(rowStatement(idstmt_id, statementstmt.statement)) session_statement_rel.declare_relation(from_idsession_id, to_idstmt_id)IdGenerator在process_session内部以 YouTube 视频 ID 为依赖创建保证跨运行稳定Session节点还做了字段回退name回退到yt_title、date回退到yt_upload_date。每集会话作为独立的处理组件运行概念见 编程指南/处理组件通过coco.use_mount挂载、以 YouTube ID 为键——所以新增一集只处理那一集raw await coco.use_mount( coco.component_subpath(session, youtube_id), process_session, youtube_id, session_table, statement_table, session_statement_rel, )process_session返回原始实体名与论断关联信息SessionRawEntities供 Phase 2 和 Phase 3 使用会话与论断此刻已经写入 SurrealDB原始实体则被携带到下一步做去重。Phase 2实体消歧embedding 相似度 LLM 确认现在手里是从每集收集来的一堆原始名字同一实体有各种拼写GPT-4 vs GPT4、Sam Altman vs Samuel Altman。CocoIndex 内置了 entity_resolution 工具其策略是先给每个名字算嵌入向量用向量相似度找出近似匹配再只对相近的候选对请 LLM 确认——廉价的嵌入负责粗筛昂贵的 LLM 调用只发生在真正模糊的地方coco.fn(memoTrue) async def _resolve_entities(all_raw_entities: set[str]) - dict[str, str | None]: result await resolve_entities( entitiesall_raw_entities, embeddercoco.use_context(EMBEDDER), # Snowflake/snowflake-arctic-embed-xs resolve_pairLlmPairResolver(modelcoco.use_context(RESOLUTION_LLM_MODEL)), ) return result.to_dict() # {Apple Inc.: None, Apple: Apple Inc., AAPL: Apple Inc.}resolve_entities的实现位于 python/cocoindex/ops/entity_resolution/init.py采用冒泡式处理每个实体计算记忆化的嵌入向量并 L2 归一化查询已建索引的最近邻IndexFlatIP对归一化向量做内积等价于余弦相似度把超过阈值的近邻收集为候选再调用LlmPairResolver见 python/cocoindex/ops/entity_resolution/llm_resolver.py让 LLM 决定这些候选是否与当前实体是同一个。默认阈值参数在源码签名中明确给出max_distance: float 0.3即余弦相似度 0.7 才进入候选、top_n: int 5候选数上限。嵌入器在 lifespan 中提供为SentenceTransformerEmbedder(Snowflake/snowflake-arctic-embed-xs)详见 conv_knowledge/app.py。消歧按实体类型独立进行因此 CocoIndex 可以让 person、tech、org 三种类型并发处理entity_dedup dict(zip( [cfg.name for cfg in ENTITY_TYPES], await asyncio.gather(*( coco.use_mount(coco.component_subpath(resolve, cfg.name), _resolve_entities, _collect_all_raw(all_session_raw, cfg.name)) for cfg in ENTITY_TYPES )), ))用于确认的 LLM 通常用更小的模型即可通过RESOLUTION_LLM_MODEL配置进一步压低成本。最终输出的dedup字典形如{Apple Inc.: None, Apple: Apple Inc., AAPL: Apple Inc.}——None表示该名字自身就是规范名字符串表示它折叠到的规范名。Phase 3知识库创建拿到去重映射后写入最终图。规范实体成为节点每条关系都使用解析后的规范名。resolve_canonical(name, dedup)会沿去重链一路追到根——resolve_canonical(AAPL, dedup)→Apple Inc.helper 定义在 conv_knowledge/models.pycoco.fn async def create_knowledge_base(all_session_raw, entity_dedup, entity_tables, ...): # Canonical entity nodes (name is the id) for cfg in ENTITY_TYPES: for name, upstream in entity_dedup[cfg.name].items(): if upstream is None: # this name is canonical entity_tables[cfg.name].declare_record(rowEntity(idname, namename)) # Relationships, using canonical names for session_raw in all_session_raw: for stmt in session_raw.statements: for cfg in ENTITY_TYPES: dedup entity_dedup[cfg.name] for canonical in {resolve_canonical(e, dedup) for e in getattr(stmt.raw, fmentioned_{cfg.name})}: statement_mentions_rel.declare_relation( from_idstmt.id, to_idcanonical, to_tableentity_tables[cfg.name]) # polymorphic target完整实现conv_knowledge/app.py 的create_knowledge_base还会写出person_session识别出的说话人参加剧集与person_statement说话人作出论断两类关系并对同一声明人去重后再建边。statement_mentions关系是**多态polymorphic**的——它的目标可以是 person、tech 或 org 表——to_table告诉 CocoIndex 目标 ID 属于哪张表。目标侧在app_main中一次性挂载statement_mentions_rel await surrealdb.mount_relation_target( SURREAL_DB, statement_mentions, statement_table, [entity_tables[cfg.name] for cfg in ENTITY_TYPES], # polymorphic TO )从 SurrealDB 连接器的实现看python/cocoindex/connectors/surrealdb/_target.py 中的mount_relation_target/relation_targetTO 侧接受单表或表集合_normalize_table_names会把集合归一化为有序的表名列表——这正是多态关系目标得以表达的地方。所有表目标由mount_table_target挂载session/statement/person/tech/org 五张表共用TableSchema.from_class(...)从 dataclass 推导的 SchemaEntity节点统一以规范名为 ID可读且无需哈希。增量更新添加剧集与演进 Schema这不是一次性任务——你会在持续时间里不断添加剧集、演进 Schema。CocoIndex 的记忆化与组件模型让两者都高效。添加剧集。新 URL 会重跑流水线但只有新剧集被处理fetch_transcript与两步抽取对已有剧集全部命中记忆化缓存实体消歧复用缓存的嵌入与决策只为真正的新名字发起新的 LLM 调用声明式目标对其余部分做 diff。移除剧集则删除其组件——对应的 session、statement 与关系会自动从 SurrealDB 中清掉无需手写迁移脚本。演进 Schema。假设你要新增一个Product实体类型流水线步骤会发生什么原因抓取转写Fetch transcript复用已记忆化输入未变第一步说话人识别复用提示词未变第二步论断抽取重跑抽取提示词变了实体消歧person/tech/org复用原始实体未变实体消歧product全新执行新类型知识库创建重新声明新节点 新关系昂贵的操作——下载、转写、说话人识别——被完全复用。给 50 集播客新增一个实体类型你只需要重跑论断抽取调用以及为新类型做消歧。值得一提的细节是LLM_MODEL、RESOLUTION_LLM_MODEL、EMBEDDER等上下文键都声明了detect_changeTrue见 conv_knowledge/models.py 与 conv_knowledge/app.py意味着模型或嵌入器一换依赖它们的记忆化调用会自动失效重算——Schema 演进之外模型升级同样可以被增量机制感知。运行整个流水线前置依赖Python 3.11、FFmpegcocoindex[surrealdb,sentence_transformers]1.0.7、yt-dlp、assemblyai、instructor、litellm、faiss-cpu等。第一步启动 SurrealDBDockerdocker run -d --name surrealdb --user root -p 8787:8000 \ -v surrealdb-data:/data surrealdb/surrealdb:latest \ start --user root --pass root surrealkv:/data/database第二步设置环境变量。必填两个 Key其余均有默认值默认值取自 conv_knowledge/app.py 的 lifespan 实现变量默认值说明ASSEMBLYAI_API_KEY必填AssemblyAI 转录带说话人分离OPENAI_API_KEY必填LLM 抽取经 LiteLLMSURREALDB_URLws://localhost:8787/rpcSurrealDB RPC 端点SURREALDB_NScocoindexSurrealDB 命名空间SURREALDB_DByt_conversationsSurrealDB 数据库SURREALDB_USER/SURREALDB_PASSroot/root认证凭据INPUT_DIR./input输入目录LLM_MODELopenai/gpt-5-mini抽取模型RESOLUTION_LLM_MODELopenai/gpt-5-mini消歧确认模型export ASSEMBLYAI_API_KEY... export OPENAI_API_KEYsk-... pip install -e .第三步准备输入。在input/sample.txt中按行添加 YouTube URL#开头为注释仓库自带的 input/sample.txt 已含 10 条示例链接。URL 的剧集 ID 由正则(?:youtube\.com/watch\?v|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})提取见 conv_knowledge/app.py支持watch?v、短链、embed 三种形态输入文件通过localfs.walk_dir按**/*.txt模式读取。第四步构建图谱——增量执行重跑会跳过已处理的剧集cocoindex update conv_knowledge.app在 Surrealist 中探索与查询SurrealDB 自带可视化工具 Surrealist。连接到ws://localhost:8787命名空间cocoindex数据库yt_conversations图视图即可展示人物蓝色与 TA 说过的论断粉色之间的person_statement边。也可以直接跑分析型查询——例如哪些技术被最多不同的说话人提及覆盖所有剧集SELECT name, array::len(array::distinct( -statement_mentions-statement-person_statement-person.id )) AS person_count FROM tech ORDER BY person_count DESC LIMIT 15;再如-- All statements a person made SELECT -person_statement-person.name AS speaker, statement FROM statement; -- Everything involved in each statement SELECT statement, -statement_mentions-person.name AS persons, -statement_mentions-tech.name AS techs, -statement_mentions-org.name AS orgs FROM statement;注意最后一条查询正是statement_mentions多态关系的直接受益者一条查询即可沿同一条边类型访问三种不同的目标表。源码导览与延伸阅读完整可运行示例位于 examples/conversation_to_knowledge/conv_knowledge/app.py——应用入口、coco_lifespan、app_main、三个阶段编排、知识库创建conv_knowledge/fetch.py——yt-dlp 下载 AssemblyAI 说话人分离转写conv_knowledge/extract.py——instructor LiteLLM 的两步抽取与提示词、转写格式化conv_knowledge/models.py——Pydantic 抽取模型、dataclass 实体模型、实体类型配置、resolve_canonicaldesign.md——完整设计文档技术选型、Schema、流水线伪代码README.md——快速上手与特性说明底层机制可继续深入仓库实体消歧实现见 python/cocoindex/ops/entity_resolution/init.py 与 llm_resolver.pySurrealDB 连接器的表/关系目标挂载见 python/cocoindex/connectors/surrealdb/_target.py相关官方文档包括 核心概念、目标状态、处理组件、App 与 use_mount、记忆化键、ID 生成、LiteLLM 封装 与 实体消歧。这套转写 → 两步 LLM 抽取 → 嵌入消歧 → 声明式图落地的骨架不限于播客换成会议纪要、访谈档案或任何对话类语料只要替换数据源与抽取提示词即可复用它把不可查询的音频文本沉淀为结构化、可增量维护的知识图谱。【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考