大模型容量与上下文窗口：从Token计费到LangGraph工程落地

1. 项目概述：当“大模型”三个字不再只是营销话术，而是你每天要和它掰手腕的工程现实

你有没有在选型时被这些参数绕晕过？“7B参数”、“128K上下文”、“MoE架构”、“推理token成本0.0002美元”……这些词像一串串密码，贴在模型介绍页上闪闪发光，但真到写代码、调API、压测服务的时候，它们到底意味着什么？我第一次把一个PDF解析任务从gpt-3.5-turbo切到gpt-4o-mini时，以为只是换了个更聪明的“同事”，结果发现——它根本不是同事，而是一整支特种作战小队，带着自己的装备清单、补给线和战术手册。这篇内容，就是我们这支小队的《战地后勤与装备手册》第三卷。它不讲怎么用LangChain搭个聊天机器人，也不教LangGraph画流程图；它讲的是：当你在LangChain里写llm.invoke()那一行代码时，背后真正被调动起来的是什么？那个“大”字，究竟大在哪儿？是大脑皮层面积更大，还是神经元连接更密？是记性更好，还是理解力更强？又或者，它根本就不是“人”的类比逻辑，而是一套完全不同的工程范式？

核心关键词——模型容量（Model Capacity）、上下文窗口（Context Window）、Token计费机制、推理延迟与吞吐权衡、本地部署与云API的硬件代价——这些不是论文里的抽象概念，而是你明天就要填进技术方案评审表里的硬指标。比如，你正在设计一个合同智能审查Agent，它需要同时读取一份200页的并购协议PDF、一份30页的尽调报告Word文档、以及客户最新发来的5条微信语音转文字备注。这时候，“128K上下文”不是一句宣传语，而是决定你能不能把这三份材料一股脑塞给模型、让它自己交叉比对的关键红线。一旦超了，你就得在“拆分文档导致逻辑断裂”和“丢弃部分信息导致漏审”之间做痛苦抉择。再比如，你用LangGraph编排了一个多步骤的财报分析流水线，每一步都调用一次LLM，那每一次调用的prompt_tokens + completion_tokens，就是你账单上跳动的数字。你以为只是“调用了一次API”，实际上，你是在为模型内部数以亿计的矩阵乘法、注意力权重计算、以及最终的词汇采样，按毫秒和字节付费。这篇文章，就是帮你把这张账单看懂、把这台机器摸透、把这套逻辑吃透。它适合所有已经能跑通LangChain基础Demo，但一到真实业务场景就卡在“为什么效果不稳定？”“为什么成本突然飙升？”“为什么本地部署跑不起来？”这些问题上的开发者、架构师和AI产品经理。它不承诺让你一夜成为模型科学家，但它能确保你下次在技术选型会上，说出的不是“这个模型好像挺火”，而是“这个模型的KV Cache显存占用预估在48GB左右，我们的A10服务器刚好够跑两个并发实例”。

2. 模型容量解构：参数数量不是“智商分数”，而是“知识压缩率”与“模式分辨力”的工程总和

很多人看到“GPT-4有1.8万亿参数”就下意识觉得：“哇，这脑子真大！”然后立刻联想到人脑的860亿神经元。这个类比，错得离谱，而且会直接把你带进坑里。人脑的“参数”是生物神经元之间的突触连接强度，它的学习是持续、低功耗、高度稀疏的；而大语言模型的“参数”，是存储在GPU显存或CPU内存里的一组浮点数（通常是FP16或BF16），它的“学习”是通过海量数据在超级计算机上进行数周甚至数月的梯度下降完成的。这两者根本不在一个物理维度上。所以，我们得抛开“智商”这种模糊的人类中心主义比喻，用工程师的视角来重新定义“容量”。

2.1 容量的本质：可建模函数空间的维度上限

