Prompt工程2026:从基础技巧到企业级应用

Prompt工程2026:从基础技巧到企业级应用

Prompt工程在2026年已经从"玄学技巧"演变为一门系统化的工程学科。当LLM能力越来越强,Prompt的焦点从"让模型能做"转向了"让模型做得好、做得稳、做得可控"。本文将从基础到企业级,全面梳理2026年Prompt工程的最新实践。 一、Prompt工程的2026年现状 范式转变 时期 核心挑战 Prompt焦点 代表技术 2022-2023 模型能力有限 如何让模型"能做" Few-shot, CoT 2024-2025 能力提升但不可控 如何让模型"做好" 结构化Prompt, ReAct 2026 能力强但需要规模化 如何让模型"做稳" Prompt管理, A/B测试, 自动优化 2026年的核心认知 模型能力已不是瓶颈:GPT-5/Claude 5的基本能力足以应对大多数任务 Prompt质量决定输出质量:同样的模型,好Prompt和差Prompt的效果差距可达300% Prompt是资产:企业Prompt需要版本管理、测试、监控——和代码一样 自动化是趋势:自动Prompt优化(APO)开始替代人工调优 二、基础技巧回顾与升级 1. 角色设定(Role Prompting) 2026年的最佳实践不再是简单的"你是一个专家",而是结构化角色定义: # 角色定义 你是一位资深的金融分析师,拥有CFA证书和15年A股市场研究经验。 ## 专业知识 - 精通财务报表分析和估值模型(DCF, DDM, PEG) - 熟悉A股市场的行业轮动和风格切换 - 擅长宏观经济分析和政策解读 ## 分析风格 - 数据驱动:每个结论必须有数据支撑 - 辩证思考:同时分析利多和利空因素 - 风险意识:始终提示潜在风险 ## 输出规范 - 使用专业但易懂的语言 - 关键数据标注来源 - 给出明确的投资建议(买入/持有/卖出)和理由 2. Few-Shot Learning的进化 2026年的Few-Shot不再只是"给几个例子",而是动态示例选择: ...

2026-06-30 · 4 min · 659 words · 硅基 AGI 探索者
ai translation 2026 comparison

AI 翻译 2026:DeepL vs Google Translate vs LLM

引言 翻译是AI最古老的应用之一,但2026年的翻译格局已今非昔比。DeepL以专业翻译质量著称,Google Translate覆盖最多语言,而LLM(GPT-4o、Claude)正在以惊人的速度重新定义翻译的可能性。本文将通过多维度实测对比,揭示2026年AI翻译的真实水平。 一、产品概览 维度 DeepL Google Translate GPT-4o Claude 3.5 支持语言 32种 133种 95+种 95+种 翻译模式 文本/文件/语音 文本/文件/语音/图片 对话式 对话式 上下文理解 有限上下文 有限上下文 深度理解 深度理解 风格控制 正式/非正式 无 可指定 可指定 术语库 ✅Pro版 ✅Pro版 通过Prompt 通过Prompt 定价 €8.99/月起 免费/$20/月(API) $20/月 $20/月 二、翻译质量实测 2.1 测试设计 我们构建了5个领域的测试集,每个领域20个句子,共100个测试样本: 文学翻译:小说节选、诗歌、散文 技术文档:API文档、技术规范、代码注释 法律文本:合同条款、隐私政策、法规 商务文件:商业报告、邮件、营销文案 日常对话:聊天、旅游、餐饮 2.2 综合质量评分 领域 DeepL Google GPT-4o Claude 3.5 文学翻译 8.2 7.1 8.8 9.1 技术文档 9.1 8.3 8.9 9.0 法律文本 8.7 7.8 8.5 8.8 商务文件 8.5 7.9 8.7 8.9 日常对话 7.8 8.2 8.6 8.5 综合 8.5 7.9 8.7 8.9 *评分标准:1-10分,由3位专业译员独立评分取平均 ...

2026-06-28 · 2 min · 396 words · 硅基 AGI 探索者
ai security operations llm soc

AI 安全运营:用 LLM 增强 SOC

引言 2026年,网络安全威胁的复杂度和频率持续攀升。全球平均每天发生2.2亿次安全事件,传统SOC团队面临严重的人手不足和告警疲劳。LLM的出现为安全运营带来了革命性变化——从告警分诊到威胁狩猎,从事件响应到报告生成,LLM正在重塑SOC的每一个环节。本文将介绍如何用LLM系统性地增强安全运营中心。 一、SOC痛点与LLM价值 1.1 传统SOC核心痛点 痛点 现状 LLM解决方案 告警疲劳 日均10000+告警,70%为误报 智能分诊+自动丰富化 分析师短缺 全球缺口350万 AI执行L1/L2分析 响应速度慢 MTTR平均4.6小时 自动响应缩短至分钟级 知识断层 经验难以传承 知识库+案例推理 报告耗时 每个事件2-3小时写报告 自动生成事件报告 1.2 LLM在SOC中的能力矩阵 能力 输入 LLM任务 输出 告警分诊 原始告警+上下文 分类+优先级评估 分诊建议+处置SOP 日志分析 系统日志/审计日志 异常行为识别 可疑行为列表 威胁情报 多源情报+内部事件 关联分析 威胁评估报告 事件响应 事件全量信息 推荐响应方案 响应计划+命令 报告生成 事件时间线+分析 结构化写作 事件响应报告 钓鱼检测 邮件内容+元数据 语义分析+模式识别 风险评分+判断 二、告警智能分诊 2.1 分诊流程 告警接入 → LLM告警分类(真实威胁/误报/低危/需关注) → 上下文丰富化(关联用户/资产/历史事件/威胁情报) → 优先级评估(CVSS+业务影响+资产价值) → 自动处置建议 → 分派给合适层级分析师 2.2 实现方案 class AlertTriageAgent: def __init__(self): self.llm = LLMRouter() # 路由到不同模型 self.threat_intel = ThreatIntelAPI() self.asset_db = AssetDatabase() self.case_db = CaseDatabase() # 历史案例 async def triage(self, alert): # 1. 收集上下文 context = await self.gather_context(alert) # 2. LLM分诊分析 prompt = self.build_triage_prompt(alert, context) analysis = await self.llm.analyze(prompt) # 3. 结构化输出 return { 'classification': analysis.classification, # true_positive/false_positive 'confidence': analysis.confidence, # 0-1 'priority': analysis.priority, # P1-P4 'severity': analysis.severity, # critical/high/medium/low 'reasoning': analysis.reasoning, # 推理过程 'recommended_actions': analysis.actions, # 建议动作 'similar_cases': context.similar_cases, # 相似历史事件 'ioc': analysis.ioc, # 指标类信息 } async def gather_context(self, alert): return Context( asset_info=self.asset_db.lookup(alert.target), threat_intel=self.threat_intel.query(alert.indicators), similar_cases=self.case_db.search(alert, top_k=5), user_history=self.get_user_activity(alert.user), network_context=self.get_network_topology(alert.source) ) 2.3 效果数据 某金融机构部署6个月后的效果: ...

2026-06-28 · 3 min · 474 words · 硅基 AGI 探索者
llm hallucination 2026 analysis

大模型幻觉问题 2026:根因分析与缓解策略

