企业官网建设流程全解析

最近做的一个挺有意思的项目：用OpenAI SDK搭了一个支持多个Agent协同工作的系统，专门用来处理超大体量的文本任务。项目跑下来累计Token消耗达到了1000万级别，而每次请求真正用到的上下文窗口只有10万tokens左右，算下来上下文利用率大概是100倍。也就是说，我没有依赖把数据硬塞进context length，而是通过多Agent的拆解、记忆、汇总，让模型在有限的窗口里“间接看完了”远超窗口的内容。

这个项目的核心启发是：如果你还在纠结“context window不够大”，方向可能偏了。真正要解决的是“如何让有限上下文承载无限信息”。这里我结合CrewAI风格的协作分工、LangGraph式的状态流转、AutoGen式的对话收敛，用OpenAI SDK自己写了一套轻量编排层，不依赖重型框架也能跑。下面把设计思路、核心代码、踩坑过程完整列出来，给想做长文本处理或多Agent系统的朋友一个参考。

1. 项目目标与核心挑战

先明确一下这个项目要解决什么问题。假设你有一批几十万字甚至几百万字的项目文档、聊天记录、日志文件，需要让大模型做全局分析，比如提取风险点、生成摘要、按主题归类。直接调用模型接口，一次性塞进去显然不现实，因为context length有限，就算硬塞进去，超长输入会导致注意力退化，后面的内容基本就“失忆”了。而且成本也扛不住。

所以这个项目定了一个指标：让模型有效消费的总Token数达到1000万以上，但每一次实际输入模型的Token数控制在context length的可承受范围内，最终目标是倍率达到100倍左右。这个倍率不是模型能力，而是工程能力。

从技术形态上看，这个系统更像是一个“过程式框架”：一个任务被拆成多个子任务，分配给不同的Agent，每个Agent只读自己需要的那部分数据，产出结构化中间结果，最终由汇总Agent结合所有中间结果形成结论。整个过程由SDK调用驱动，状态由我们自己管理，并没有依赖某个现成的多Agent框架。

这样做的好处也很明显：不绑定特定框架，所有控制点都在自己手里，方便调节和排查。适合已经熟悉OpenAI API、想自己搭建编排逻辑的开发者。如果你是想直接拿来用，CrewAI、LangGraph、AutoGen这些库确实能省事，但当你的任务量到了千万Token这个量级，框架本身的调度、缓存、上下文管理可能反而不够灵活。自己写一套反而更可控。

2. 设计思路：如何用有限的上下文窗口消化无限的信息

核心问题只有一个：模型每次能看到的Token是固定的，怎么让它处理的数据量超出上限100倍？

2.1 先算一笔账：1000万Token意味着什么

1000万Token差不多相当于7500万个英文字符，或者1500万字中文。哪怕调用的是支持百万上下文的新型模型，单次请求也撑不住这个量级。更关键的是，很多任务并不需要模型一次性看到全部内容。比如要分析一年份的客服会话记录，模型不需要同时读完全部一万条记录才能统计高频问题。它只需要分批阅读，把每条记录的结构化信息抽出来，然后聚合统计。

所以我一开始就把目标从“扩大context length”改成“缩小每个子任务的信息密度”。这句话是整个系统的设计基石。

2.2 多Agent的本质：多个视角、多批读取、一次聚合

CrewAI的典型设计是给不同角色分配不同职责，比如研究者、写作者、审校者。AutoGen则更强调两个Agent之间的对话式协作，一个出题一个解题。LangGraph提供的是图状态流，可以在节点之间传递数据。

我借鉴了三者的共同点：任务按阶段拆分，每个阶段由一个专用的Agent执行，Agent之间不直接传原始文本，而是传结构化摘要和结论。这样每个Agent的输入窗口都只装下“当前阶段最需要的信息”，而全局信息被保存在外部存储中，按需取用。这就是“100倍利用率”的来源。

2.3 三种编排模式的取舍与融合