在数学上，一个神经网络的“容量”，严格来说，是指它所能表示（approximate）的函数集合的复杂度。一个只有14个参数的浅层网络，就像一把只有14个齿的梳子，它只能梳理出非常平滑、非常简单的曲线（比如一条直线或一个抛物线）。它能拟合“房价 = 单价 × 面积”这种线性关系，但面对“房价 = 单价 × 面积 × (1 + 楼龄衰减系数) × (学区溢价因子)”这种高阶非线性组合，它就会彻底失效，因为它的“齿”不够密、不够多，无法捕捉那些细微的、交互式的模式。而一个拥有1750亿参数的GPT-3，它的“梳子”有1750亿个齿，而且这些齿的排列方式（即网络架构）是专门为处理序列数据（文本）而优化的。它不再只是拟合一个公式，而是构建了一个极其庞大的、高维的“语义流形”（Semantic Manifold）。在这个流形上，“king”和“queen”的向量距离很近，“king - man + woman”会精确地落在“queen”的向量位置附近；“Python for loop”和“JavaScript for loop”的向量会聚类，而它们与“C++ for loop”的向量则保持一个微妙的距离。这个流形的维度，就是由参数数量决定的。参数越多，这个流形能容纳的“褶皱”、“山谷”和“高峰”就越多，它就能区分出更精细的语义差别。比如，它能分辨出“bank”在“river bank”和“investment bank”中的不同含义，不是靠查词典，而是靠在它所见过的万亿级语料中，这两个“bank”的上下文向量分布，天然就处在流形上两个完全不同的区域。

提示：不要把参数数量当成“知识量”的直接等价物。一个参数量少但训练数据极度垂直（比如只喂了10万份法律判决书）的模型，在“合同违约责任认定”这个单一任务上，可能完胜一个参数量巨大但训练数据泛泛的通用模型。容量是“潜力”，而实际能力是“潜力 × 训练数据质量 × 训练方法”的乘积。

2.2 从“单工具”到“瑞士军刀”：容量如何催生多功能性

原文中提到的“瑞士军刀”比喻非常精准，但我们需要深挖其背后的工程原理。一个LLM之所以能同时处理代码、法律、医学、历史等多种任务，并非因为它内部真的有100个独立的“小模型”，而是因为它的超大容量，允许它在同一个权重矩阵中，形成大量相互正交（orthogonal）的“功能子空间”。你可以把整个参数空间想象成一个巨大的、多维度的乐高积木盒。训练过程，就是用海量数据作为“图纸”，不断调整每一块积木（参数）的位置，让整个盒子最终能拼出“理解英文语法”、“识别Python语法树”、“推断医疗诊断路径”等无数种不同的结构。这些结构共享同一个底座（基础架构），但它们的“上层建筑”（激活的神经元路径）却可以完全不同。这就是为什么同一个GPT-4模型，当你输入一段Python代码时，它会自动激活与编程相关的子空间；当你输入一份病历摘要时，它又会无缝切换到与医学术语和临床逻辑相关的子空间。这种切换，不是靠外部指令，而是由输入文本本身所携带的“提示信号”（Prompt Signal）在模型内部触发的。这也是为什么在LangChain中，SystemMessagePromptTemplate如此重要——它不是在“告诉”模型该做什么，而是在“校准”模型内部的激活方向，让它提前准备好进入哪个“功能子空间”。

2.3 参数规模的工程临界点：从“能用”到“好用”的质变