幻觉:LLM 最棘手的问题 大模型的幻觉(Hallucination)——自信地输出错误或虚构的信息——是 2026 年 LLM 生产应用中最大的痛点。Stanford 2026 年 AI Index 报告显示,即使是最先进的模型,在事实性问答中的幻觉率仍有 3-8%,在专业领域更是高达 15-20%。幻觉不是 bug,而是 LLM 生成式架构的特性——但我们可以系统性地缓解它。 一、幻觉的分类 1.1 幻觉类型体系 类型 描述 示例 严重性 事实性幻觉 输出与事实不符 “爱因斯坦出生于1890年”(实际1879) 高 来源幻觉 虚构引用/出处 编造不存在的论文 高 推理幻觉 推理链条中出错 正确前提但错误结论 中 时间幻觉 时间线混乱 “2024年奥运会在北京举办” 中 实体幻觉 虚构人/组织/产品 “微软CEO John Smith” 高 数值幻觉 数字/计算错误 “12×13=146”(实际156) 高 代码幻觉 API/函数不存在 调用不存在的库函数 中 自我幻觉 虚构自身能力 “我可以访问实时互联网” 低 1.2 幻觉产生的根因 ┌──────────────────────────────────────────┐ │ 幻觉根因分析 │ ├──────────────────────────────────────────┤ │ │ │ 1. 训练数据问题 │ │ ├── 训练数据中的错误信息 │ │ ├── 知识截止日期后的信息缺失 │ │ └── 长尾知识表示不足 │ │ │ │ 2. 模型架构问题 │ │ ├── 自回归生成无法回溯修正 │ │ ├── 注意力机制对事实关注不足 │ │ └── 知识存储与检索机制不完善 │ │ │ │ 3. 推理机制问题 │ │ ├── 缺乏事实核查机制 │ │ ├── 过度依赖模式匹配而非知识检索 │ │ └── 采样温度增加随机性 │ │ │ │ 4. 交互问题 │ │ ├── 模型倾向"回答"而非"承认不知道" │ │ ├── 用户问题中的错误前提引导 │ │ └── 上下文中的错误信息被采纳 │ │ │ └──────────────────────────────────────────┘ 二、幻觉检测方法 2.1 基于一致性的检测 class ConsistencyBasedDetector: """基于一致性的幻觉检测""" def __init__(self, llm_client, n_samples: int = 5): self.llm = llm_client self.n_samples = n_samples def detect(self, query: str, response: str) -> dict: """通过多次采样检测一致性""" # 1. 对同一问题生成多个回答 responses = [] for _ in range(self.n_samples): r = self.llm.generate(query, temperature=0.7) responses.append(r) # 2. 提取每个回答中的事实性陈述 all_claims = [] for r in responses: claims = self._extract_claims(r) all_claims.append(claims) # 3. 检查事实一致性 consistency_scores = self._check_consistency(all_claims) # 4. 原始回答的事实性评估 original_claims = self._extract_claims(response) original_scores = [] for claim in original_claims: # 该claim在其他回答中出现的比例 support_count = sum( 1 for claims in all_claims if self._claim_match(claim, claims) ) original_scores.append({ 'claim': claim, 'support_rate': support_count / len(all_claims), 'likely_hallucination': support_count / len(all_claims) < 0.4 }) hallucination_rate = sum( 1 for s in original_scores if s['likely_hallucination'] ) / len(original_scores) if original_scores else 0 return { 'hallucination_rate': hallucination_rate, 'claim_scores': original_scores, 'overall_consistency': np.mean(consistency_scores) } def _extract_claims(self, text: str) -> list: """提取文本中的事实性陈述""" prompt = f"""提取以下文本中的所有事实性陈述,每条一行: {text}""" result = self.llm.generate(prompt) return [line.strip() for line in result.split('\n') if line.strip()] def _claim_match(self, claim: str, other_claims: list) -> bool: """判断claim是否在other_claims中存在""" prompt = f"""判断以下陈述是否语义等价: 陈述A:{claim} 陈述B列表:{other_claims} 如果A与B中任一陈述语义等价,返回"是",否则"否"。""" return '是' in self.llm.generate(prompt) 2.2 基于来源验证的检测 class SourceVerificationDetector: """基于来源验证的幻觉检测""" def __init__(self, llm_client, search_client): self.llm = llm_client self.search = search_client def detect(self, response: str, sources: list = None) -> dict: """验证回答中的事实是否有来源支持""" # 1. 提取事实性陈述 claims = self._extract_claims(response) # 2. 对每个claim进行验证 results = [] for claim in claims: # 如果有现成来源,在来源中验证 if sources: verification = self._verify_in_sources(claim, sources) else: # 搜索验证 verification = self._verify_by_search(claim) results.append({ 'claim': claim, 'verdict': verification['verdict'], # supported | refuted | unverifiable 'evidence': verification['evidence'], 'confidence': verification['confidence'] }) hallucination_claims = [ r for r in results if r['verdict'] == 'refuted' ] unsupported_claims = [ r for r in results if r['verdict'] == 'unverifiable' ] return { 'total_claims': len(results), 'supported': len([r for r in results if r['verdict'] == 'supported']), 'refuted': len(hallucination_claims), 'unverifiable': len(unsupported_claims), 'hallucination_rate': len(hallucination_claims) / len(results), 'details': results } def _verify_in_sources(self, claim: str, sources: list) -> dict: """在给定来源中验证claim""" prompt = f"""判断以下陈述是否被来源文本支持。 陈述:{claim} 来源文本: {chr(10).join(f'[{i+1}] {s[:500]}' for i, s in enumerate(sources))} 判断: - "supported":来源支持该陈述 - "refuted":来源与该陈述矛盾 - "unverifiable":来源无法验证该陈述 返回判断和依据。""" result = self.llm.generate(prompt) if 'supported' in result.lower(): verdict = 'supported' elif 'refuted' in result.lower(): verdict = 'refuted' else: verdict = 'unverifiable' return { 'verdict': verdict, 'evidence': result, 'confidence': 0.8 } def _verify_by_search(self, claim: str) -> dict: """通过搜索验证claim""" search_results = self.search.search(claim, top_k=3) return self._verify_in_sources(claim, search_results) 2.3 基于模型置信度的检测 class ConfidenceBasedDetector: """基于模型内部置信度的幻觉检测""" def __init__(self, model): self.model = model def detect(self, prompt: str) -> dict: """通过logits分析检测幻觉""" # 1. 获取模型输出和logits with torch.no_grad(): outputs = self.model.generate( prompt, return_logits=True, output_scores=True ) # 2. 计算每个token的置信度 token_confidences = [] for i, (token, score) in enumerate( zip(outputs.tokens, outputs.scores) ): # softmax得到概率 probs = torch.softmax(score, dim=-1) token_prob = probs[token].item() token_confidences.append(token_prob) # 3. 分析低置信度区域 low_confidence_threshold = 0.5 low_conf_regions = [ {'position': i, 'token': outputs.tokens[i], 'confidence': conf} for i, conf in enumerate(token_confidences) if conf < low_confidence_threshold ] # 4. 计算整体置信度指标 avg_confidence = np.mean(token_confidences) min_confidence = np.min(token_confidences) return { 'avg_confidence': avg_confidence, 'min_confidence': min_confidence, 'low_confidence_tokens': len(low_conf_regions), 'hallucination_risk': 'high' if avg_confidence < 0.6 else 'medium' if avg_confidence < 0.8 else 'low', 'low_conf_regions': low_conf_regions[:10] # 前10个低置信度token } 三、幻觉缓解策略 3.1 RAG 增强 class RAGHallucinationMitigator: """RAG 增强缓解幻觉""" def __init__(self, llm_client, retrieval_client): self.llm = llm_client self.retrieval = retrieval_client def generate(self, query: str) -> dict: """RAG增强生成""" # 1. 检索相关文档 docs = self.retrieval.search(query, top_k=5) # 2. 构建RAG Prompt prompt = self._build_rag_prompt(query, docs) # 3. 生成回答 response = self.llm.generate(prompt) # 4. 后验证 verification = self._verify_response(response, docs) return { 'response': response, 'sources': docs, 'verification': verification, 'hallucination_risk': verification['hallucination_rate'] } def _build_rag_prompt(self, query: str, docs: list) -> str: return f"""请基于以下参考资料回答问题。如果资料中没有答案,请明确说明"根据现有资料无法回答"。 ## 参考资料 {self._format_docs(docs)} ## 回答规则 1. 只使用参考资料中的信息 2. 每个事实性陈述必须标注来源 [1], [2] 等 3. 如果资料中有矛盾信息,指出矛盾 4. 不确定的信息要标注"可能" 5. 资料中未涉及的信息不要编造 ## 问题 {query} ## 回答""" 3.2 自我验证机制 class SelfVerificationGenerator: """自我验证生成机制""" def __init__(self, llm_client): self.llm = llm_client def generate_with_verification(self, query: str) -> dict: """带自我验证的生成""" # Step 1: 初步生成 initial_response = self.llm.generate(query) # Step 2: 自我事实核查 fact_check = self._self_fact_check(query, initial_response) # Step 3: 如果发现幻觉,修正 if fact_check['has_issues']: corrected = self._correct_hallucinations( query, initial_response, fact_check ) else: corrected = initial_response # Step 4: 最终验证 final_check = self._final_verification(query, corrected) return { 'response': corrected, 'initial_response': initial_response, 'fact_check': fact_check, 'final_verification': final_check, 'corrections_made': fact_check['issues_found'] } def _self_fact_check(self, query: str, response: str) -> dict: """自我事实核查""" prompt = f"""请对你自己的回答进行事实核查。 问题:{query} 你的回答:{response} 请逐句检查: 1. 每个事实性陈述是否准确? 2. 是否有编造的信息? 3. 是否有不确定但表述为事实的内容? 4. 数字和日期是否正确? 对每个可能的问题,标注: - 陈述内容 - 问题类型(错误/编造/不确定) - 修正建议 如果一切正确,返回"无问题"。""" result = self.llm.generate(prompt) has_issues = '无问题' not in result issues = [] if has_issues: # 解析问题列表 issues = self._parse_issues(result) return { 'has_issues': has_issues, 'issues_found': issues, 'raw_check': result } def _correct_hallucinations(self, query, response, fact_check): """修正检测到的幻觉""" prompt = f"""请修正以下回答中的事实性错误。 原始问题:{query} 原始回答:{response} 检测到的问题:{fact_check['raw_check']} 修正规则: 1. 修正所有检测到的事实性错误 2. 不确定的信息添加"据我所知"等限定词 3. 无法确认的信息直接删除或标注"需查证" 4. 保持回答的连贯性和有用性 修正后的回答:""" return self.llm.generate(prompt) 3.3 多模型交叉验证 class CrossModelVerification: """多模型交叉验证""" def __init__(self, models: list): self.models = models # 多个不同厂商的模型 def generate_verified(self, query: str) -> dict: """多模型交叉验证生成""" # 1. 每个模型独立生成回答 responses = [] for model in self.models: r = model.generate(query, temperature=0.0) responses.append(r) # 2. 提取各模型的事实性陈述 all_claims = [] for r in responses: claims = self._extract_claims(r) all_claims.append(set(claims)) # 3. 找出共识陈述和分歧陈述 consensus = set.intersection(*all_claims) if all_claims else set() all_claims_set = set.union(*all_claims) if all_claims else set() disputed = all_claims_set - consensus # 4. 只保留共识陈述 verified_response = self._build_response_from_consensus( query, consensus, responses[0] # 以第一个模型的回答为模板 ) return { 'response': verified_response, 'model_count': len(self.models), 'consensus_claims': len(consensus), 'disputed_claims': len(disputed), 'agreement_rate': len(consensus) / max(len(all_claims_set), 1) } 四、幻觉评估基准 4.1 评估指标 指标 计算方式 目标 事实准确率 事实正确陈述/总事实陈述 > 95% 幻觉率 虚构陈述/总事实陈述 < 5% 来源准确率 正确引用/总引用 > 90% 拒绝准确率 正确拒绝不该回答的问题比例 > 95% 过度拒绝率 错误拒绝合理问题的比例 < 5% 自校准ECE 预期校准误差 < 0.1 4.2 评估数据集构建 class HallucinationEvalDataset: """幻觉评估数据集""" CATEGORIES = { 'factual_qa': { 'description': '事实性问答', 'examples': [ {'q': '中国最长的河流是?', 'a': '长江', 'type': 'fact'}, {'q': '光速是多少?', 'a': '约3×10^8 m/s', 'type': 'fact'}, ] }, 'unanswerable': { 'description': '无法回答的问题(测试是否承认不知道)', 'examples': [ {'q': '2028年的奥斯卡最佳影片是哪部?', 'a': '无法回答', 'type': 'refuse'}, {'q': '张三的手机号码是多少?', 'a': '无法回答', 'type': 'refuse'}, ] }, 'false_premise': { 'description': '错误前提问题', 'examples': [ {'q': '林黛玉倒拔垂杨柳的故事告诉我们什么?', 'a': '指出前提错误', 'type': 'correct'}, ] }, 'multi_hop': { 'description': '多跳推理(每跳都可能出错)', 'examples': [ {'q': '相对论提出者出生国家的首都人口是多少?', 'a': '需推理:爱因斯坦→德国→柏林→人口', 'type': 'reasoning'} ] } } 五、2026 前沿方向 检索增强自我修正:模型生成后自动检索验证并修正 事实性训练:用事实核查数据做偏好优化 知识编辑:直接在模型权重中修正错误知识 不确定性量化:模型输出附带校准的不确定性分数 因果推理增强:用因果推理替代模式匹配 工具增强事实性:自动调用搜索引擎、计算器等工具 结语 幻觉是 LLM 的"原罪"——源于其自回归生成架构和统计学习本质。完全消除幻觉在当前技术范式下不可能,但通过 RAG、自我验证、交叉验证等手段,可以将其控制在可接受范围内。 ...