在多Agent系统的实现上，业界主要有三种模式：Pipeline、群聊、状态机。我最终用了一个混合方案：主流程是状态机，子任务内部是Pipeline，跨Agent信息同步用群聊事件广播。直接上一张对比表：

在代码层面，这些模式最终都归结为“循环调用OpenAI SDK，把前一步输出注入下一步的System Message或User Message”。理解了这一点，你就能绕开所有框架，自己实现。

3. 核心实现：用OpenAI SDK搭建多Agent骨架

我直接给出一个精简可运行的版本。这个版本不依赖任何多Agent框架，只使用OpenAI SDK和标准库。为了演示，假设我们要处理1000篇本地文档，总Token量约1000万，context length是10万。

3.1 任务切片与元数据登记

第一步是把大任务切片，并给每一片建立索引。这一步不需要模型参与，纯文本处理。按字符数量或Token估算进行切分，切分粒度要根据子任务复杂度调整。

import hashlib import json from openai import OpenAI client = OpenAI(api_key="your-api-key") CHUNK_TOKEN_BUDGET = 8000 # 每个子任务预算 def split_text_by_tokens(text, max_tokens=CHUNK_TOKEN_BUDGET): # 简易估算：1中文约2token，1英文约0.8token # 真实项目建议用tiktoken精确计算 parts = [] current = [] current_len = 0 for para in text.split("\n"): est = len(para) * 1.2 if current_len + est > max_tokens and current: parts.append("\n".join(current)) current, current_len = [], 0 current.append(para) current_len += est if current: parts.append("\n".join(current)) return parts # 对每篇文档切片后登记 docs_meta = [] for doc_id, raw_text in enumerate(all_docs): chunks = split_text_by_tokens(raw_text) for idx, chunk in enumerate(chunks): doc_meta = { "doc_id": doc_id, "chunk_id": idx, "text": chunk, "hash": hashlib.md5(chunk.encode()).hexdigest(), } docs_meta.append(doc_meta)

为什么要先做切片和登记？因为后续Agent需要按需检索，而不是把全部文本预加载。登记信息里的hash可以用来做缓存，后续如果同一段文本需要多次分析，可以直接命中历史结果，节省大量Token。

3.2 Agent基类：统一输入输出协议

所有Agent都使用相同的调用接口，内部只是System Prompt不同。这是让多Agent协作可维护的关键。如果每个Agent函数长得都不一样，后续扩展会非常痛苦。

class BaseAgent: def __init__(self, name, system_prompt): self.name = name self.system_prompt = system_prompt def run(self, user_content: str, **kwargs) -> str: messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_content}, ] resp = client.chat.completions.create( model="gpt-4o-mini", messages=messages, temperature=0.2, **kwargs ) return resp.choices[0].message.content

注意我这里没有传max_tokens。实际生产里一定要传，并且要根据任务类型设置不同上限。不设max_tokens可能导致模型输出超长，打爆下游Agent的context。

3.3 实现Pipeline：摘要Agent链

一个简单的摘要链路包括“分块摘要Agent”和“合并摘要Agent”。分块摘要把8000Token的文档片段压缩成300Token的结构化要点，合并摘要再把所有要点汇成总摘要。

summarizer_prompt = """ 你是一个文档分析摘要员。输入是文本片段。 请提取以下信息，使用JSON格式返回： - key_points: 要点列表（不超过5条，每条不超过30字） - risk_flags: 风险标记（如无则为空列表） - entities: 重要实体列表 不要输出任何多余解释。 """ merge_prompt = """ 你是一个总摘要员。输入是多个片段摘要组成的列表。 请合并这些摘要，生成一段不超过500字的全局摘要。 保留冲突信息，去除重复内容。 """ def pipeline_summarize(doc_id): chunks = [m for m in docs_meta if m["doc_id"] == doc_id] partial_summaries = [] for chunk in chunks: agent = BaseAgent(f"chunk_summarizer_{doc_id}", summarizer_prompt) raw = agent.run(chunk["text"], response_format={"type": "json_object"}) partial_summaries.append(raw) agent = BaseAgent("merge_summarizer", merge_prompt) return agent.run(json.dumps(partial_summaries[:20], ensure_ascii=False))