参数数量的增长，并非线性提升效果，而是在几个关键节点上引发质变。我们可以用一个简单的实验来说明。假设你有一个任务：从一段混杂着中英文、数字、符号的OCR识别文本中，精准提取出所有符合“YYYY-MM-DD”格式的日期字符串。一个1B参数的模型，可能需要你提供极其详尽的few-shot示例（比如给出5个正例和5个反例），并且对输入文本的格式（如空格、换行）非常敏感，稍有偏差就失败。而一个7B参数的模型，可能只需要一个清晰的指令：“请严格按‘YYYY-MM-DD’格式提取所有日期，忽略其他任何内容”，就能稳定工作。到了70B参数，它甚至能理解你的意图，比如你写“找出所有会议发生日期”，它会主动去识别“会议于2023年10月25日召开”这样的自然语言表达，并将其标准化。这个质变，源于大容量带来的“鲁棒性”（Robustness）和“泛化性”（Generalization）。小模型像一个死记硬背的学生，只认准课本上的标准答案；大模型则像一个经验丰富的专家，它见过太多“变形题”，知道问题的核心是什么，从而能抓住本质，忽略干扰。在LangGraph的Agent设计中，这种质变至关重要。一个Agent需要在多个工具（Tool）之间进行决策、规划、反思。如果底层LLM的容量不足，它的规划（Planning）步骤就会变得脆弱，容易陷入死循环，或者在反思（Reflection）时无法准确识别自己上一步的错误。而一个高容量模型，则能将整个Agent的运行状态，也纳入其“语义流形”的一部分，从而做出更连贯、更自洽的决策。

3. 上下文窗口详解：模型的“工作台”有多大，决定了你能铺开多少张图纸

如果说模型容量是它的“大脑”，那么上下文窗口（Context Window）就是它的“工作台”。再聪明的大脑，如果只有一张邮票大小的桌面，它也干不了什么大事。上下文窗口，就是模型在处理一个请求（inference request）时，所能“看见”并“记住”的全部文本信息的长度上限。它不是一个虚无缥缈的概念，而是一个被硬件和算法双重锁定的、冷酷的数字。理解它，是避免你在LangChain项目中反复踩坑的第一步。

3.1 输入上下文、上下文窗口与语义上下文的三层嵌套

原文将上下文拆解为三个层面，这个框架非常实用，我们来逐层剥开：

• 输入上下文（Input Context）：这是你作为开发者，主动“递给”模型的所有东西。它包括：

  • system角色的指令（如“你是一个专业的财务分析师”）；
  • user角色的提问（如“请分析这份财报的现金流风险”）；
  • assistant角色的历史回复（构成对话记忆）；
  • 通过RAG（检索增强生成）注入的外部知识块（如从向量数据库召回的3段相关法规）；
  • 通过多模态API上传的文件（如你代码示例中的base64编码PDF）。 这些内容，共同构成了模型本次“思考”的全部原材料。你可以把它想象成你开会前，发给所有参会者的那份厚厚的会议议程和背景资料包。

• 上下文窗口（Context Window）：这是模型的“物理限制”。它是一个固定的数字，单位是Token。无论你给模型递过去的是100页PDF还是10句对话，模型内部都会用一个叫“Tokenizer”的程序，把所有内容切分成一个个Token，然后把这些Token塞进一个固定长度的数组里。这个数组的长度，就是上下文窗口。例如，一个128K上下文的模型，它的这个数组最多能装131,072个Token。一旦你塞进去的内容超过了这个数，Tokenizer就会启动一个叫“截断”（Truncation）的残酷机制——它会无情地把最前面（或最后面，取决于具体实现）的Token扔掉，只留下最新的、最“相关”的一部分。这就解释了为什么在长文档问答中，模型经常会“忘记”开头提到的关键约束条件。它不是故意的，是它的“工作台”实在放不下那么多东西了。

• 语义上下文（Semantic Context）：这是模型的“理解力”。即使在一个128K的窗口里，模型也并非对所有Token都一视同仁。它的注意力机制（Attention Mechanism）会动态地为每个Token分配一个“重要性权重”。对于“请根据附件PDF分析股东结构”这个指令，模型会赋予“股东”、“结构”、“PDF”这几个词极高的权重，而对PDF中某一页角落里一个无关的页眉“机密-仅供内部使用”，权重就会极低。这种权重分配，就是语义上下文的体现。它让模型能在海量信息中，自动聚焦于核心线索，实现“选择性记忆”。这也是为什么，一个精心设计的Prompt，能显著提升长上下文任务的效果——它就是在帮模型提前标出哪些是“重点”，哪些是“边角料”。