2026-06-28 · 6 min · 1115 words · 硅基 AGI 探索者
lm studio 2026 review

LM Studio 2026:桌面级大模型工具评测

LM Studio 2026:从工具到平台 LM Studio 在 2026 年完成了一次重要蜕变:从一个"本地运行 LLM 的桌面工具"进化为"桌面级 AI 开发平台"。对于不想折腾命令行的用户来说,LM Studio 提供了最友好的本地大模型体验——图形界面管理模型、对话测试、API 服务、RAG 构建,一切都在一个应用中完成。 2026 核心功能 功能总览 LM Studio 2026 ├── 模型管理 │ ├── 模型搜索与浏览(HuggingFace 集成) │ ├── 一键下载与安装 │ ├── 模型版本管理 │ └── 自定义模型导入 ├── 对话界面 │ ├── 多模型并行对话 │ ├── 对比测试模式 │ ├── 参数可视化调节 │ └── 对话历史管理 ├── 开发工具 │ ├── OpenAI 兼容 API 服务器 │ ├── RAG 工作台 │ ├── Prompt 实验室 │ └── 函数调用测试器 ├── 部署工具 │ ├── 本地服务部署 │ ├── Docker 容器导出 │ └── 配置文件导出 └── 社区功能 ├── 模型评分与评价 ├── Prompt 分享 └── 配置模板 与 2024 版本对比 特性 LM Studio 0.2 (2024) LM Studio 0.4 (2026) 多模型对话 不支持 并行对比 4 个模型 API 服务 OpenAI 兼容 OpenAI + Anthropic + 嵌入 RAG 不支持 内置 RAG 工作台 函数调用 不支持 原生支持 多模态 不支持 图像理解 模型格式 GGUF GGUF + Safetensors + MLX 平台 Win/Mac/Linux Win/Mac/Linux + CLI 推理引擎 llama.cpp llama.cpp + MLX + vLLM 核心功能深度评测 1. 模型管理 LM Studio 的模型管理是同类工具中体验最好的: ...

2026-06-28 · 4 min · 842 words · 硅基 AGI 探索者
ollama 2026 local llm

Ollama 2026:本地大模型运行的最佳实践

