一、定义Prompt Engineering是针对当前任务生成Prompt模板Prompt模板主要是通过自然语言提示引导人工智能模型生成想要的回答或输出。二、不同层级的Prompt Engineering层级一基础单轮提示词工程手写 Prompt简单使用直接写提示词给两三条示例强制输出 JSON一次性任务脚本直接跑没有代码层面的模板封装。适合单轮简单任务。【角色】你是XX专家 【任务】完成xxx 【约束】禁止xxx 【Few‑shot样例】 输入xxx 输出xxx #输出格式控制 【输出格式】必须返回JSON字段a,b,c 【用户输入】{query}核心能力1提示词优化清晰指令背景、角色、任务、约束消除歧义 技巧明确身份明确目标、明确禁止行为、 作用指定思考逻辑减少模型自由发挥 全部约束写在 prompt 文本里面没有代码层面校验。核心能力2Few-shot示例给模型输入、输出样例 作用告诉大模型 “就照着这个格式 风格做” 0-shot无示例 1-shot一组示例 Few-shot多组示例。常用于分类处理核心能力3输出格式控制强制返回 JSON/ markdown/ XML/ YAML等结构化结果 作用方便代码解析避免自然语言乱输出层级二Prompt工程化组件复用把角色、规则抽成模板片段代码运行时把内容动态拼到模板里。面向线上简单服务。核心能力1prompt片段化与复用把大 Prompt 拆成独立片段角色片段、约束片段、few‑shot样例片段、格式片段。 作用多处业务直接引用修改一处全局生效避免复制粘贴带来版本不一致。 实践Prompt模板库、常量文件、模板字符串、Prompt 版本管理。核心能力2动态编排与组合运行时根据业务条件拼接不同片段而不是写死完整 prompt 示例 用户 A 场景拼接【规则 A】片段用户 B 场景拼接【规则 B】片段 根据输入数据动态插入 Few‑shot 样例、业务上下文 实践模板引擎、字符串模板、条件拼接、Prompt Chain 提示词链。核心能力3上下文管理多轮对话场景管理历史消息窗口 作用解决上下文窗口超限问题层级三Prompt系统化设计Agent 场景下要求大模型输出结构化任务计划代码校验返回结果格式错就回传给模型修复编排多轮思考、工具调用循环处理复杂业务。使用生产级LLM应用。核心能力1系统结构化约束结构化约束 Schema 契约 Prompt 约束 模型强制输出 代码校验 Schema 契约写死在代码里的约定相当于 LLM 和业务程序之间的接口文档。 有哪些字段、字段是什么类型、哪些必须给、允许写什么值不允许写什么 接口文档就是前后端的契约。后端规定接口返回name字符串age数字前端就按这个契约去解析。 JSON Schema / Pydantic 模型就是 LLM ↔ 你的代码之间的接口文档契约。 结构化输出优先用 response_format依然要做代码校验模型强制输出也会出错。 作用解决模型输出格式乱的问题。 结构化约束与提示词优化的区别 1、Level1 手写 Prompt契约写在 prompt自然语言里口头约定没有代码校验。模型违约代码识别不出来 2、Level3 Schema 契约契约写在代码 Schema代码强制校验每一条约定违约就捕获错误自动修复重试。核心能力2验证与修复机制LLM 输出不一定合规不能直接交给业务代码输出后做校验失败自动重试 / 修复。 完整链路LLM输出 → 校验器校验 → 校验通过直接使用校验失败 → 修复重试 1、校验做什么 1格式校验JSON 是否合法能不能 loads 2字段校验必填字段是否缺失、字段类型是否正确、枚举值是否合法、数值范围校验 3值校验业务参数合理性额度不能负数、ID 格式 2、修复策略按成本从低到高 1本地修复低成本代码层面做简单清洗剥离 markdown json 标记、剔除多余文本截取 json 子串。 2LLM 自修复最常用把错误输出 校验报错信息丢回大模型让模型按照 schema 重新修正输出。 3完整重试重建 prompt重新调用大模型设置最大重试次数防止死循环。 4降级兜底重试耗尽返回业务降级结果不抛异常崩溃。核心能力3多轮对话编排将复杂任务拆解为多个阶段每个阶段有明确的输入/输出/契约AI 按阶段顺序执行支持中断恢复。 常见实现模式 1、Prompt Chain 提示词链串行执行多个子 Prompt上一轮输出作为下一轮输入。 2、状态机编排维护会话状态根据状态决定下一步执行什么 Prompt / 调用什么工具 3、分支条件跳转根据上一轮输出结果选择不同分支不同分支加载不同 Prompt 片段。 4、结合前面上下文管理器每一轮执行前做上下文裁剪、老旧 tool 消息过滤控制 token。三、上下文管理核心要点❌ 不要直接把消息字符串中间截断会出现残缺 JSON、残缺指令要以完整对话单元为单位删除。 ❌ System 提示词不要裁剪一旦丢失模型直接丢失业务约束。 ❌ 不要无限累积历史用户会话跑 100 轮后 token 会爆炸导致接口报错、耗时暴涨。 ⚠️ 摘要不能过度压缩关键参数订单号、金额、业务 id务必保留。1、基础方案滑动窗口固定保留最近 N 轮最简单工业界最常用。逻辑只保留最近k轮对话更早的历史直接丢弃。实现维护messages消息数组超过最大轮数从头部删掉旧消息。优点实现简单token 可控缺点会丢失早期重要信息长会话前面的业务记忆直接没了。MAX_ROUND 6 messages [sys_prompt, user1, ai1, user2, ai2, user3, ai3 ...] # 超过轮数删除最早的用户助手对系统提示词保留不动 while len(messages) MAX_ROUND 1: del messages[1:3]2、基于 Token 的截断比按轮数更精准不要按轮数切统计 token 数量当总 token 接近模型上限从头部裁剪历史消息系统 prompt 永远保留。关键点系统 System Prompt 优先保留业务历史优先删旧的。注意不能粗暴截断单条消息中间会造成文本残缺要按完整对话单元删除。max_context_tokens 3000 # 循环计算总token超阈值就删掉最老的一轮(usrassistant) while count_tokens(messages) max_context_tokens: # 删除最早的一对用户助手消息索引跳过system(0号) del messages[1:3]3、上下文摘要压缩解决长对话丢失早期关键信息滑动窗口会丢老信息摘要把久远历史浓缩成一小段总结文本代替原始完整对话。执行时机当历史消息 token 超过阈值。优点不会丢失早期关键业务信息缺点增加一轮 LLM 调用带来额外开销摘要会存在信息失真风险。流程把老旧的历史消息发给 LLM让模型总结把下面对话浓缩成简短业务摘要保留关键业务参数、用户诉求将原始旧消息全部删除把生成的摘要文本插入 messages保留最近 N 轮原始对话 历史摘要 system 提示词。[ {role:system,content:业务角色指令}, {role:assistant,content:【历史摘要】用户之前申请动产融资额度50万关注放款时效}, # 压缩后的旧会话 {role:user,content:最近用户问题1}, # 最近几轮完整原始对话 {role:assistant,content:回答1}, {role:user,content:最近用户问题2}, {role:assistant,content:回答2}, ]4、分层记忆Agent 常用短期记忆 长期记忆短期记忆messages 数组存放最近几轮原始对话滑动窗口管理长期记忆外部向量库 / 数据库把久远对话做向量化存储。当用户新 query 进来做向量检索把相关的历史片段召回拼入本轮 prompt不相关的旧对话不塞进上下文。典型用于 AI Agent重要业务信息持久化存入向量库不占实时上下文窗口。缺点检索不准会带来幻觉需要做召回过滤。5、消息角色区分过滤无效消息线上业务经常产生大量工具返回、中间思考日志不要无脑全部塞给后续轮次system必须保留核心业务规则用户 user不能丢业务输入assistant正常回答保留Agent 内部思考 / ReAct 中间思考日志可以选择性丢弃tool 返回只保留上一轮工具返回更早的工具调用结果直接丢弃。很多新手踩坑把几十轮完整工具调用全部塞进 messagestoken 直接爆炸。6、业务侧特殊优化工程实践区分会话隔离不同用户、不同业务会话独立维护 messages不要混在一起。Prompt 片段隔离动态业务数据RAG 检索结果、工具返回单独控制长度做截断。开关控制短会话不走压缩长会话才触发摘要 / 裁剪。兜底保护无论什么策略设置硬 token 上限防止异常数据打满上下文。四、上下文管理核心实现1、tiktoken 精确 token 计数# 优先调 LLM 的 Tokenizer 接口精确 TOKENIZER_URL f{BASE_URL}/xxx/xxx/tokenizer/estimate-token-count resp requests.post(TOKENIZER_URL, ...) # 降级接口挂了就用字符数估算粗略 return sum(len(m[content]) for m in clean) // 2def count_tokens(messages: list[dict], model: str JoyAI-LLM-Flash) - int: clean [{role: m[role], content: m[content]} for m in messages if m.get(content)] if not clean: return 0 try: resp requests.post( TOKENIZER_URL, headers{api-key: API_KEY, Content-Type: application/json}, json{messages: clean, model: model}, timeout30, ) resp.raise_for_status() data resp.json().get(data, {}) count data.get(total_tokens) or data.get(token_count) or 0 return int(count) except Exception as e: print(f[token计数] 接口异常降级估算: {e}) return sum(len(m[content]) for m in clean) // 22、滑动窗口裁剪深拷贝数据 def trim(self, messages: list[dict]) - list[dict]: result copy.deepcopy(messages) current_tokens count_tokens(result, self.model) print(f[SlideWindowTrim] 裁剪前: token{current_tokens}, 上限{self.max_total_tokens}, 消息数{len(result)}) if current_tokens self.max_total_tokens: result self._sliding_trim(result) print(f[SlideWindowTrim] 裁剪后: token{count_tokens(result, self.model)}, 消息数{len(result)}) else: print(f[SlideWindowTrim] token{current_tokens} 未超上限无需裁剪) return result #真正的滑动窗口裁剪逻辑默认保留system消息 def _sliding_trim(self, messages: list[dict]) - list[dict]: sys_count sum(1 for m in messages if m[role] system) min_keep self.keep_last_n_round * 2 history_count len(messages) - sys_count trim_count 0 while count_tokens(messages, self.model) self.max_total_tokens: if history_count min_keep: print(f[SlideWindowTrim] 已达到最小保留轮数({self.keep_last_n_round}轮{min_keep}条)停止裁剪) break #下面3行是裁剪循环的核心操作 del messages[sys_count: sys_count 2] history_count - 2 trim_count 1 print(f[SlideWindowTrim] 第{trim_count}次裁剪: 删除1对消息, 当前token{count_tokens(messages, self.model)}, 剩余消息数{len(messages)}) print(f[SlideWindowTrim] 滑动窗口裁剪完成: 共删除{trim_count}对消息 if trim_count 0 else [SlideWindowTrim] 未执行裁剪已达最小保留轮数或token已达标) return messages3、保留最近 N 轮原始对话达到阈值自动触发历史摘要压缩#分割旧对话和近期对话 def split_old_and_recent( history: list[dict], keep_last_n_round: int 4 ) - tuple[list[dict], list[dict]]: keep_msg_count keep_last_n_round * 2 if len(history) keep_msg_count: return [], history return history[:-keep_msg_count], history[-keep_msg_count:] #旧消息压缩为摘要 def get_summary_once( old_history: list[dict], model: str JoyAI-LLM-Flash, temperature: float 0.3, max_tokens: int 500, ) - str: if not old_history: return conversation_text \n.join( f[{m[role]}]: {m[content]} for m in old_history ) prompt f 请对下面的对话历史做精简摘要。 要求保留用户核心诉求、业务关键数字、订单、金额、ID等关键信息去掉冗余闲聊。 不要输出多余解释直接输出摘要文本。 对话历史 {conversation_text} try: from openai import OpenAI client OpenAI(api_keyAPI_KEY, base_urlSUMMARY_BASE_URL) resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, max_tokensmax_tokens, ) summary resp.choices[0].message.content.strip() print(f[SummaryOnce] 生成摘要: {summary[:100]}...) return summary except Exception as e: print(f[SummaryOnce] 摘要生成失败降级为简单拼接: {e}) return | .join(f{m[role]}:{m[content][:50]} for m in old_history) #摘要文字消息插到近期对话之前 def make_summary_message(summary_text: str) - dict: return {role: assistant, content: f【历史对话摘要】:\n{summary_text}}4、增加一个开关可以关闭摘要功能只保留滑动窗口# 开关True 开启摘要False 关闭 SUMMARY_ON False def split_old_and_recent( history: list[dict], keep_last_n_round: int 4 ) - tuple[list[dict], list[dict]]: keep_msg_count keep_last_n_round * 2 if len(history) keep_msg_count: return [], history return history[:-keep_msg_count], history[-keep_msg_count:] def get_summary_once( old_history: list[dict], model: str JoyAI-LLM-Flash, temperature: float 0.3, max_tokens: int 500, ) - str: if not old_history: return conversation_text \n.join( f[{m[role]}]: {m[content]} for m in old_history ) prompt f 请对下面的对话历史做精简摘要。 要求保留用户核心诉求、业务关键数字、订单、金额、ID等关键信息去掉冗余闲聊。 不要输出多余解释直接输出摘要文本。 对话历史 {conversation_text} try: from openai import OpenAI client OpenAI(api_keyAPI_KEY, base_urlSUMMARY_BASE_URL) resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, max_tokensmax_tokens, ) summary resp.choices[0].message.content.strip() print(f[SummaryOnce] 生成摘要: {summary[:100]}...) return summary except Exception as e: print(f[SummaryOnce] 摘要生成失败降级为简单拼接: {e}) return | .join(f{m[role]}:{m[content][:50]} for m in old_history) def make_summary_message(summary_text: str) - dict: return {role: assistant, content: f【历史对话摘要】:\n{summary_text}} def build_context_messages(history, keep_last_n_round4): if not SUMMARY_ON: return list(history) old_part, recent_part split_old_and_recent(history, keep_last_n_round) if not old_part: return list(history) return [make_summary_message(get_summary_once(old_part))] recent_part5、reAct Agent 场景ReAct Reason Act每圈循环就是一个「思考 → 行动 → 观察」周期。经典的Thought → Action → Observation三段式循环Thought: 我该做什么LLM 推理Action: 调哪个工具、传什么参数LLM 输出Observation: 工具执行结果程序执行后喂回→ 回到 Thought直到能给出 Final AnswerMAX_TOTAL_TOKENS 500 TRIMMER SlideWindowTrimmer(max_total_tokensMAX_TOTAL_TOKENS, keep_last_n_round3) SYSTEM_PROMPT 你是贷款业务助手可以通过调用工具解决问题。 工具清单 1. query_order(order_id)查询订单状态 2. calc(amount, rate, years)计算贷款总利息amount本金rate年利率years年限 请严格按以下格式回复一次只执行一步 Thought: 思考当前情况决定下一步 Action: 工具名(参数) 当你已经能给出最终答案时回复 Thought: 我已经得到答案 Final Answer: 最终回答 def query_order(order_id): orders { PRE202601001: {status: 预审中, amount: 800000}, PRE202601002: {status: 已放款, amount: 1200000}, } return orders.get(order_id, 订单不存在) def calc(amount, rate, years): return amount * rate * years def manage_context(messages): tokens count_tokens(messages) print(f[Context] 当前 token{tokens}, 上限{MAX_TOTAL_TOKENS}) if tokens MAX_TOTAL_TOKENS: return messages if SUMMARY_ON: system, history messages[:1], messages[1:] messages system build_context_messages(history, keep_last_n_round4) return TRIMMER.trim(messages) def run_agent(question, max_steps5): client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: question}, ] for step in range(1, max_steps 1): messages manage_context(messages) resp client.chat.completions.create( modelMODEL, messagesmessages, temperature0, ) reply resp.choices[0].message.content.strip() messages.append({role: assistant, content: reply}) print(f\n---- 第{step}步 ----\n{reply}\n) if Final Answer: in reply: return reply.split(Final Answer:)[-1].strip() m re.search(rAction:\s*(\w)\((.*)\), reply) if not m: return 模型未按格式输出流程终止 name, args_raw m.group(1), m.group(2).strip() try: observation eval(f{name}({args_raw})) except NameError: observation eval(f{name}({args_raw})) except Exception as e: observation f工具执行失败: {e} print(f[Tool] {name}({args_raw}) - {observation}) messages.append({role: user, content: fObservation: {observation}}) return 达到最大步数流程终止 【Prompt Engineering】
一、定义Prompt Engineering是针对当前任务生成Prompt模板Prompt模板主要是通过自然语言提示引导人工智能模型生成想要的回答或输出。二、不同层级的Prompt Engineering层级一基础单轮提示词工程手写 Prompt简单使用直接写提示词给两三条示例强制输出 JSON一次性任务脚本直接跑没有代码层面的模板封装。适合单轮简单任务。【角色】你是XX专家 【任务】完成xxx 【约束】禁止xxx 【Few‑shot样例】 输入xxx 输出xxx #输出格式控制 【输出格式】必须返回JSON字段a,b,c 【用户输入】{query}核心能力1提示词优化清晰指令背景、角色、任务、约束消除歧义 技巧明确身份明确目标、明确禁止行为、 作用指定思考逻辑减少模型自由发挥 全部约束写在 prompt 文本里面没有代码层面校验。核心能力2Few-shot示例给模型输入、输出样例 作用告诉大模型 “就照着这个格式 风格做” 0-shot无示例 1-shot一组示例 Few-shot多组示例。常用于分类处理核心能力3输出格式控制强制返回 JSON/ markdown/ XML/ YAML等结构化结果 作用方便代码解析避免自然语言乱输出层级二Prompt工程化组件复用把角色、规则抽成模板片段代码运行时把内容动态拼到模板里。面向线上简单服务。核心能力1prompt片段化与复用把大 Prompt 拆成独立片段角色片段、约束片段、few‑shot样例片段、格式片段。 作用多处业务直接引用修改一处全局生效避免复制粘贴带来版本不一致。 实践Prompt模板库、常量文件、模板字符串、Prompt 版本管理。核心能力2动态编排与组合运行时根据业务条件拼接不同片段而不是写死完整 prompt 示例 用户 A 场景拼接【规则 A】片段用户 B 场景拼接【规则 B】片段 根据输入数据动态插入 Few‑shot 样例、业务上下文 实践模板引擎、字符串模板、条件拼接、Prompt Chain 提示词链。核心能力3上下文管理多轮对话场景管理历史消息窗口 作用解决上下文窗口超限问题层级三Prompt系统化设计Agent 场景下要求大模型输出结构化任务计划代码校验返回结果格式错就回传给模型修复编排多轮思考、工具调用循环处理复杂业务。使用生产级LLM应用。核心能力1系统结构化约束结构化约束 Schema 契约 Prompt 约束 模型强制输出 代码校验 Schema 契约写死在代码里的约定相当于 LLM 和业务程序之间的接口文档。 有哪些字段、字段是什么类型、哪些必须给、允许写什么值不允许写什么 接口文档就是前后端的契约。后端规定接口返回name字符串age数字前端就按这个契约去解析。 JSON Schema / Pydantic 模型就是 LLM ↔ 你的代码之间的接口文档契约。 结构化输出优先用 response_format依然要做代码校验模型强制输出也会出错。 作用解决模型输出格式乱的问题。 结构化约束与提示词优化的区别 1、Level1 手写 Prompt契约写在 prompt自然语言里口头约定没有代码校验。模型违约代码识别不出来 2、Level3 Schema 契约契约写在代码 Schema代码强制校验每一条约定违约就捕获错误自动修复重试。核心能力2验证与修复机制LLM 输出不一定合规不能直接交给业务代码输出后做校验失败自动重试 / 修复。 完整链路LLM输出 → 校验器校验 → 校验通过直接使用校验失败 → 修复重试 1、校验做什么 1格式校验JSON 是否合法能不能 loads 2字段校验必填字段是否缺失、字段类型是否正确、枚举值是否合法、数值范围校验 3值校验业务参数合理性额度不能负数、ID 格式 2、修复策略按成本从低到高 1本地修复低成本代码层面做简单清洗剥离 markdown json 标记、剔除多余文本截取 json 子串。 2LLM 自修复最常用把错误输出 校验报错信息丢回大模型让模型按照 schema 重新修正输出。 3完整重试重建 prompt重新调用大模型设置最大重试次数防止死循环。 4降级兜底重试耗尽返回业务降级结果不抛异常崩溃。核心能力3多轮对话编排将复杂任务拆解为多个阶段每个阶段有明确的输入/输出/契约AI 按阶段顺序执行支持中断恢复。 常见实现模式 1、Prompt Chain 提示词链串行执行多个子 Prompt上一轮输出作为下一轮输入。 2、状态机编排维护会话状态根据状态决定下一步执行什么 Prompt / 调用什么工具 3、分支条件跳转根据上一轮输出结果选择不同分支不同分支加载不同 Prompt 片段。 4、结合前面上下文管理器每一轮执行前做上下文裁剪、老旧 tool 消息过滤控制 token。三、上下文管理核心要点❌ 不要直接把消息字符串中间截断会出现残缺 JSON、残缺指令要以完整对话单元为单位删除。 ❌ System 提示词不要裁剪一旦丢失模型直接丢失业务约束。 ❌ 不要无限累积历史用户会话跑 100 轮后 token 会爆炸导致接口报错、耗时暴涨。 ⚠️ 摘要不能过度压缩关键参数订单号、金额、业务 id务必保留。1、基础方案滑动窗口固定保留最近 N 轮最简单工业界最常用。逻辑只保留最近k轮对话更早的历史直接丢弃。实现维护messages消息数组超过最大轮数从头部删掉旧消息。优点实现简单token 可控缺点会丢失早期重要信息长会话前面的业务记忆直接没了。MAX_ROUND 6 messages [sys_prompt, user1, ai1, user2, ai2, user3, ai3 ...] # 超过轮数删除最早的用户助手对系统提示词保留不动 while len(messages) MAX_ROUND 1: del messages[1:3]2、基于 Token 的截断比按轮数更精准不要按轮数切统计 token 数量当总 token 接近模型上限从头部裁剪历史消息系统 prompt 永远保留。关键点系统 System Prompt 优先保留业务历史优先删旧的。注意不能粗暴截断单条消息中间会造成文本残缺要按完整对话单元删除。max_context_tokens 3000 # 循环计算总token超阈值就删掉最老的一轮(usrassistant) while count_tokens(messages) max_context_tokens: # 删除最早的一对用户助手消息索引跳过system(0号) del messages[1:3]3、上下文摘要压缩解决长对话丢失早期关键信息滑动窗口会丢老信息摘要把久远历史浓缩成一小段总结文本代替原始完整对话。执行时机当历史消息 token 超过阈值。优点不会丢失早期关键业务信息缺点增加一轮 LLM 调用带来额外开销摘要会存在信息失真风险。流程把老旧的历史消息发给 LLM让模型总结把下面对话浓缩成简短业务摘要保留关键业务参数、用户诉求将原始旧消息全部删除把生成的摘要文本插入 messages保留最近 N 轮原始对话 历史摘要 system 提示词。[ {role:system,content:业务角色指令}, {role:assistant,content:【历史摘要】用户之前申请动产融资额度50万关注放款时效}, # 压缩后的旧会话 {role:user,content:最近用户问题1}, # 最近几轮完整原始对话 {role:assistant,content:回答1}, {role:user,content:最近用户问题2}, {role:assistant,content:回答2}, ]4、分层记忆Agent 常用短期记忆 长期记忆短期记忆messages 数组存放最近几轮原始对话滑动窗口管理长期记忆外部向量库 / 数据库把久远对话做向量化存储。当用户新 query 进来做向量检索把相关的历史片段召回拼入本轮 prompt不相关的旧对话不塞进上下文。典型用于 AI Agent重要业务信息持久化存入向量库不占实时上下文窗口。缺点检索不准会带来幻觉需要做召回过滤。5、消息角色区分过滤无效消息线上业务经常产生大量工具返回、中间思考日志不要无脑全部塞给后续轮次system必须保留核心业务规则用户 user不能丢业务输入assistant正常回答保留Agent 内部思考 / ReAct 中间思考日志可以选择性丢弃tool 返回只保留上一轮工具返回更早的工具调用结果直接丢弃。很多新手踩坑把几十轮完整工具调用全部塞进 messagestoken 直接爆炸。6、业务侧特殊优化工程实践区分会话隔离不同用户、不同业务会话独立维护 messages不要混在一起。Prompt 片段隔离动态业务数据RAG 检索结果、工具返回单独控制长度做截断。开关控制短会话不走压缩长会话才触发摘要 / 裁剪。兜底保护无论什么策略设置硬 token 上限防止异常数据打满上下文。四、上下文管理核心实现1、tiktoken 精确 token 计数# 优先调 LLM 的 Tokenizer 接口精确 TOKENIZER_URL f{BASE_URL}/xxx/xxx/tokenizer/estimate-token-count resp requests.post(TOKENIZER_URL, ...) # 降级接口挂了就用字符数估算粗略 return sum(len(m[content]) for m in clean) // 2def count_tokens(messages: list[dict], model: str JoyAI-LLM-Flash) - int: clean [{role: m[role], content: m[content]} for m in messages if m.get(content)] if not clean: return 0 try: resp requests.post( TOKENIZER_URL, headers{api-key: API_KEY, Content-Type: application/json}, json{messages: clean, model: model}, timeout30, ) resp.raise_for_status() data resp.json().get(data, {}) count data.get(total_tokens) or data.get(token_count) or 0 return int(count) except Exception as e: print(f[token计数] 接口异常降级估算: {e}) return sum(len(m[content]) for m in clean) // 22、滑动窗口裁剪深拷贝数据 def trim(self, messages: list[dict]) - list[dict]: result copy.deepcopy(messages) current_tokens count_tokens(result, self.model) print(f[SlideWindowTrim] 裁剪前: token{current_tokens}, 上限{self.max_total_tokens}, 消息数{len(result)}) if current_tokens self.max_total_tokens: result self._sliding_trim(result) print(f[SlideWindowTrim] 裁剪后: token{count_tokens(result, self.model)}, 消息数{len(result)}) else: print(f[SlideWindowTrim] token{current_tokens} 未超上限无需裁剪) return result #真正的滑动窗口裁剪逻辑默认保留system消息 def _sliding_trim(self, messages: list[dict]) - list[dict]: sys_count sum(1 for m in messages if m[role] system) min_keep self.keep_last_n_round * 2 history_count len(messages) - sys_count trim_count 0 while count_tokens(messages, self.model) self.max_total_tokens: if history_count min_keep: print(f[SlideWindowTrim] 已达到最小保留轮数({self.keep_last_n_round}轮{min_keep}条)停止裁剪) break #下面3行是裁剪循环的核心操作 del messages[sys_count: sys_count 2] history_count - 2 trim_count 1 print(f[SlideWindowTrim] 第{trim_count}次裁剪: 删除1对消息, 当前token{count_tokens(messages, self.model)}, 剩余消息数{len(messages)}) print(f[SlideWindowTrim] 滑动窗口裁剪完成: 共删除{trim_count}对消息 if trim_count 0 else [SlideWindowTrim] 未执行裁剪已达最小保留轮数或token已达标) return messages3、保留最近 N 轮原始对话达到阈值自动触发历史摘要压缩#分割旧对话和近期对话 def split_old_and_recent( history: list[dict], keep_last_n_round: int 4 ) - tuple[list[dict], list[dict]]: keep_msg_count keep_last_n_round * 2 if len(history) keep_msg_count: return [], history return history[:-keep_msg_count], history[-keep_msg_count:] #旧消息压缩为摘要 def get_summary_once( old_history: list[dict], model: str JoyAI-LLM-Flash, temperature: float 0.3, max_tokens: int 500, ) - str: if not old_history: return conversation_text \n.join( f[{m[role]}]: {m[content]} for m in old_history ) prompt f 请对下面的对话历史做精简摘要。 要求保留用户核心诉求、业务关键数字、订单、金额、ID等关键信息去掉冗余闲聊。 不要输出多余解释直接输出摘要文本。 对话历史 {conversation_text} try: from openai import OpenAI client OpenAI(api_keyAPI_KEY, base_urlSUMMARY_BASE_URL) resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, max_tokensmax_tokens, ) summary resp.choices[0].message.content.strip() print(f[SummaryOnce] 生成摘要: {summary[:100]}...) return summary except Exception as e: print(f[SummaryOnce] 摘要生成失败降级为简单拼接: {e}) return | .join(f{m[role]}:{m[content][:50]} for m in old_history) #摘要文字消息插到近期对话之前 def make_summary_message(summary_text: str) - dict: return {role: assistant, content: f【历史对话摘要】:\n{summary_text}}4、增加一个开关可以关闭摘要功能只保留滑动窗口# 开关True 开启摘要False 关闭 SUMMARY_ON False def split_old_and_recent( history: list[dict], keep_last_n_round: int 4 ) - tuple[list[dict], list[dict]]: keep_msg_count keep_last_n_round * 2 if len(history) keep_msg_count: return [], history return history[:-keep_msg_count], history[-keep_msg_count:] def get_summary_once( old_history: list[dict], model: str JoyAI-LLM-Flash, temperature: float 0.3, max_tokens: int 500, ) - str: if not old_history: return conversation_text \n.join( f[{m[role]}]: {m[content]} for m in old_history ) prompt f 请对下面的对话历史做精简摘要。 要求保留用户核心诉求、业务关键数字、订单、金额、ID等关键信息去掉冗余闲聊。 不要输出多余解释直接输出摘要文本。 对话历史 {conversation_text} try: from openai import OpenAI client OpenAI(api_keyAPI_KEY, base_urlSUMMARY_BASE_URL) resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, max_tokensmax_tokens, ) summary resp.choices[0].message.content.strip() print(f[SummaryOnce] 生成摘要: {summary[:100]}...) return summary except Exception as e: print(f[SummaryOnce] 摘要生成失败降级为简单拼接: {e}) return | .join(f{m[role]}:{m[content][:50]} for m in old_history) def make_summary_message(summary_text: str) - dict: return {role: assistant, content: f【历史对话摘要】:\n{summary_text}} def build_context_messages(history, keep_last_n_round4): if not SUMMARY_ON: return list(history) old_part, recent_part split_old_and_recent(history, keep_last_n_round) if not old_part: return list(history) return [make_summary_message(get_summary_once(old_part))] recent_part5、reAct Agent 场景ReAct Reason Act每圈循环就是一个「思考 → 行动 → 观察」周期。经典的Thought → Action → Observation三段式循环Thought: 我该做什么LLM 推理Action: 调哪个工具、传什么参数LLM 输出Observation: 工具执行结果程序执行后喂回→ 回到 Thought直到能给出 Final AnswerMAX_TOTAL_TOKENS 500 TRIMMER SlideWindowTrimmer(max_total_tokensMAX_TOTAL_TOKENS, keep_last_n_round3) SYSTEM_PROMPT 你是贷款业务助手可以通过调用工具解决问题。 工具清单 1. query_order(order_id)查询订单状态 2. calc(amount, rate, years)计算贷款总利息amount本金rate年利率years年限 请严格按以下格式回复一次只执行一步 Thought: 思考当前情况决定下一步 Action: 工具名(参数) 当你已经能给出最终答案时回复 Thought: 我已经得到答案 Final Answer: 最终回答 def query_order(order_id): orders { PRE202601001: {status: 预审中, amount: 800000}, PRE202601002: {status: 已放款, amount: 1200000}, } return orders.get(order_id, 订单不存在) def calc(amount, rate, years): return amount * rate * years def manage_context(messages): tokens count_tokens(messages) print(f[Context] 当前 token{tokens}, 上限{MAX_TOTAL_TOKENS}) if tokens MAX_TOTAL_TOKENS: return messages if SUMMARY_ON: system, history messages[:1], messages[1:] messages system build_context_messages(history, keep_last_n_round4) return TRIMMER.trim(messages) def run_agent(question, max_steps5): client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: question}, ] for step in range(1, max_steps 1): messages manage_context(messages) resp client.chat.completions.create( modelMODEL, messagesmessages, temperature0, ) reply resp.choices[0].message.content.strip() messages.append({role: assistant, content: reply}) print(f\n---- 第{step}步 ----\n{reply}\n) if Final Answer: in reply: return reply.split(Final Answer:)[-1].strip() m re.search(rAction:\s*(\w)\((.*)\), reply) if not m: return 模型未按格式输出流程终止 name, args_raw m.group(1), m.group(2).strip() try: observation eval(f{name}({args_raw})) except NameError: observation eval(f{name}({args_raw})) except Exception as e: observation f工具执行失败: {e} print(f[Tool] {name}({args_raw}) - {observation}) messages.append({role: user, content: fObservation: {observation}}) return 达到最大步数流程终止