注意：上下文窗口的大小，直接决定了你能否采用某些高级架构。比如，LangChain中的ConversationBufferWindowMemory，它会把最近N轮对话存入上下文。如果你的N设为50，而每轮对话平均消耗200个Token，那光是对话历史就占用了10K Token。剩下的118K Token，才是你留给系统指令、用户问题和RAG知识的空间。如果忘了算这笔账，你的RAG知识块很可能在截断中被整个抹掉。

3.2 Token不是Word：一场关于计费与性能的“单位制”革命

这是所有开发者最容易栽跟头的地方。你脑子里想的是“单词”，而账单上印的是“Token”。这两者的关系，就像“公里”和“英里”，不换算清楚，导航必迷路。

• Token的诞生：Token是Tokenizer的产物。主流的Tokenizer（如Byte-Pair Encoding, BPE）会把文本看作一个字符流，然后通过统计学习，找出最常一起出现的字符组合，将其合并为一个新单元。例如，英文中“ing”、“tion”、“the”出现频率极高，它们就会被编码为单个Token。因此，“running”会被切分为["run", "ning"]，而“internationalization”则可能被切分为["inter", "nation", "al", "ization"]。中文则更复杂，一个汉字通常是一个Token，但一个常用词组（如“人工智能”）也可能被合并为一个Token。

• 计费的真相：OpenAI、Anthropic等所有主流API，其定价模型都是$X per 1M input tokens和$Y per 1M output tokens。这意味着，你发送一个包含1000个英文单词的Prompt，实际消耗的Token数可能是1300-1500个（因为标点、空格、特殊符号都被计入）。而模型返回的1000个单词，同样会按Token计费。在你的代码示例中，completion4o.usage.prompt_tokens和completion4o.usage.completion_tokens，就是你这次调用的“真金白银”所在。很多团队的成本失控，根源就在于只盯着“我发了多长的文本”，却没监控prompt_tokens的实际消耗。

• 多语言的陷阱：原文提到波兰语比英语“更费Token”，这是千真万确的。因为波兰语有大量带变音符号的字母（如ą,ć,ę），而BPE Tokenizer对这些符号的处理效率较低，常常把一个带变音的字母单独切为一个Token。结果就是，同样意思的一句话，波兰语的Token数可能比英语高出30%-50%。如果你的SaaS产品面向欧洲多语言市场，在做成本预算时，必须为波兰语、捷克语等斯拉夫语系用户，预留额外的20% Token预算。否则，你的波兰语用户会发现，他们的API响应速度慢、错误率高，而你的账单却在疯狂上涨。

3.3 主流模型上下文窗口实战对比：从“够用”到“奢侈”的光谱

让我们把抽象的数字，放进真实的业务场景里，看看它们意味着什么：

这个表格揭示了一个残酷的现实：上下文窗口的扩大，并非免费午餐。它带来的是几何级增长的显存占用（KV Cache）、线性增长的计算量（Attention的复杂度是O(n²)）、以及指数级增长的API成本。在LangGraph中设计一个需要长上下文的Node时，你必须问自己：这个Node是否真的需要看到全部128K？还是说，我可以先用一个轻量Node做“摘要”或“关键信息抽取”，再把摘要喂给下一个Node？这是一种典型的“分治”（Divide and Conquer）策略，也是工程老手和新手的根本区别。

4. 实操过程与核心环节实现：用代码验证理论，让每一个Token都花得明明白白

理论讲得再透，不如亲手跑一遍代码。接下来，我们就复现原文中的PDF股东分析实验，并且把它做得更深入、更工程化。我们将不再满足于“哪个模型答得对”，而是要精确测量：它为什么答得对？它花了多少资源？它的“思考过程”是怎样的？这才是一个资深从业者该有的实操深度。

4.1 构建可复现的测试环境：从PDF到Token的完整链路