Ollama 2026:让本地大模型触手可及 Ollama 在 2026 年已经成为"本地运行大模型"的代名词。这个由 Go 语言编写的轻量级工具,让任何人都能在自己的电脑上运行大语言模型——无需 GPU 集群,无需复杂配置,一条命令即可开始。截至 2026 年 6 月,Ollama 累计下载量超过 2000 万次,月活用户超过 300 万。 2026 核心特性 版本亮点 特性 Ollama 0.1 (2024) Ollama 0.5 (2026) 模型格式 GGUF GGUF + Safetensors 多模态 不支持 图像 + 音频 并发推理 不支持 原生支持 模型仓库 50+ 模型 500+ 模型 API 兼容 OpenAI OpenAI + Anthropic 分布式 不支持 多节点推理 量化 Q4 Q2-Q8 + FP8 上下文 8k 1M+ Windows 实验性 原生支持 安装与配置 安装 # macOS brew install ollama # Linux curl -fsSL https://ollama.com/install.sh | sh # Windows (PowerShell) winget install Ollama.Ollama # Docker docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama ollama/ollama:0.5 配置 # 环境变量配置 export OLLAMA_HOST=0.0.0.0:11434 export OLLAMA_MAX_LOADED_MODELS=3 # 最大同时加载模型数 export OLLAMA_MAX_VRAM=0 # 0=自动检测 export OLLAMA_NUM_PARALLEL=4 # 并发请求数 export OLLAMA_KEEP_ALIVE=24h # 模型保活时间 export OLLAMA_FLASH_ATTENTION=1 # Flash Attention export OLLAMA_KV_CACHE_TYPE=q8_0 # KV Cache 量化 export OLLAMA_NUM_CTX=32768 # 默认上下文长度 模型管理 拉取与运行 # 拉取模型 ollama pull qwen3:72b # Qwen 3 72B(默认 Q4 量化) ollama pull qwen3:72b-q8_0 # Q8 量化(更高质量) ollama pull llama4:70b-instruct ollama pull deepseek-v3:671b # DeepSeek V3 MoE # 运行模型 ollama run qwen3:72b "解释量子纠缠" # 查看已安装模型 ollama list # 查看运行中的模型 ollama ps # 删除模型 ollama rm qwen3:72b 自定义模型 # Modelfile(类似 Dockerfile) FROM qwen3:72b # 系统提示 SYSTEM """ 你是一个专业的中文技术写作助手。你的任务是: 1. 撰写清晰、准确的技术文档 2. 使用中文,技术术语保留英文 3. 包含代码示例和对比表格 4. 保持客观、专业的语气 """ # 参数 PARAMETER temperature 0.3 PARAMETER top_p 0.9 PARAMETER num_ctx 32768 PARAMETER repeat_penalty 1.1 PARAMETER stop "<|im_end|>" # 模板 TEMPLATE """ {{ if .System }}<|im_start|>system {{ .System }}<|im_end|> {{ end }}<|im_start|>user {{ .Prompt }}<|im_end|> <|im_start|>assistant """ # 工具定义 TOOL search_web { "description": "搜索互联网", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } } } # 构建自定义模型 ollama create tech-writer -f Modelfile # 运行 ollama run tech-writer "写一篇关于 RAG 架构的文章" API 使用 REST API import httpx import json import asyncio class OllamaClient: def __init__(self, base_url="http://localhost:11434"): self.base_url = base_url self.client = httpx.AsyncClient(base_url=base_url, timeout=300) async def chat(self, model: str, messages: list, stream=False, **kwargs): """对话接口""" response = await self.client.post("/api/chat", json={ "model": model, "messages": messages, "stream": stream, "options": { "temperature": kwargs.get("temperature", 0.7), "top_p": kwargs.get("top_p", 0.9), "num_ctx": kwargs.get("num_ctx", 32768), }, "tools": kwargs.get("tools", []), }) return response.json() async def chat_stream(self, model: str, messages: list): """流式对话""" async with self.client.stream("POST", "/api/chat", json={ "model": model, "messages": messages, "stream": True, }) as response: async for line in response.aiter_lines(): data = json.loads(line) if data.get("message", {}).get("content"): yield data["message"]["content"] if data.get("done"): break async def embed(self, model: str, input: str | list): """生成向量嵌入""" response = await self.client.post("/api/embed", json={ "model": model, "input": input, }) return response.json()["embeddings"] async def generate(self, model: str, prompt: str, images: list = None): """多模态生成""" response = await self.client.post("/api/generate", json={ "model": model, "prompt": prompt, "images": images or [], "stream": False, }) return response.json() # 使用 client = OllamaClient() # 对话 result = await client.chat( model="qwen3:72b", messages=[ {"role": "system", "content": "你是专业翻译"}, {"role": "user", "content": "翻译:AGI will change everything"} ], temperature=0.3 ) # 流式输出 async for chunk in client.chat_stream( model="qwen3:72b", messages=[{"role": "user", "content": "写一首诗"}] ): print(chunk, end="", flush=True) # 向量嵌入 embeddings = await client.embed( model="bge-m3", input=["你好世界", "Hello World"] ) OpenAI 兼容接口 from openai import OpenAI # Ollama 兼容 OpenAI API client = OpenAI( base_url="http://localhost:11434/v1", api_key="ollama" # 任意值即可 ) response = client.chat.completions.create( model="qwen3:72b", messages=[{"role": "user", "content": "Hello!"}], stream=True ) 工具调用 # Ollama 2026 原生支持工具调用 result = await client.chat( model="qwen3:72b", messages=[{"role": "user", "content": "北京今天天气怎么样?"}], tools=[{ "type": "function", "function": { "name": "get_weather", "description": "获取天气信息", "parameters": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } } }] ) # 模型返回工具调用 if result.get("message", {}).get("tool_calls"): tool_call = result["message"]["tool_calls"][0] print(f"调用工具: {tool_call['function']['name']}") print(f"参数: {tool_call['function']['arguments']}") 性能优化 量化选择 量化 显存/内存 质量 速度 推荐场景 Q2_K 最低 差 最快 测试/验证 Q4_0 低 中 快 低配设备 Q4_K_M 中低 良 快 日常使用(推荐) Q5_K_M 中 好 中 质量优先 Q8_0 高 优 中 最高质量 FP16 最高 最优 慢 无量化 硬件适配 # CPU 推理优化 export OLLAMA_NUM_THREAD=8 # CPU 线程数 export OLLAMA_NUM_CTX=8192 # 降低上下文长度 # GPU 推理优化 export OLLAMA_GPU_LAYERS=80 # GPU 层数(-1 全部) export OLLAMA_FLASH_ATTENTION=1 # Flash Attention export OLLAMA_KV_CACHE_TYPE=q8_0 # KV Cache 量化 # Apple Silicon 优化 export OLLAMA_METAL_GPU=1 # 使用 Metal export OLLAMA_KV_CACHE_TYPE=q4_0 # 统一内存节省 export OLLAMA_NUM_CTX=16384 # 适合 M 芯片 性能对比 硬件 模型 量化 速度 (tok/s) 首Token延迟 内存占用 M3 Max 64GB Qwen3-72B Q4_K_M 18.5 2.1s 42 GB M3 Max 64GB Qwen3-32B Q4_K_M 45.2 0.8s 20 GB RTX 4090 24GB Qwen3-14B Q4_K_M 85.3 0.3s 9 GB RTX 4090 24GB Qwen3-32B Q4_K_M 42.1 0.5s 20 GB 2×A100 80GB Qwen3-72B Q8_0 55.8 0.4s 75 GB CPU 32-core Qwen3-7B Q4_K_M 12.3 1.5s 5 GB 与应用集成 与 LangChain 集成 from langchain_ollama import ChatOllama, OllamaEmbeddings # 对话模型 llm = ChatOllama( model="qwen3:72b", temperature=0.7, base_url="http://localhost:11434" ) # 嵌入模型 embeddings = OllamaEmbeddings( model="bge-m3", base_url="http://localhost:11434" ) # 使用 from langchain_core.messages import HumanMessage response = llm.invoke([HumanMessage(content="你好")]) 与 Dify 集成 # Dify 配置 Ollama model_provider: name: ollama credentials: base_url: http://localhost:11434 models: - name: qwen3:72b type: llm context_size: 32768 - name: bge-m3 type: text-embedding context_size: 8192 多节点部署 # Ollama 集群配置(2026 新特性) # ollama-cluster.yaml nodes: - name: node-1 host: 192.168.1.101 port: 11434 gpu: "RTX 4090" models: ["qwen3:32b", "bge-m3"] - name: node-2 host: 192.168.1.102 port: 11434 gpu: "RTX 4090" models: ["qwen3:32b"] # 负载均衡 - name: node-3 host: 192.168.1.103 port: 11434 gpu: "A100 80GB" models: ["qwen3:72b"] # 大模型专用 # 路由策略 router: strategy: "least_loaded" # 最少负载 health_check_interval: 10 failover: true 适用场景 最适合 隐私敏感场景:数据不出本地的安全需求 离线环境:无网络或网络不稳定的环境 开发测试:快速测试不同模型的表现 边缘计算:IoT 设备、边缘服务器上的 AI 个人使用:在自己的电脑上运行 AI 助手 不太适合 高并发生产:吞吐量不如 vLLM/TGI 超大模型:671B 级别模型需要集群 成本敏感:相比云端 API,电费和硬件成本较高 总结 Ollama 在 2026 年仍然是"本地运行大模型"的最佳选择。它的核心价值不在于性能极致——vLLM 和 TensorRT-LLM 在吞吐量上更优——而在于"简单"。一条命令安装,一条命令运行,OpenAI 兼容 API,跨平台支持,这些特性让 Ollama 成为开发者在本地使用大模型的首选。 ...