真实项目中不可能一次把1000个片段的摘要全部塞给合并Agent，所以需要引入“多轮合并”：每20个片段摘要合并一次，得到中间摘要，中间摘要再合并，直到产出最终摘要。这个递归合并结构就是LangGraph里常见的节点循环，只是我用普通函数实现。

3.4 实现状态机：任务路由与状态流转

为了让流程具备可扩展性，我用一个字典表示当前状态，在每一次Agent调用后会更新状态。类似LangGraph的StateGraph。

state = { "stage": "parse", "pending_docs": [0, 1, 2, ..., 999], "summaries": {}, "logs": [], } def run_state_machine(state, max_iterations=500): iteration = 0 while state["stage"] != "done" and iteration < max_iterations: iteration += 1 if state["stage"] == "parse": # 调用文档解析Agent，简单模拟 doc_id = state["pending_docs"].pop() state["summaries"][doc_id] = pipeline_summarize(doc_id) if not state["pending_docs"]: state["stage"] = "analyze" elif state["stage"] == "analyze": # 全局风险分析Agent all_sums = list(state["summaries"].values()) risk_agent = BaseAgent("risk_analyzer", risk_prompt) state["final_report"] = risk_agent.run(json.dumps(all_sums[:50])) state["stage"] = "done" # 记录日志 state["logs"].append({"iter": iteration, "stage": state["stage"]}) return state

实际项目中，这个状态还需要持久化到外部存储，比如SQLite或Redis。因为1000万Token的处理不是一次就能跑完的，必然涉及断点续跑。每处理完一个子任务就写入一条状态记录，进程崩溃后可以从最近的检查点恢复，而不是从头再来。

3.5 实现群聊协作：让不同角色互相质疑

单一路径虽然清晰，但容易出现“信息茧房”。比如风险分析Agent可能漏掉某个细节，这时候可以引入一个“质疑Agent”，对最终报告提出疑问，再把疑问返回给分析Agent进行修正。这类似AutoGen的多轮对话，只不过我没有让Agent自由发言，而是用固定的“质疑-修订”循环，保证收敛。

critic_prompt = """ 你是审查员。下面有一份分析报告。 请找出其中的逻辑漏洞、信息缺口、过度推断。 只输出问题列表，不要修改报告。 """ reviser_prompt = """ 你是修订员。下面有一份报告以及审查员的问题。 请结合问题修订报告，输出最终版本。 如果问题确实不成立，保留原文并说明原因。 """ def critic_revise_loop(report, max_rounds=3): current = report for round_idx in range(max_rounds): critic = BaseAgent("critic", critic_prompt) issues = critic.run(current) # 如果没有实质问题，就停止 if "无问题" in issues or len(issues.strip()) < 20: break reviser = BaseAgent("reviser", reviser_prompt) current = reviser.run(f"报告：{current}\n问题：{issues}") return current

关键点在于每轮对话都必须显式传入前一轮的上下文，而且上下文只保留必要部分，不要把历史对话全部塞进去。这样Agent的输入窗口永远保持紧凑。

4. Token利用率的优化细节

真正让Token利用率提升到100倍的，是下面几个细节。这些细节决定了系统是真能处理千万Token，还是只是能跑通Demo。

4.1 Token预算分层

我给每一类Agent分配了不同的Token预算：

这样做有一个额外好处：你可以在代码里加一个“Token计数器”，每次调用前估算输入Token，如果超过预算就自动截断或拆分。简单估算可以用len(text.split())乘以一个系数，但要精确建议用tiktoken库。

import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") def count_tokens(text): return len(enc.encode(text)) def fit_budget(text, budget): if count_tokens(text) <= budget: return text # 从后往前裁剪并加标记 tokens = enc.encode(text) return enc.decode(tokens[:budget]) + "...[truncated]"