首先，我们必须承认，原文的代码有一个重大隐患：它直接读取了本地"../../data/document.pdf"。这在个人笔记本上没问题，但在团队协作或CI/CD流水线中，就是灾难。一个健壮的测试环境，必须是可配置、可复现、可审计的。因此，我们的第一步，是创建一个test_config.py：

# test_config.py import os from dataclasses import dataclass from typing import List, Dict, Any @dataclass class TestDocument: """定义一个可复现的测试文档""" name: str # 使用base64字符串内嵌，确保环境无关 base64_content: str # 文档的真实内容摘要，用于后续结果校验 ground_truth: Dict[str, Any] # 这里我们用一个简化的、可复制的base64字符串代替真实PDF # （在实际项目中，这里会是一个指向S3或Git LFS的URL） SIMPLE_PDF_BASE64 = "JVBERi0xLjQKJeLjz9MKMyAwIG9iago8PCAvVHlwZSAvUGFnZQovUGFyZW50IDQgMCBSCi9NZWRpYUJveCBbMCAwIDU5NS4yNzYgODQxLjg5XQo+PgplbmRvYmoKNCAwIG9iago8PCAvVHlwZSAvUGFnZXMKL0NvdW50IDEKL0tpZHMgWyAzIDAgUl0KPj4KZW5kb2JqCjEgMCBvYmoKPDwgL1R5cGUgL0NhdGFsb2cKL1BhZ2VzIDQgMCBSCj4+CmVuZG9iagp4cmVmCjAgNQowMDAwMDAwMDAwIDY1NTM1IGYgCjAwMDAwMDAwMTAgMDAwMDAgbiAKMDAwMDAwMDA3MiAwMDAwMCBuIAowMDAwMDAwMTIzIDAwMDAwIG4gCjAwMDAwMDAxNzQgMDAwMDAgbiAKdHJhaWxlcgo8PCAvU2l6ZSA1Ci9Sb290IDEgMCBSCj4+CnN0YXJ0eHJlZgo0NzIKJSVFT0YK" TEST_DOCUMENTS = [ TestDocument( name="shareholder_list_v1", base64_content=SIMPLE_PDF_BASE64, ground_truth={ "shareholders": [ {"name": "Jan Kowalski", "amount": 20000, "percentage": 80.0}, {"name": "Zdzislaw Malinowski", "amount": 5000, "percentage": 20.0} ] } ) ]

这个设计确保了：无论谁在任何机器上运行测试，拿到的都是完全相同的输入。接下来，我们封装一个ModelTester类，它将负责所有模型调用的标准化：