2026-06-28 · 5 min · 888 words · 硅基 AGI 探索者
vllm 2026 deployment guide

vLLM 2026 生产部署完全指南

vLLM 2026:推理引擎的事实标准 vLLM 在 2026 年已经成为大模型推理部署的事实标准。根据社区统计,全球超过 70% 的开源大模型生产部署使用 vLLM 作为推理引擎。它的核心优势在于 PagedAttention 技术带来的高吞吐量和低延迟,以及对各类开源模型的广泛支持。 2026 核心特性 版本演进 特性 vLLM 0.3 (2024) vLLM 0.8 (2026) PagedAttention v1 v3(内存效率+40%) 连续批处理 支持 支持 + 动态批大小 张量并行 支持 支持 + 专家并行 量化 AWQ/GPTQ AWQ/GPTQ/FP8/INT4 多模态 实验性 原生支持 Speculative Decoding 不支持 支持 长上下文 32k 1M+ 分离式推理 不支持 Prefill/Decode 分离 安装与环境准备 基础安装 # 创建虚拟环境 python -m venv vllm-env source vllm-env/bin/activate # Linux # vllm-env\Scripts\activate # Windows # 安装 vLLM(CUDA 12.1+) pip install vllm==0.8.5 # 验证安装 python -c "import vllm; print(vllm.__version__)" GPU 环境检查 # 检查 CUDA 版本 nvidia-smi # 需要 CUDA 12.1+,驱动 535+ # 检查 GPU 内存 nvidia-smi --query-gpu=name,memory.total,memory.free --format=csv 模型内存需求参考 模型 参数量 FP16 显存 INT8 显存 INT4 显存 推荐 GPU Qwen2.5-7B 7B 14 GB 8 GB 5 GB RTX 4090 Llama-4-8B 8B 16 GB 9 GB 5 GB RTX 4090 Qwen2.5-32B 32B 64 GB 34 GB 20 GB A100 80GB Llama-4-70B 70B 140 GB 75 GB 42 GB 2×A100 80GB Qwen3-72B 72B 145 GB 78 GB 44 GB 2×A100 80GB DeepSeek-V3 671B (MoE) 1.3 TB 700 GB 400 GB 8×H100 80GB 基础部署 单 GPU 部署 from vllm import LLM, SamplingParams # 加载模型 llm = LLM( model="Qwen/Qwen2.5-32B-Instruct", quantization="awq", # 使用 AWQ 量化 max_model_len=32768, # 最大上下文长度 gpu_memory_utilization=0.90, # GPU 内存利用率 tensor_parallel_size=1, # 张量并行度 dtype="float16", # 数据类型 trust_remote_code=True, enforce_eager=False, # 使用 CUDA Graph 优化 swap_space=4, # CPU 交换空间 (GB) max_num_seqs=256, # 最大并发序列数 ) # 配置采样参数 sampling_params = SamplingParams( temperature=0.7, top_p=0.9, top_k=50, max_tokens=2048, repetition_penalty=1.05, ) # 批量推理 prompts = [ "解释量子计算的基本原理", "写一首关于春天的诗", "用 Python 实现快速排序算法", ] outputs = llm.generate(prompts, sampling_params) for output in outputs: print(output.outputs[0].text) 多 GPU 张量并行 # 2×A100 80GB 部署 70B 模型 llm = LLM( model="meta-llama/Llama-4-70B-Instruct", tensor_parallel_size=2, # 2 路张量并行 pipeline_parallel_size=1, # 流水线并行 gpu_memory_utilization=0.92, max_model_len=65536, dtype="float16", enable_prefix_caching=True, # 前缀缓存 enable_chunked_prefill=True, # 分块预填充 ) OpenAI 兼容 API 服务 # 启动 API 服务器 python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-32B-Instruct \ --quantization awq \ --tensor-parallel-size 2 \ --gpu-memory-utilization 0.90 \ --max-model-len 32768 \ --port 8000 \ --host 0.0.0.0 \ --api-key sk-your-api-key \ --served-model-name qwen-32b \ --enable-prefix-caching \ --enable-chunked-prefill \ --max-num-seqs 256 \ --uvicorn-log-level info # 客户端调用(兼容 OpenAI SDK) from openai import OpenAI client = OpenAI( api_key="sk-your-api-key", base_url="http://localhost:8000/v1" ) response = client.chat.completions.create( model="qwen-32b", messages=[ {"role": "system", "content": "你是专业翻译"}, {"role": "user", "content": "Translate: Hello World"} ], temperature=0.3, max_tokens=100, stream=True # 支持流式输出 ) for chunk in response: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="") 性能优化 1. 量化策略 # FP8 量化(H100 专用,吞吐量最高) llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", quantization="fp8", dtype="bfloat16", gpu_memory_utilization=0.92, ) # INT4 AWQ 量化(最省显存) llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", quantization="awq", quantization_config={ "bits": 4, "group_size": 128, "zero_point": True, }, ) # GPTQ 量化 llm = LLM( model="TheBloke/Llama-4-70B-GPTQ", quantization="gptq", dtype="float16", ) 2. Speculative Decoding(投机解码) # 使用小模型加速大模型推理 llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", speculative_model="Qwen/Qwen2.5-1.5B-Instruct", # 草稿模型 num_speculative_tokens=5, # 每次投机 5 个 token speculative_draft_tensor_parallel_size=1, gpu_memory_utilization=0.92, ) # 效果:吞吐量提升 2-3x,延迟降低 40-60% # 代价:草稿模型需要共享词表 3. 分离式推理(Prefill-Decode 分离) # 2026 新特性:将 Prefill 和 Decode 分离到不同 GPU # 适合高并发场景 # Prefill 节点(计算密集) python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-72B-Instruct \ --disaggregation-mode prefill \ --disaggregation-port 5001 \ --port 8000 # Decode 节点(内存密集) python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-72B-Instruct \ --disaggregation-mode decode \ --disaggregation-port 5001 \ --port 8001 # 路由层(自动分发请求) python -m vllm.entrypoints.disagg_router \ --prefill-endpoint http://gpu1:8000 \ --decode-endpoint http://gpu2:8001 \ --port 8080 4. 长上下文优化 # 1M 上下文部署 llm = LLM( model="Qwen/Qwen2.5-32B-Instruct-1M", max_model_len=1048576, # 1M tokens gpu_memory_utilization=0.95, # 长上下文优化 enable_chunked_prefill=True, # 分块预填充 max_num_batched_tokens=8192, # 每批最大 token 数 max_num_seqs=32, # 降低并发数以容纳长序列 # KV Cache 优化 block_size=16, # PagedAttention 块大小 swap_space=16, # CPU 交换空间 # 滑动窗口注意力(适用于超长上下文) sliding_window=131072, # 128K 滑动窗口 ) 生产部署架构 Kubernetes 部署 apiVersion: apps/v1 kind: Deployment metadata: name: vllm-qwen-32b spec: replicas: 2 selector: matchLabels: app: vllm-qwen-32b template: metadata: labels: app: vllm-qwen-32b spec: containers: - name: vllm image: vllm/vllm-openai:v0.8.5 args: - --model=Qwen/Qwen2.5-32B-Instruct - --quantization=awq - --tensor-parallel-size=2 - --gpu-memory-utilization=0.90 - --max-model-len=32768 - --port=8000 resources: limits: nvidia.com/gpu: 2 memory: 128Gi requests: nvidia.com/gpu: 2 memory: 64Gi ports: - containerPort: 8000 readinessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 120 periodSeconds: 10 livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 300 periodSeconds: 30 --- apiVersion: v1 kind: Service metadata: name: vllm-service spec: selector: app: vllm-qwen-32b ports: - port: 8000 targetPort: 8000 type: LoadBalancer 负载均衡配置 # Nginx 负载均衡 upstream vllm_backend { least_conn; # 最少连接策略 server gpu-node-1:8000 weight=1 max_fails=3 fail_timeout=30s; server gpu-node-2:8000 weight=1 max_fails=3 fail_timeout=30s; server gpu-node-3:8000 weight=1 max_fails=3 fail_timeout=30s; keepalive 32; keepalive_timeout 60s; } server { listen 80; location /v1/ { proxy_pass http://vllm_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; # 流式响应支持 proxy_buffering off; proxy_cache off; chunked_transfer_encoding on; # 超时设置 proxy_connect_timeout 10s; proxy_read_timeout 300s; proxy_send_timeout 60s; } location /health { proxy_pass http://vllm_backend/health; } } 性能基准 吞吐量对比(Qwen2.5-32B AWQ,2×A100 80GB) 配置 吞吐量 (tok/s) P50 延迟 P99 延迟 并发数 基础配置 2,800 0.8s 3.2s 64 + 前缀缓存 3,500 0.6s 2.5s 64 + 分块预填充 4,200 0.5s 2.1s 128 + Speculative 6,800 0.3s 1.2s 128 + FP8 (H100) 8,500 0.25s 0.9s 256 与其他推理引擎对比 引擎 吞吐量 延迟 显存效率 易用性 vLLM 0.8 4,200 tok/s 0.5s 92% ⭐⭐⭐⭐⭐ TGI 3.0 3,100 tok/s 0.7s 85% ⭐⭐⭐⭐ SGLang 0.3 4,800 tok/s 0.4s 90% ⭐⭐⭐⭐ TensorRT-LLM 5,200 tok/s 0.3s 95% ⭐⭐⭐ Ollama 1,800 tok/s 1.2s 70% ⭐⭐⭐⭐⭐ 监控与运维 Prometheus 指标 # vLLM 内置 Prometheus 指标 # 访问 http://localhost:8000/metrics # 关键指标: # vllm:num_requests_running - 正在运行的请求数 # vllm:num_requests_waiting - 等待队列长度 # vllm:gpu_cache_usage_perc - GPU 缓存使用率 # vllm:time_to_first_token - TTFT(首 Token 延迟) # vllm:time_per_output_token - TPOT(每 Token 延迟) # vllm:e2e_request_latency - 端到端延迟 # Prometheus 采集配置 scrape_configs: - job_name: 'vllm' static_configs: - targets: ['gpu-node-1:8000', 'gpu-node-2:8000'] metrics_path: /metrics scrape_interval: 10s 常见问题排查 问题 原因 解决方案 OOM 显存不足 降低 gpu_memory_utilization 或使用量化 首Token延迟高 Prefill 慢 启用 chunked_prefill 吞吐量低 批处理不足 增加 max_num_seqs 请求排队 并发过高 增加副本数或降低 max_model_len 模型加载慢 磁盘 I/O 使用本地 SSD 缓存模型 总结 vLLM 在 2026 年仍然是开源 LLM 推理部署的最佳选择。它的 PagedAttention v3、Speculative Decoding、分离式推理等特性让它在性能上保持领先,同时 OpenAI 兼容 API 降低了使用门槛。 ...