这里有个经验：长文本截断时保留开头比保留结尾好，因为摘要任务里开头往往包含核心信息。如果两端都重要，那就不该截断，而是继续拆分子任务。

4.2 上下文压缩链

所谓100倍利用率，本质上就是“每读一次原文只产生少量Token，层层压缩，最终保留的精华信息远小于原文本量”。举个例子：

• 原文总量：10,000,000 Tokens
• 第一次分块摘要：每8,000 Token片段 -> 300 Token摘要，共1,250条摘要，总计375,000 Tokens
• 第二次合并摘要：每20条摘要 -> 500 Token中间摘要，共63条中间摘要，总计31,500 Tokens
• 第三次汇总：63条中间摘要 -> 最终报告2,000 Tokens

整个过程中，模型真正处理的有效输入Token大约是10,000,000 + 375,000 + 31,500 + 2,000，远大于输出，但关键在于没有哪一次请求超过10,000 Token输入。这就是“利用率”的数学基础。

需要注意的是，摘要压缩会损失信息，尤其是不常见但对任务重要的细节。所以生产系统中我会为关键环节设置“原文回溯”机制。比如风险分析Agent在觉得信息不足时，可以通过工具调用去查询原始片段，而不是只依赖摘要。这就是工具调用的意义。

4.3 用工具调用实现按需检索

OpenAI SDK支持tools参数，让模型自主决定是否调用外部函数。我把“原文检索”封装成一个工具，Agent在汇总时如果需要看某个具体的chunk，就会触发工具调用，由我的代码从向量库或文档Meta里取出对应文本。这样Agent永远只读取自己真正需要的那一小块原文，不会把全文装进context。

def search_docs(query): # 实际项目里可以是向量相似度检索或关键词匹配 results = [] for meta in docs_meta: if query in meta["text"]: results.append(meta["text"][:2000]) if len(results) >= 3: break return "\n---\n".join(results) tools = [ { "type": "function", "function": { "name": "search_docs", "description": "在原始文档库中搜索包含关键信息的片段", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"} }, "required": ["query"] } } } ] resp = client.chat.completions.create( model="gpt-4o-mini", messages=[...], tools=tools, tool_choice="auto" )

如果模型返回tool_calls，你就执行对应函数，然后把函数结果以tool角色消息追加到对话中，再次请求模型生成最终答案。这个模式能显著降低“失效Token”，即模型输出里跟任务无关的废话。对于千万级Token处理，每减少1%无效输出都影响成本。

4.4 防止长对话漂移

AutoGen风格的自由对话最大问题是容易跑题。我见过两个Agent互相复述“你说得对”持续十几轮，浪费大量Token。我通常用三个手段控制：

1. 每一轮对话增加“本轮目标”字段，放在System Prompt最前面。
2. 设置最大轮数，超过后强制由汇总Agent输出当前成果。
3. 对话历史只保留本轮之前最后一条消息，而不是保留全部。

def chat_coordinator(goal, agent_a, agent_b, max_rounds=4): context = goal for _ in range(max_rounds): a_reply = agent_a.run(context) b_reply = agent_b.run(f"对方说：{a_reply}\n请继续推进目标，不要复述。") context = b_reply # 只保留这一轮 if "完成" in b_reply or "结束" in b_reply: break return context

这样每一轮对话的输入都只有一条上下文，Token开销极小，且不会让模型迷失。

5. 实操过程与踩坑实录

上面是理论框架，下面是实际跑项目中遇到的几个典型问题和我的解决方案。

5.1 第一次跑通时Token统计失真

一开始我用len(text)估算Token，错误地以为1000万Token花的很慢。后来接入tiktoken精确计量，发现每篇文档的实际Token比估算多了将近40%。原因很简单，中文标点和换行在分词器里也是Token，而且英文单词会被拆成多个子词单元。这个问题直接影响任务切分的均匀性：原本以为单个chunk 8000 Token，实际可能已经超过12000 Token，导致部分请求报context length错误。

