自研 GEO 系统实战:从架构设计到“一键投喂”多平台 AI 大模型

📅 2026/7/2 8:43:55
自研 GEO 系统实战:从架构设计到“一键投喂”多平台 AI 大模型
一、为什么我要专门做一套 GEO 系统过去十几年我们习惯讨论的是 SEO怎么让自己的页面在搜索结果里排到前面。但这两年明显的变化是——用户越来越习惯直接问大模型“有什么推荐”、“哪个产品更适合我”而不是自己翻搜索结果页。当入口从“搜索结果列表”变成“AI 直接给出的答案”传统的 SEO 优化就不够用了页面排名高不代表大模型一定会引用你的内容。文章很多不代表模型真的记住了你的品牌和产品。更别说如何让模型在比较不同方案时“优先想起你”。这就是我做 GEOGenerative Engine Optimization生成式引擎优化系统的原因用工程化方式持续、可控地给大模型“投喂”企业的知识资产让它在回答用户问题时更有机会引用我们的品牌、产品和解决方案。二、总体设计思路把 GEO 系统当成一个“内容操作系统”我不把 GEO 系统当成一个简单的发文工具而是按“资产 → 语义 → 生成 → 投喂 → 监测 → 反馈”的闭环来设计。整体可以拆成五层内容资产层沉淀品牌介绍、产品参数、案例、FAQ、技术文档等原始资料统一管理。语义加工层将长文档切片、清洗抽取实体品牌、产品、行业、问题构建主题簇和问答库把企业知识转成适合模型理解的语义块。内容生成层结合模板、事实库和大模型能力生成不同形态的投喂内容技术文章、问答对、使用案例、平台短内容等。渠道分发层适配不同站点和平台的发布规则实现“一套内容多渠道投放”技术社区、博客、资讯媒体、本地站点等。效果监测层定期向主流大模型提问抓取答案分析是否出现自家品牌和产品、出现频率、位置和叙事方式再把结果反哺到下一轮优化。我更喜欢用一句话概括这不是一个工具而是一套围绕企业知识资产的自动化投喂和观测系统。三、技术栈选择为什么我用 Python 队列 检索从落地效率和生态来看技术栈我选的是前端管理后台Vue 3 / React后端 APIFastAPI / Flask异步任务Celery / RQ消息队列Redis / RabbitMQ数据库PostgreSQL检索Elasticsearch 向量检索如 pgvector / Milvus 等文档解析BeautifulSoup / pdfplumber / trafilatura模型调用兼容 OpenAI 风格的接口或自建网关原因很简单Python 在文本处理、Embedding、RAG 场景下工具齐全、社区成熟。队列非常适合“批量生成、批量分发、批量回测”这种典型异步任务。结构化数据库负责事实与关系向量库负责语义召回两者组合能更好支撑 GEO 的“事实可靠 语义相关”。四、数据库设计我实际用的 8 张核心表下面是我生产环境里基本都会有的八张核心表你可以按自己需要做精简或扩展。1. 品牌资产表brand_assetsCREATE TABLE brand_assets ( id BIGSERIAL PRIMARY KEY, brand_name VARCHAR(128) NOT NULL, asset_type VARCHAR(64) NOT NULL, -- intro/case/faq/spec/news title VARCHAR(255), source_text TEXT, source_url TEXT, lang VARCHAR(16) DEFAULT zh-CN, status VARCHAR(32) DEFAULT active, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() );2. 知识切片表knowledge_chunksCREATE TABLE knowledge_chunks ( id BIGSERIAL PRIMARY KEY, asset_id BIGINT REFERENCES brand_assets(id), chunk_index INT, chunk_text TEXT NOT NULL, token_count INT, embedding_id VARCHAR(128), topic_label VARCHAR(128), created_at TIMESTAMP DEFAULT NOW() );3. 实体表entitiesCREATE TABLE entities ( id BIGSERIAL PRIMARY KEY, entity_name VARCHAR(255) NOT NULL, entity_type VARCHAR(64) NOT NULL, -- brand/product/industry/problem alias JSONB, metadata JSONB, created_at TIMESTAMP DEFAULT NOW() );4. 问答表qa_pairsCREATE TABLE qa_pairs ( id BIGSERIAL PRIMARY KEY, question TEXT NOT NULL, answer TEXT NOT NULL, related_entity_ids JSONB, source_chunk_ids JSONB, confidence NUMERIC(5,2), created_at TIMESTAMP DEFAULT NOW() );5. 内容生成任务表content_jobsCREATE TABLE content_jobs ( id BIGSERIAL PRIMARY KEY, job_type VARCHAR(64) NOT NULL, -- article/faq/post/case target_keyword VARCHAR(255), platform VARCHAR(64), prompt_version VARCHAR(64), status VARCHAR(32) DEFAULT pending, result_content_id BIGINT, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() );6. 生成内容表generated_contentsCREATE TABLE generated_contents ( id BIGSERIAL PRIMARY KEY, title TEXT, content_markdown TEXT, summary TEXT, content_type VARCHAR(64), platform_style VARCHAR(64), quality_score NUMERIC(5,2), duplicate_score NUMERIC(5,2), created_at TIMESTAMP DEFAULT NOW() );7. 发布记录表publish_recordsCREATE TABLE publish_records ( id BIGSERIAL PRIMARY KEY, content_id BIGINT REFERENCES generated_contents(id), platform VARCHAR(64), publish_account VARCHAR(128), external_post_id VARCHAR(255), publish_status VARCHAR(32), response_payload JSONB, published_at TIMESTAMP );8. AI 观测日志表ai_observation_logsCREATE TABLE ai_observation_logs ( id BIGSERIAL PRIMARY KEY, platform VARCHAR(64), query_text TEXT NOT NULL, answer_text TEXT, brand_hit BOOLEAN, mention_score NUMERIC(5,2), citation_score NUMERIC(5,2), rank_label VARCHAR(32), observed_at TIMESTAMP DEFAULT NOW() );这八张表基本支撑了资产沉淀assets / chunks / entities / qa内容生产和任务管理jobs / contents渠道发布和状态记录publishAI 回答观测与分析observation五、内容生成服务不是“直接调用模型”那么简单生成内容这一步我的原则是召回事实 → 构建大纲 → 调用模型生成 → 质量评估 → 必要时再修正。Python 伪代码示例class ContentGenerator: def __init__(self, retriever, llm_client, scorer): self.retriever retriever # 检索知识切片 self.llm_client llm_client # 调用大模型 self.scorer scorer # 质量评分器 def generate_article(self, keyword: str, brand: str, platform: str): # 1. 从知识库召回相关事实 docs self.retriever.search(keywordkeyword, top_k12) facts [doc[text] for doc in docs] # 2. 构建文章大纲 outline [ f{keyword}是什么, f{keyword}为什么值得单独做成系统, f{keyword}的整体架构设计, f{keyword}的核心模块与代码细节, f{keyword}在实际业务中的效果评估 ] # 3. 调用模型生成草稿 prompt self._render_prompt(keyword, brand, platform, outline, facts) draft self.llm_client.generate(prompt) # 4. 做质量评估 score self.scorer.evaluate(draft, facts) # 5. 如果事实覆盖率不够再修正一轮 if score[fact_score] 0.8: draft self._repair(draft, facts) return { outline: outline, content: draft, score: score } def _render_prompt(self, keyword, brand, platform, outline, facts): return f 你是一名技术作者请围绕 GEO 系统写一篇文章 面向平台{platform}品牌{brand} 核心关键词{keyword}。 请按以下大纲写作 {outline} 必须优先使用以下事实和资料 {facts} 六、多平台分发用适配器模式解耦不同渠道不同平台的接口和风格差异很大用适配器模式会省很多坑。基础适配器抽象类from abc import ABC, abstractmethod class BasePublisher(ABC): abstractmethod def publish(self, content: dict) - dict: 发布内容返回平台的响应信息 pass abstractmethod def update(self, post_id: str, content: dict) - dict: 更新已发布内容 pass这样一来一键投喂的流程大致就是前端勾选渠道矩阵比如 CSDN、博客、公司官网。后端为每个渠道创建一个content_jobs任务。异步任务调用ContentGenerator生成内容再调PublishDispatcher发布。发布结果写入publish_records表前端可查看每个渠道的发布状态。七、AI 回测服务让系统持续知道“模型现在怎么说你”很多人做 GEO 停留在“发出去就完事”的阶段其实最关键的是观测。回测服务的三个核心指标是否出现品牌名或产品名品牌命中率。是否引用了我们投喂过的内容或结论引用率。在推荐列表或比较结果中的位置排序情况。Python 伪代码示例class AiObserver: def __init__(self, llm_clients: dict, brand_terms: list): self.llm_clients llm_clients # {deepseek: client1, platformX: client2} self.brand_terms brand_terms # [某品牌, 某产品, 某系统] def observe(self, questions: list): results [] for platform, client in self.llm_clients.items(): for q in questions: answer client.ask(q) hit any(term in answer for term in self.brand_terms) mention_score self._mention_score(answer) results.append({ platform: platform, question: q, answer: answer, brand_hit: hit, mention_score: mention_score }) return results def _mention_score(self, answer: str) - float: score 0.0 for term in self.brand_terms: if term in answer: score 1.0 return score回测结果会写入ai_observation_logs表然后我通常会做一个可视化看板不同平台、不同问题的命中率曲线。最近一次投喂后的变化趋势。哪些问题目前完全没提到我们需要加强内容覆盖。八、异步任务与调度为什么“一键投喂”一定要走队列GEO 系统里最重的三个动作是批量解析原始资料。批量生成文章、问答和内容。批量向不同大模型发起观测请求。如果用同步接口做很容易出现请求超时、前端一直转圈、接口偶尔挂掉。所以我几乎所有关键流程都放到任务队列里通过 Celery 来调度。Celery 任务示例生成并发布from celery import Celery celery_app Celery( geo_tasks, brokerredis://localhost:6379/0, backendredis://localhost:6379/1 ) celery_app.task def generate_and_publish(keyword, brand, platform): generator build_generator() # 构造 ContentGenerator dispatcher build_dispatcher() # 构造 PublishDispatcher article generator.generate_article(keyword, brand, platform) content { title: f{keyword}实战解析, body: article[content], tags: [keyword, GEO, AI搜索优化] } result dispatcher.dispatch(platform, content) return result后端返回一个job_id所有后续过程都在后台自动完成前端通过GET /api/jobs/{job_id}看最新状态即可。九、内容质量控制避免系统变成“批量制造垃圾”的机器GEO 爆火之后一个现实问题是市场上开始出现大量低质量甚至错误的“投喂内容”。如果我们只做“机器多写点文多发一点链接”长期来看其实会伤害品牌信誉。所以我在系统里加了一个简单的质量评分器至少从三个维度去衡量事实覆盖度生成结果是否覆盖了我们知识库里最重要的事实点。重复度和历史内容以及同批次内容相比有没有过度相似或洗稿。可读性句子是否通顺、结构是否清晰、对开发者是否友好。简单的评分器示例class QualityScorer: def evaluate(self, draft: str, facts: list): fact_score self._fact_coverage(draft, facts) duplicate_score self._duplicate_check(draft) readability_score self._readability(draft) final_score ( 0.5 * fact_score 0.3 * readability_score 0.2 * (1 - duplicate_score) ) return { fact_score: fact_score, duplicate_score: duplicate_score, readability_score: readability_score, final_score: final_score } def _fact_coverage(self, draft: str, facts: list) - float: covered sum(1 for f in facts if f[:50] in draft) return covered / max(len(facts), 1) def _duplicate_check(self, draft: str) - float: # 这里可以接相似度模型或现有检测服务 return 0.2 def _readability(self, draft: str) - float: # 粗略以句长、分段、标点等计算 return 0.85对于质量评分较低的内容可以直接标红不允许自动发布必须走人工审核避免系统在无意中制造“AI 垃圾”。十、总结把 GEO 当成基础设施而不是短期投机从我的实践来看如果把 GEO 当成一个短期流量工具很容易落到“批量发文、批量铺链接”的老路上。真正值得长期投入的是把它当成企业在 AI 时代的基础设施知识资产被持续结构化和语义化。内容生成和分发流程尽可能自动化同时保留关键节点的人工审核。AI 回答观测变成例行任务而不是偶尔“去问一下看看有没有提到我们”。每一次发布、每一次观测都能反过来优化下一轮的策略和系统。从这个角度看GEO 系统的核心竞争力并不在于“我能调用哪个大模型”而在于你是否有能力把真实的业务知识用工程化方式转成 AI 持续引用的“语义基础设施”。