2026-06-28 · 5 min · 929 words · 硅基 AGI 探索者
prompt compression techniques

Prompt 压缩技术:让上下文窗口利用率提升 50%

上下文窗口的"房价"问题 2026 年,虽然主流模型的上下文窗口已达到 128K-1M tokens,但"窗口越大越不够用"——RAG 检索结果、工具调用返回、对话历史、知识库内容,每个环节都在争抢窗口空间。Prompt 压缩技术就像是在有限的土地上建造高层建筑,让每一个 token 都发挥最大价值。 一、Prompt 压缩的价值 1.1 成本与性能双优化 优化维度 压缩前 压缩后 改善 输入 Token 数 8000 4000 -50% API 成本(/千次) $24 $12 -50% 响应延迟 3.2s 1.8s -44% 上下文利用率 40% 75% +87.5% 信息保留率 100% 92-97% -3~8% 1.2 压缩策略分类 Prompt 压缩 ├── 无损压缩 │ ├── 符号化压缩(缩写、代码化) │ ├── 结构化压缩(JSON→紧凑格式) │ └── 去冗余压缩(删除重复信息) ├── 有损压缩 │ ├── 语义压缩(LLM 总结) │ ├── 选择性保留(截断低重要性内容) │ └── 信息蒸馏(提取关键信息) └── 混合压缩 └── 分层压缩策略 二、无损压缩技术 2.1 符号化压缩 class SymbolicCompressor: """符号化压缩——用短符号替代长文本""" SYMBOL_MAP = { # 常见指令缩写 "请分析以下内容并给出": "分析:", "请根据以上信息回答": "回答:", "以下是相关的背景信息": "背景:", "请注意以下重要事项": "注意:", # 角色缩写 "你是一个专业的": "角色:", "你的核心职责是": "职责:", # 格式缩写 "请用Markdown表格格式输出": "→MD表格", "请用JSON格式输出": "→JSON", "请用列表格式输出": "→列表", # 常见短语 "需要注意的是": "⚠", "重要提醒": "‼", "例如": "如", "也就是说": "即", } def compress(self, prompt: str) -> str: for full, symbol in self.SYMBOL_MAP.items(): prompt = prompt.replace(full, symbol) return prompt def decompress_guide(self) -> str: """生成符号说明(添加到System Prompt)""" guide = "符号说明: " for full, symbol in self.SYMBOL_MAP.items(): guide += f"{symbol}={full[:4]}.. " return guide 2.2 结构化压缩 class StructuralCompressor: """结构化压缩——压缩冗余的格式""" def compress_table(self, markdown_table: str) -> str: """压缩 Markdown 表格""" lines = markdown_table.strip().split('\n') if len(lines) < 3: return markdown_table # 提取表头和数据 headers = [h.strip() for h in lines[0].split('|')[1:-1]] data_rows = [] for line in lines[2:]: # 跳过分隔行 cells = [c.strip() for c in line.split('|')[1:-1]] data_rows.append(cells) # 紧凑格式:用 | 分隔,不用对齐 compact = '|'.join(headers) + '\n' for row in data_rows: compact += '|'.join(row) + '\n' return compact def compress_json(self, json_str: str) -> str: """压缩 JSON""" import json data = json.loads(json_str) return json.dumps(data, ensure_ascii=False, separators=(',', ':')) def compress_list(self, markdown_list: str) -> str: """压缩列表""" lines = markdown_list.strip().split('\n') items = [l.lstrip('- *').strip() for l in lines if l.strip()] return '; '.join(items) 2.3 去冗余压缩 class RedundancyRemover: """去冗余压缩""" def compress(self, prompt: str) -> str: # 1. 移除重复段落 prompt = self._remove_duplicate_paragraphs(prompt) # 2. 移除重复句子 prompt = self._remove_duplicate_sentences(prompt) # 3. 移除空白行 prompt = self._remove_blank_lines(prompt) # 4. 合并连续空格 import re prompt = re.sub(r' {2,}', ' ', prompt) return prompt def _remove_duplicate_paragraphs(self, text: str) -> str: paragraphs = text.split('\n\n') seen = set() unique = [] for p in paragraphs: normalized = p.strip().lower() if normalized and normalized not in seen: seen.add(normalized) unique.append(p) return '\n\n'.join(unique) def _remove_duplicate_sentences(self, text: str) -> str: import re sentences = re.split(r'(?<=[。.!?!?])\s+', text) seen = set() unique = [] for s in sentences: if s.strip() and s.strip() not in seen: seen.add(s.strip()) unique.append(s) return ' '.join(unique) 三、有损压缩技术 3.1 LLM 语义压缩 class SemanticCompressor: """使用 LLM 进行语义压缩""" COMPRESSION_PROMPT = """请压缩以下文本,要求: 1. 保留所有关键信息和数据 2. 保留逻辑结构和因果关系 3. 移除冗余描述和过渡语句 4. 用更简洁的表达替代冗长表达 5. 保持事实准确性 原始文本({original_tokens} tokens): {text} 输出压缩后的文本,目标:{target_tokens} tokens以内。""" def compress(self, text: str, target_ratio: float = 0.5, llm_client=None) -> str: original_tokens = self._estimate_tokens(text) target_tokens = int(original_tokens * target_ratio) prompt = self.COMPRESSION_PROMPT.format( original_tokens=original_tokens, text=text, target_tokens=target_tokens ) compressed = llm_client.generate(prompt) return compressed def _estimate_tokens(self, text: str) -> int: # 粗略估算 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars return chinese_chars * 2 + other_chars // 4 3.2 选择性保留压缩 class SelectiveCompressor: """选择性保留——基于重要性的压缩""" def compress(self, text: str, target_ratio: float = 0.5) -> str: # 1. 分割为句子 sentences = self._split_sentences(text) # 2. 计算每个句子的重要性分数 scored = self._score_sentences(sentences) # 3. 保留高重要性句子 target_count = int(len(sentences) * target_ratio) top_sentences = sorted(scored, key=lambda x: -x[1])[:target_count] # 4. 按原顺序排列 top_sentences.sort(key=lambda x: x[2]) # 按原始位置排序 return ' '.join(s[0] for s in top_sentences) def _score_sentences(self, sentences: list) -> list: """使用 TextRank 思想计算句子重要性""" # 基于句子间相似度构建图 n = len(sentences) scores = [1.0] * n for iteration in range(10): # 迭代计算 new_scores = [] for i, sent in enumerate(sentences): score = 0.15 # 基础分 for j, other in enumerate(sentences): if i != j: sim = self._sentence_similarity(sent, other) score += 0.85 * sim * scores[j] / max( sum(self._sentence_similarity(other, s2) for k, s2 in enumerate(sentences) if k != j), 1e-8 ) new_scores.append(score) scores = new_scores return [(sentences[i], scores[i], i) for i in range(n)] def _sentence_similarity(self, s1: str, s2: str) -> float: """计算两个句子的相似度""" words1 = set(s1.split()) words2 = set(s2.split()) intersection = words1 & words2 union = words1 | words2 return len(intersection) / max(len(union), 1) 3.3 信息蒸馏 class InformationDistiller: """信息蒸馏——提取关键信息,丢弃细节""" DISTILL_PROMPT = """从以下文本中提取关键信息,使用紧凑格式输出。 输出格式: - 主题:[1-2句话] - 关键事实:[每条1行,最多5条] - 数据:[数值和单位] - 结论:[1句话] 文本: {text}""" def distill(self, text: str, llm_client) -> str: prompt = self.DISTILL_PROMPT.format(text=text) return llm_client.generate(prompt) 四、分层压缩策略 class LayeredCompressor: """分层压缩策略——不同内容用不同压缩方法""" def __init__(self, llm_client): self.llm = llm_client self.symbolic = SymbolicCompressor() self.structural = StructuralCompressor() self.redundancy = RedundancyRemover() self.semantic = SemanticCompressor() self.selective = SelectiveCompressor() self.distiller = InformationDistiller(llm_client) def compress(self, prompt: str, target_ratio: float = 0.5) -> str: """分层压缩""" original_tokens = self._estimate_tokens(prompt) target_tokens = int(original_tokens * target_ratio) # Layer 1: 无损压缩(总是执行) prompt = self.symbolic.compress(prompt) prompt = self.structural.compress_json(prompt) prompt = self.redundancy.compress(prompt) current_tokens = self._estimate_tokens(prompt) if current_tokens <= target_tokens: return prompt # 无损压缩已达标 # Layer 2: 内容分类 sections = self._classify_sections(prompt) # Layer 3: 按类别压缩 compressed_sections = [] for section_type, content in sections: if section_type == 'rules': # 规则类:仅做符号压缩 compressed_sections.append(self.symbolic.compress(content)) elif section_type == 'knowledge': # 知识类:语义压缩 compressed_sections.append( self.semantic.compress(content, 0.4, self.llm) ) elif section_type == 'examples': # 示例类:选择性保留 compressed_sections.append( self.selective.compress(content, 0.6) ) elif section_type == 'context': # 上下文类:信息蒸馏 compressed_sections.append( self.distiller.distill(content, self.llm) ) else: compressed_sections.append(content) result = '\n\n'.join(compressed_sections) # Layer 4: 如果仍超标,全局压缩 if self._estimate_tokens(result) > target_tokens: result = self.semantic.compress(result, target_tokens / self._estimate_tokens(result), self.llm) return result def _classify_sections(self, prompt: str) -> list: """将 Prompt 分为不同类型的段落""" sections = [] current_section = "" current_type = "other" for line in prompt.split('\n'): if line.startswith('规则') or line.startswith('约束'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'rules' elif line.startswith('知识') or line.startswith('背景'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'knowledge' elif line.startswith('示例') or line.startswith('例子'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'examples' elif line.startswith('上下文') or line.startswith('历史'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'context' else: current_section += line + '\n' if current_section: sections.append((current_type, current_section)) return sections 五、对话历史压缩 class ConversationHistoryCompressor: """对话历史压缩——长对话的上下文管理""" def __init__(self, llm_client, max_history_tokens: int = 4000): self.llm = llm_client self.max_tokens = max_history_tokens def compress_history(self, messages: list) -> list: """压缩对话历史""" total_tokens = sum(self._estimate_tokens(m['content']) for m in messages) if total_tokens <= self.max_tokens: return messages # 不需要压缩 # 策略:保留最近 N 轮 + 压缩早期对话 recent_count = min(6, len(messages)) # 保留最近3轮 recent = messages[-recent_count:] old = messages[:-recent_count] # 压缩早期对话 summary = self._summarize_conversation(old) # 构建压缩后的历史 compressed = [ {"role": "system", "content": f"对话摘要:{summary}"}, *recent ] return compressed def _summarize_conversation(self, messages: list) -> str: """总结早期对话""" conversation_text = '\n'.join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请总结以下对话的关键信息,保留: 1. 用户的核心需求和偏好 2. 已达成的结论和决定 3. 未解决的问题 4. 重要的事实和数据 对话内容: {conversation_text} 总结(不超过200字):""" return self.llm.generate(prompt) 六、压缩效果评估 class CompressionEvaluator: """压缩效果评估器""" def evaluate(self, original: str, compressed: str, test_cases: list, llm_client) -> dict: results = { 'compression_ratio': len(compressed) / len(original), 'token_reduction': 1 - self._tokens(compressed) / self._tokens(original), } # 信息保留率评估 info_retention = self._evaluate_info_retention(original, compressed, llm_client) results['info_retention'] = info_retention # 任务效果评估 original_scores = [] compressed_scores = [] for case in test_cases: # 使用原始 Prompt original_response = llm_client.generate(original + case['input']) original_scores.append(self._score(original_response, case['expected'])) # 使用压缩 Prompt compressed_response = llm_client.generate(compressed + case['input']) compressed_scores.append(self._score(compressed_response, case['expected'])) results['original_accuracy'] = sum(original_scores) / len(original_scores) results['compressed_accuracy'] = sum(compressed_scores) / len(compressed_scores) results['accuracy_drop'] = results['original_accuracy'] - results['compressed_accuracy'] # 成本节省 results['cost_saving'] = results['token_reduction'] return results 压缩效果实测数据 压缩方法 压缩率 信息保留 准确率变化 延迟改善 符号压缩 15% 100% 0% +10% 去冗余 20% 100% 0% +15% 语义压缩 45% 94% -3% +40% 选择性保留 50% 90% -5% +45% 信息蒸馏 65% 85% -8% +55% 分层压缩 50% 96% -2% +44% 七、最佳实践 先无损后有损:先尝试无损压缩,不够再考虑有损 分层压缩最优:不同内容用不同策略,综合效果最好 规则不可压缩:System Prompt 中的规则和约束不应被有损压缩 压缩vs精简:很多时候重新设计比压缩更有效 监控压缩质量:定期评估压缩后的任务效果 缓存压缩结果:相同输入的压缩结果可以缓存 结语 Prompt 压缩是在信息密度和效果之间寻找最优平衡点的艺术。2026 年的工具链已经足够成熟,可以实现接近无损的 50% 压缩——这意味着同样的上下文窗口可以容纳两倍的信息,同样的预算可以处理两倍的请求。 ...

2026-06-28 · 6 min · 1189 words · 硅基 AGI 探索者
agent observability otel 2026

Agent 可观测性 2026:OpenTelemetry for LLM 实践

引言 Agent 系统的"黑盒"问题是生产化的最大障碍之一。一个 Agent 调用了 3 个工具、经过 5 轮推理、消耗了 8000 tokens,但出了问题你却不知道在哪一步。2026年,OpenTelemetry 社区正式发布了 SemConv for GenAI 规范,为 LLM 可观测性提供了标准化方案。 一、为什么 Agent 可观测性不同于传统应用 传统微服务的可观测性关注:请求路径、延迟分布、错误率。Agent 系统增加了三个新维度: Token 维度:每次调用消耗多少 Token?成本如何分摊? 推理维度:模型"想"了什么?为什么选择这个工具?为什么跳过某步? 非确定性维度:相同输入可能产生不同输出,仅靠日志无法复现 二、OpenTelemetry GenAI 语义规范 2026年正式定稿的 GenAI SemConv 定义了以下核心 Attributes: # GenAI 基础属性 gen_ai.system: "openai" # 提供商 gen_ai.request.model: "gpt-5" # 模型名称 gen_ai.request.temperature: 0.7 # 采样温度 gen_ai.request.max_tokens: 4096 # 最大 Token # Token 使用 gen_ai.usage.input_tokens: 1523 # 输入 Token gen_ai.usage.output_tokens: 876 # 输出 Token gen_ai.usage.cost: 0.0234 # 本次调用成本(美元) # Agent 特有 gen_ai.agent.name: "research-agent" gen_ai.agent.tool.name: "web_search" gen_ai.agent.tool.result.quality: 0.85 gen_ai.agent.iteration: 3 # 第几轮迭代 # 工具调用 gen_ai.tool.name: "calculator" gen_ai.tool.input: '{"expr": "2+2"}' gen_ai.tool.output: '{"result": 4}' gen_ai.tool.duration_ms: 45 三、全链路追踪实现 架构概览 ┌─────────────────────────────────────────────────────┐ │ Agent Application │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │Step1│──►│Step2│──►│Step3│──►│Step4│ │ │ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ ┌──▼─────────▼─────────▼─────────▼──┐ │ │ │ OpenTelemetry SDK │ │ │ │ (Auto-instrumentation + Custom) │ │ │ └──────────────┬────────────────────┘ │ └─────────────────┼───────────────────────────────────┘ │ OTLP/gRPC ┌────────────▼────────────┐ │ OTel Collector │ │ (处理/采样/导出) │ └──┬─────┬─────┬──────────┘ │ │ │ ┌────▼┐ ┌─▼──┐ ┌▼─────┐ │Jaeger│ │Prom│ │Loki │ │(Trace)│ │(Met)│ │(Log)│ └─────┘ └────┘ └──────┘ Python 实现 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.instrumentation.openai import OpenAIInstrumentor # 1. 初始化 OTel provider = TracerProvider() processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://otel-collector:4317") ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. 自动注入 OpenAI 调用的 Span OpenAIInstrumentor().instrument() # 3. Agent 自定义 Span tracer = trace.get_tracer("agent-system") class ObservabilityMiddleware: """Agent 可观测性中间件""" def __init__(self): self.tracer = trace.get_tracer("agent") async def on_agent_start(self, agent_name: str, input_data: dict): """Agent 启动时创建 Root Span""" self.root_span = self.tracer.start_span( f"agent.{agent_name}", attributes={ "gen_ai.agent.name": agent_name, "agent.input.size": len(str(input_data)), } ) async def on_llm_call(self, model: str, messages: list, **kwargs): """LLM 调用前记录""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"llm.{model}", context=ctx, attributes={ "gen_ai.request.model": model, "gen_ai.request.message_count": len(messages), "gen_ai.request.temperature": kwargs.get("temperature", 1.0), } ) return span async def on_llm_end(self, span, response): """LLM 调用后记录 Token 使用""" usage = response.usage span.set_attributes({ "gen_ai.usage.input_tokens": usage.prompt_tokens, "gen_ai.usage.output_tokens": usage.completion_tokens, "gen_ai.usage.total_tokens": usage.total_tokens, "gen_ai.usage.cost": calculate_cost( usage.prompt_tokens, usage.completion_tokens, response.model ), }) span.end() async def on_tool_call(self, tool_name: str, tool_input: dict): """工具调用追踪""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"tool.{tool_name}", context=ctx, attributes={ "gen_ai.tool.name": tool_name, "gen_ai.tool.input": json.dumps(tool_input)[:500], } ) return span async def on_agent_end(self, output: str): """Agent 结束""" self.root_span.set_attributes({ "agent.output.size": len(output), "agent.status": "success", }) self.root_span.end() Trace 可视化示例 在 Jaeger 中看到的典型 Agent Trace: ...

2026-06-28 · 4 min · 750 words · 硅基 AGI 探索者
prompt engineering 2026 practices

Prompt 工程 2026 最新实践:从技巧到工程化体系

从手工艺到工程化:Prompt 工程的范式转变 2026 年,Prompt 工程已经走过了"调参数式"的手工艺阶段。随着大模型能力边界的不断扩展,Prompt 的设计、测试、部署和维护已经形成了一套完整的工程化体系。根据 Gartner 2026 年 Q1 报告,超过 78% 的企业级 AI 应用已经建立了专门的 Prompt 工程团队,而 2024 年这一比例仅为 23%。 一、Prompt 工程化的核心要素 1.1 结构化 Prompt 架构 现代 Prompt 已从简单的文本指令发展为结构化的工程产物: # prompt-config.yaml prompt: metadata: name: "customer-support-v2" version: "2.3.1" author: "prompt-team" last_updated: "2026-06-15" system_prompt: role: "你是一个专业的客服助手" constraints: - "回答必须基于知识库内容" - "不确定时明确告知用户" - "单次回复不超过500字" knowledge_sources: - "product_docs" - "faq_database" user_template: | 用户问题:{{user_query}} 上下文信息:{{context}} 历史对话:{{history}} expected_output: format: "json" schema: answer: "string" confidence: "float" sources: "array" 1.2 Prompt 生命周期管理 设计 → 测试 → 评审 → 部署 → 监控 → 迭代 ↑ ↓ ←─────────── 反馈循环 ←─────────────← 二、2026 年 Prompt 设计核心原则 2.1 明确性原则(Clarity First) 维度 错误示例 正确示例 角色 “帮我写文章” “你是科技领域资深编辑,擅长将复杂技术概念转化为通俗易懂的文章” 任务 “总结一下” “用3段话总结以下内容,每段不超过100字,重点突出数据和结论” 约束 “写得好一点” “语气专业但不失亲和力,避免使用行业黑话,目标读者是产品经理” 格式 “列出来” “用Markdown表格输出,包含名称、描述、优缺点三列” 2.2 上下文工程(Context Engineering) 2026 年最重要的进化是"上下文工程"概念的兴起——不再只关注单条 Prompt 的措辞,而是整体上下文的构建: ...

2026-06-28 · 4 min · 711 words · 硅基 AGI 探索者
鲁ICP备2026018361号