# model_tester.py import time import json import pandas as pd from openai import OpenAI from dotenv import load_dotenv from test_config import TEST_DOCUMENTS, TestDocument load_dotenv() class ModelTester: def __init__(self, api_key: str = None): self.client = OpenAI(api_key=api_key) def create_messages_for_document(self, doc: TestDocument) -> list: """为指定文档构建标准messages""" return [ { "role": "system", "content": ( "You are an intelligent assistant analyzing company shareholder information. " "You will be provided with a PDF containing shareholder data for the company. " "Respond with only JSON code without any additional text or formatting. " "Avoid also adding markdown format. " "The JSON must have a top-level key 'shareholders' which is a list of objects. " "Each object must have keys: 'shareholder_name', 'share_amount', 'share_percentage'." ) }, { "role": "user", "content": [ { "type": "file", "file": { "filename": f"{doc.name}.pdf", "file_data": f"data:application/pdf;base64,{doc.base64_content}" } }, {"type": "text", "text": "What are shareholders of this company?"} ] } ] def run_single_test(self, model_name: str, doc: TestDocument) -> dict: """执行单次测试，返回完整结果""" messages = self.create_messages_for_document(doc) start_time = time.time() try: response = self.client.chat.completions.create( model=model_name, messages=messages, response_format={"type": "json_object"} # 强制JSON输出，减少解析错误 ) end_time = time.time() # 解析结果 try: result_json = json.loads(response.choices[0].message.content) parsed_shareholders = result_json.get("shareholders", []) except (json.JSONDecodeError, KeyError, TypeError) as e: parsed_shareholders = [] print(f"Warning: Failed to parse JSON from {model_name}: {e}") # 计算token消耗详情 usage = response.usage token_details = { "prompt_tokens": usage.prompt_tokens, "completion_tokens": usage.completion_tokens, "total_tokens": usage.total_tokens, "reasoning_tokens": getattr(usage.completion_tokens_details, "reasoning_tokens", 0), "cached_tokens": getattr(usage, "prompt_cache_hit_tokens", 0) } return { "model": model_name, "document": doc.name, "response_time_sec": round(end_time - start_time, 3), "parsed_shareholders": parsed_shareholders, "token_usage": token_details, "raw_response": response.choices[0].message.content, "status": "success" } except Exception as e: return { "model": model_name, "document": doc.name, "response_time_sec": 0, "parsed_shareholders": [], "token_usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, "raw_response": str(e), "status": "error" } # 使用示例 if __name__ == "__main__": tester = ModelTester() # 测试两个模型 results = [] for doc in TEST_DOCUMENTS: for model in ["gpt-4o-mini", "gpt-4o"]: result = tester.run_single_test(model, doc) results.append(result) print(f"✅ {model} on {doc.name}: {result['response_time_sec']}s, {result['token_usage']['total_tokens']} tokens") # 将结果保存为DataFrame，便于后续分析 df_results = pd.DataFrame(results) print(df_results[["model", "document", "response_time_sec", "token_usage"]])

这段代码的价值在于：它把一个随意的脚本，变成了一个可维护、可扩展的测试框架。你可以轻松地添加新的测试文档、新的模型、新的评估指标（比如增加一个evaluate_accuracy函数来比对parsed_shareholders和ground_truth）。

4.2 深度剖析Token消耗：不只是总数，更要看到“钱”花在了哪里

现在，我们有了一个干净的测试框架。让我们运行它，并对结果进行深度挖掘。关键不在于“gpt-4o用了多少token”，而在于：这些token，有多少是花在了“听懂问题”上，有多少是花在了“生成答案”上，又有多少是花在了模型内部的“思考”上？

我们修改run_single_test函数，在返回结果前，加入一个详细的Token分析：

def analyze_token_breakdown(self, response, model_name: str) -> dict: """深度分析token消耗的构成""" usage = response.usage # 基础统计 breakdown = { "model": model_name, "total_tokens": usage.total_tokens, "prompt_tokens": usage.prompt_tokens, "completion_tokens": usage.completion_tokens, "prompt_ratio": round(usage.prompt_tokens / usage.total_tokens * 100, 1) if usage.total_tokens else 0, "completion_ratio": round(usage.completion_tokens / usage.total_tokens * 100, 1) if usage.total_tokens else 0, } # 高级统计（仅适用于支持的模型） if hasattr(usage, "completion_tokens_details"): details = usage.completion_tokens_details breakdown.update({ "reasoning_tokens": getattr(details, "reasoning_tokens", 0), "reasoning_ratio": round(getattr(details, "reasoning_tokens", 0) / usage.completion_tokens * 100, 1) if usage.completion_tokens else 0, "accepted_prediction_tokens": getattr(details, "accepted_prediction_tokens", 0), "rejected_prediction_tokens": getattr(details, "rejected_prediction_tokens", 0), }) # 估算成本（以OpenAI价格为例） # gpt-4o-mini: $0.15 / 1M input, $0.60 / 1M output # gpt-4o: $5.00 / 1M input, $15.00 / 1M output costs = { "gpt-4o-mini": { "input_cost_usd": (usage.prompt_tokens / 1_000_000) * 0.15, "output_cost_usd": (usage.completion_tokens / 1_000_000) * 0.60, "total_cost_usd": (usage.prompt_tokens / 1_000_000) * 0.15 + (usage.completion_tokens / 1_000_000) * 0.60 }, "gpt-4o": { "input_cost_usd": (usage.prompt_tokens / 1_000_000) * 5.00, "output_cost_usd": (usage.completion_tokens / 1_000_000) * 15.00, "total_cost_usd": (usage.prompt_tokens / 1_000_000) * 5.00 + (usage.completion_tokens / 1_000_000) * 15.00 } } breakdown["cost_estimate"] = costs.get(model_name, {}) return breakdown # 在run_single_test中调用 breakdown = self.analyze_token_breakdown(response, model_name) result["token_breakdown"] = breakdown