解决方式很简单：所有切分操作都统一用tiktoken计算Token数，不要用字符数估算。另外建议在切分时预留20%余量，因为后续可能因为对话历史而增加输入Token。

5.2 并行执行导致API限流

千万级Token处理必然要并发。我把分块摘要任务用线程池并发执行，结果很快触发OpenAI API的每分钟请求数限制和TPM限制。这里的经验是：

• 按文档粒度串行，文档内部并行，避免并发数过高。
• 设置一个简单的令牌桶，控制每分钟请求数不超过限制的80%。
• 对429错误做指数退避重试，退避公式为min(2^retry_count, 60) + random()*2。

import time, threading class RateLimiter: def __init__(self, max_per_minute): self.interval = 60.0 / max_per_minute self.next_available = 0 self.lock = threading.Lock() def wait(self): with self.lock: now = time.time() if now < self.next_available: time.sleep(self.next_available - now) self.next_available = time.time() + self.interval

5.3 摘要合并时丢信息

最初直接把所有摘要序列化成一个长字符串，让合并Agent一次性读取。当摘要总量超过2万Token时，Agent开始出现幻觉，比如编造不存在的实体。后来我把合并改为层次化，强制每组最多合并20条摘要，并且让合并Agent输出带引用ID的摘要，方便溯源。例如：

{"source_ids": [1,2,3], "summary": "..."}

这样后续如果发现信息错误，可以直接定位到源片段重新分析。

5.4 上下文缓存命中失败

OpenAI提供了Prompt Caching功能，如果请求的前缀完全一致，可以降低输入Token成本。但在多Agent系统里，每条请求的System Prompt不同，User Message也带动态内容，导致缓存命中率很低。改善方法是把固定部分（任务说明、Agent角色定义）放在System Prompt最前面，把动态部分放在User Message最后。这样即使动态内容变化，前部固定前缀也可能被缓存。

实测下来，缓存命中率从不到5%提升到了40%左右，千万Token级别的成本节约非常可观。

5.5 最终报告输出不稳定

汇总Agent的输出格式如果不强制JSON，会出现各种格式飘移。我后来统一在请求里加了response_format={"type": "json_object"}，并且在System Prompt里给出具体JSON示例。注意：GPT-4o系列支持强制JSON后，速度会略降，但可靠性提升明显。如果模型不支持强制JSON格式（比如某些小模型），就需要在提示词里强调“只输出JSON，不要markdown代码块”，并在后处理时剥离代码块标记。

6. 常见问题排查速查表

我把项目运行期间遇到的高频问题整理成一个表，方便后面维护的人快速参照。

这些坑几乎每个多Agent长文本项目都会遇到，与其后面Debug，不如最开始就设计好。

7. 关于Token利用率的个人体会

说到最后，我想分享一点实际感受。很多人看到“1000万Token”和“100倍context length”会觉得是炫技，但其实核心思路特别朴素：让模型每一秒都只处理当前最重要的事，其他东西先存在外面，需要的时候再拿。就像人类读一本书，不会把整本书复制进大脑再开始思考，而是一页页读、做笔记、最后写综述。

在我这个项目里，最初的版本也尝试过直接堆context length，用了当时支持最长上下文的模型，结果成本和效果都不理想。后来换成多Agent流水线，反而跑得更稳。原因是多Agent天然会强制结构化，每一步的输入输出都有明确边界，不会让模型陷入看不见全局的迷雾。

对于想复现这个项目的朋友，我的建议是从小规模开始。先拿10万Token的数据跑通，再逐步放大到100万、1000万。不要一开始就追求大，否则你会被并发、限流、状态恢复这些工程问题淹没。先把Agent的输入输出协议定义好，把Token计量做准，把断点续跑做好，大Token量反而是水到渠成的事情。