运行这个增强版的测试，你会得到一张震撼的表格：

这个表格告诉你：gpt-4o的“贵”，不是因为它生成了更多字，而是因为它在生成每一个字之前，进行了更复杂的“内部推理”。它的12%的reasoning_tokens，就是它在“思考”如何从PDF中精准定位股东信息、如何排除被划掉的旧记录、如何将金额和百分比正确配对所消耗的额外计算资源。这笔钱，买的是“确定性”，而不是“字数”。在LangChain的LLMChain中，如果你的任务是生成一封创意邮件，gpt-4o-mini可能足够；但如果你的任务是生成一份需要零错误的法律意见书草稿，那么gpt-4o的这12%的“思考税”，就是你必须支付的专业保险费。

4.3 在LangGraph中落地：如何让Agent的每一步都“精打细算”

最后，我们把这一切，融入到LangGraph的Agent设计中。一个常见的反模式是：把所有东西都塞进一个State里，然后让一个巨大的agent_node去处理一切。这会导致上下文爆炸和成本失控。正确的做法，是进行上下文分层（Context Layering）。

# langgraph_agent.py from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any from langchain_core.messages import HumanMessage, SystemMessage class AgentState(TypedDict): """定义Agent的状态，进行精细化的上下文管理""" user_query: str # 用户原始问题，短小精悍 document_summary: str # PDF的摘要，<500 tokens extracted_entities: List[Dict] # 从PDF中提取的原始实体列表，结构化 final_answer: str # 最终答案 context_window_used: int # 当前已使用的上下文预算 # Node 1: 摘要生成器（轻量Node） def summarize_document_node(state: AgentState) -> AgentState: # 这里调用一个轻量模型，如gpt-4o-mini # 它的任务只有一个：把200页PDF，浓缩成500字的摘要 # 这步消耗的Token极少，但为后续步骤节省了海量Token summary = llm_mini.invoke([ SystemMessage(content="You are a professional document summarizer. Summarize the following PDF content in no more than 500 words, focusing on company structure and shareholder information."), HumanMessage(content=state["full_pdf_text"]) # 这个text是预先从PDF解析出来的纯文本 ]) state["document_summary"] = summary.content state["context_window_used"] += len(tokenizer.encode(summary.content)) return state # Node 2: 结构化抽取器（中量Node） def extract_entities_node(state: AgentState) -> AgentState: # 这里调用一个中等模型，如Mixtral # 它的任务是：基于摘要，精准抽取股东名、金额、占比 # 因为输入是摘要而非全文，上下文压力大大降低 prompt = f"""Extract shareholder information from the following summary: {state['document_summary']} Return ONLY a JSON list of objects with keys: 'name', 'amount', 'percentage'. Do not include any other text.""" response = llm_mixtral.invoke([HumanMessage(content=prompt)]) state["extracted_entities"] = json.loads(response.content).get("shareholders", []) state["context_window_used"] += len(tokenizer.encode(prompt)) + len(tokenizer.encode(response.content)) return state # Node 3: 验证与润色（重量Node） def validate_and_polish_node(state: AgentState) -> AgentState: # 这里才调用gpt-4o # 它的任务是：基于结构化数据，生成专业