简介《构建上下文感知的AI应用》是一本面向生成式AI工程师、机器学习开发者及企业技术决策者的实战型电子书旨在解决大语言模型在真实业务中知识截止、幻觉频发、难以处理多模态数据等痛点。全书围绕AI项目全生命周期展开系统阐述模型选型、提示工程、微调、部署优化等关键环节并重点分析检索增强生成RAG与智能体Agent协同的上下文感知推理架构使模型能够即时引用外部知识并执行复杂任务。同时结合Amazon Bedrock等托管服务演示如何融合文本、图像、音频等多模态数据构建具备非语言推理能力的智能系统借助LangChain与ReAct框架实现对外部API和私有数据源的动态访问赋能企业级AI应用落地。资源包为PDF格式电子书文件总数1个压缩包大小84.56MB目前已有126人学习。读者可从中掌握从架构设计到工程实现的完整方法并获得基于AWS云服务的生成式AI应用开发范例。1. 上下文感知的AI应用为什么同一个问题换个场景问就得到另一个答案你在做AI应用开发时一定撞过这一幕用户早上问“今天适合跑步吗”助手回答“空气良好适合慢跑”下午同一个用户换了手机登录再问同一句话得到的是“请提供您所在的城市”。模型没变提示词没变变的是对话历史、设备位置和账号身份——这些信息丢了应用就退化成一句一问的聊天玩具。上下文感知的AI应用核心就是把用户、会话、环境和业务四类信息在请求到达模型之前补齐让同一个模型在不同场景下给出不同答案。它不是某个模型自带的能力而是应用层要主动构建的输入工程。这篇文章想把这套方案的选型、实现和踩坑点一次讲清适合正在做客服助手、智能体编排或个人知识库的开发者参考。2. 拆解上下文感知的四个输入源先想清楚该感知什么再选AI应用技术栈2.1 用户上下文、会话上下文、环境上下文、业务上下文缺一层都会失真做上下文感知的AI应用先别急着接向量数据库第一步是盘点你的应用能拿到哪些信号。我习惯把上下文拆成四层来看。用户上下文是长期画像包括偏好、历史订单、会员等级、常用语言。它变化慢但最能决定回复基调。老客户报修和首访客报修语气和流程完全不同。会话上下文是多轮对话里的短期状态包括当前正在问的问题、刚刚提到的商品型号、已经确认过的约束条件。它变化最快丢了它就会出现“刚才还说选蓝色的现在问尺寸是什么颜色”这种翻车。环境上下文来自设备、时间和位置。移动端还是桌面端工作日还是深夜定位在城市还是郊区直接决定信息优先级。比如查“附近的药店”没有定位就只能返回通用列表有了定位才能按距离排序。业务上下文是应用特有的对象状态。客服场景里是工单号对应的处理进度购票场景里是座位图和余票数这类数据通常不在对话里出现必须从业务系统主动查询。四层里最容易漏的是这一层因为它需要写代码去对接内部接口而不能只从请求参数里捞。判断上下文是否完备有一个简单标准把应用收到的会话记录打出来遮住当前这轮用户输入让一个人猜用户想干什么。如果猜不出大概率是上下文信息不完整。这个排查动作我在新项目启动时都会做一次。2.2 上下文存储选型从Redis、SQLite到pgvector别把KV当万能药确定要感知哪些信号之后接着要决定这些信号放在哪里。存储选型直接影响延迟和一致性不同上下文对存储的要求差别很大。会话上下文适合放在Redis这类内存数据库里。它写入频繁、过期快、单条数据不大用TTL管理多轮窗口非常顺。以6小时会话窗口为例每轮请求都把新的消息追加到Redis的List结构配合过期时间基本不需要额外写清理任务。用户上下文和业务上下文适合放在PostgreSQL这类关系库里。它们结构稳定、需要按条件查询而且经常要和订单表、商品表做关联。把这类数据塞进Redis等要做条件筛选时你就得全量拉出来在代码里过滤越往后越难受。如果检索场景涉及语义相似度比如“找一条历史上相似故障的处理记录”就在PostgreSQL上挂pgvector扩展或者单独部署一个向量库。但这里提醒一句向量检索只是检索招数之一不是上下文感知的银弹。电商场景里“查用户上次买的型号”用一条SQL就完成非要做向量化既慢又没必要。给一张我常用的选型参考表。上下文类型典型数据存储访问模式过期策略会话上下文多轮对话记录、临时选择Redis/内存追加、范围读取TTL 6小时用户上下文偏好、标签、历史摘要PostgreSQL点查、条件查询长期保留环境上下文时间、地域、设备请求处理时采集即时读取不落库业务上下文工单状态、库存、订单进度PostgreSQL/业务API点查、联表查询按业务周期落库之前的另一个决定是保留期。用户画像可以保留很久会话详情最多留7天环境信息只用当次请求。保留期没有想清楚的系统半年之后存储膨胀还得回来补清理任务。2.3 用JSON Schema和Pydantic给上下文设边界先约定再采集采集上下文之前先定义它的数据结构这是我在做过几个翻车项目后养成的习惯。没有边界的数据后端采集时漏字段前端调用时字段名对不上排错要花掉半天。我用Pydantic定义一个上下文物件两层结构外层是用户和会话标识内层按四类信号分块。这样每个字段在接入时就有明确的默认值模型端也能拿到统一的JSON结构。from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class EnvironmentContext(BaseModel): timezone: Optional[str] Asia/Shanghai device_type: Optional[str] web region_code: Optional[str] None class UserContext(BaseModel): user_id: str member_level: int 0 preferred_lang: str zh-CN preference_tags: List[str] Field(default_factorylist) class BusinessContext(BaseModel): entity_type: str Field(..., examples[order, ticket, product]) entity_id: str Field(..., examples[SO2024001]) entity_status: Optional[dict] None class SessionContext(BaseModel): session_id: str dialog_tags: List[str] Field(default_factorylist) recent_topics: List[str] Field(default_factorylist) class UnifiedContext(BaseModel): request_id: str timestamp: datetime user: UserContext session: SessionContext environment: EnvironmentContext business: Optional[BusinessContext] None def to_prompt_block(self) - str: return fcontext\n{self.model_dump_json(exclude_noneTrue)}\n/context这段代码把上下文分成四个子模型每个模型只负责自己的字段。to_prompt_block方法把对象序列化成JSON字符串方便后续拼进系统提示词。注意我给每个字段都设了默认值特别是business是可选的避免构造请求时因为某个业务字段缺失而直接抛异常。实际接入时通常会遇到业务方传参不齐的情况。我的做法是在采集入口做一次校验把校验失败的请求记录到日志同时用带默认值的模型继续向下游传递保证主流程不被上下文采集拖死。字段命名提前统一也很重要比如user_id还是userId在接口对接阶段就必须敲定混用会让检索和拼装逻辑越写越乱。3. 给AI应用装上记忆层从请求日志到可检索的上下文仓库3.1 用FastAPI中间件采集会话事件不侵入业务代码的接入方式上下文数据不能靠业务代码每个接口手动攒那会漏得千疮百孔。常见做法是在网关层或框架中间件统一采集。下面是一个FastAPI中间件的实现它拦截每个请求把与上下文相关的关键事件写入SQLite。import json import sqlite3 from datetime import datetime, timezone from fastapi import FastAPI, Request from starlette.middleware.base import BaseHTTPMiddleware app FastAPI() DB_PATH context_events.db def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute( CREATE TABLE IF NOT EXISTS context_events ( event_id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT, user_id TEXT, session_id TEXT, event_type TEXT, entity_type TEXT, entity_id TEXT, content TEXT, event_time INTEGER ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_session_time ON context_events(session_id, event_time)) init_db() class ContextCollectMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 只采集接口路径不采集静态资源 if request.url.path.startswith(/static): return await call_next(request) request_id request.headers.get(X-Request-Id, ) user_id request.headers.get(X-User-Id, ) session_id request.headers.get(X-Session-Id, ) event_time int(datetime.now(timezone.utc).timestamp() * 1000) response await call_next(request) # 响应正常才算一次有效业务事件 if 200 response.status_code 400: body_summary self._summarize_request(request) self._write_event( request_idrequest_id, user_iduser_id, session_idsession_id, event_typeapi_access, entity_type, entity_id, contentjson.dumps(body_summary, ensure_asciiFalse), event_timeevent_time, ) return response def _summarize_request(self, request: Request) - dict: # 传query参数和路径参数不dump整个body避免塞入敏感数据 return {path: request.url.path, query: dict(request.query_params)} def _write_event(self, **fields): with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT INTO context_events (request_id, user_id, session_id, event_type, entity_type, entity_id, content, event_time) VALUES (:request_id, :user_id, :session_id, :event_type, :entity_type, :entity_id, :content, :event_time), fields, ) app.add_middleware(ContextCollectMiddleware)这个中间件只做了三件事从请求头读取用户和会话标识调用下游接口拿到响应状态最后把事件写进SQLite。代码里的entity_type和entity_id字段是给业务上下文预留的比如一个工单查询接口可以在这层额外解析路径参数/orders/SO2024001把订单号填进entity_id后续检索时就能精确拉起这个订单的状态。采集层有一个参数需要特别调事件时间使用的是UTC毫秒时间戳而不是接收请求时的本地时间。时间在上下文系统里是最关键的对齐依据后面做保鲜和排序时全靠它。我在项目里见过直接写datetime.now()的部署到不同时区后上下文顺序就乱了这个字段必须统一。3.2 上下文清洗与去重写入仓库前的三道闸口采集到的事件不能直接当上下文用原始请求日志里一半以上都是噪音。我在写入之前加三道闸口。第一道是去重。中间件在某些接口重试时会重复写同一条事件靠request_id做唯一性校验能挡住大部分重复。SQLite这边没有唯一的约束所以在写入前先查一次或者把唯一索引加上。用一个INSERT OR IGNORE就能处理。第二道是过滤。健康检查、静态资源、灰度探活这类请求不能进上下文库。我在事件表里增加一个event_type字段只有在白名单里的事件类型才进检索。检索时再加一个排除条件把event_typehealth_check的条目直接过滤掉。第三道是截断和脱敏。长文本正文存入之前只保留前2000个字符对涉及卡号、手机号等字段打码。这一步要在文本进入数据库之前做而不是检索之后做因为检索到的片段可能直接拼进提示词发给模型脱敏不及早处理就会泄露。清洗后的事件进入一个分层的查询接口查询会话上下文用SQLite的范围查询查询用户上下文用Redis的点查查询业务上下文则实时调用内部接口。三种数据来源在拼装层统一汇合。3.3 检索并拼装Prompt把相关上下文放回请求的最小路径采集是基础真正决定用户体验的是检索质量。下面这段函数从SQLite里取回一个会话最近N分钟内的相关事件并按时间和主题相关度各占一半权重做排序。import sqlite3, time, math DEFAULT_WINDOW_MINUTES 60 DEFAULT_TOP_K 5 DECAY_FACTOR 0.1 def retrieve_context(session_id: str, query_text: str, window_minutes: int DEFAULT_WINDOW_MINUTES, top_k: int DEFAULT_TOP_K): now_ms int(time.time() * 1000) since_ms now_ms - window_minutes * 60 * 1000 with sqlite3.connect(context_events.db) as conn: rows conn.execute( SELECT content, event_time, event_type FROM context_events WHERE session_id ? AND event_time ? ORDER BY event_time DESC LIMIT 100, (session_id, since_ms), ).fetchall() scored [] for content, event_time, event_type in rows: if not content: continue # 时间相关度越新分越高 age_hours (now_ms - event_time) / 3600000.0 time_score math.exp(-DECAY_FACTOR * age_hours) # 内容相关度查询词出现的次数简单计分 content_score content.count(query_text) * 0.1 scored.append((time_score content_score, content, event_time, event_type)) scored.sort(keylambda x: x[0], reverseTrue) return [{content: s[1], event_time: s[2], event_type: s[3]} for s in scored[:top_k]] def build_prompt(system_prompt: str, user_query: str, context_items: list) - str: context_text \n.join( f- [{item[event_type]}] {item[content]} for item in context_items ) return ( f{system_prompt}\n\n本应用可参考的上下文信息\n{context_text}\n\n f用户当前问题{user_query} )retrieve_context里有两个参数值得细讲。window_minutes控制时间范围默认60分钟客服场景比较合适如果是工单处理这类业务上下文关联强的场景建议拉长到24小时并降低时间权重否则用户早上反馈的问题下午就查不到了。DECAY_FACTOR控制时间衰减速度数值越大旧信息的权重掉得越快一般取值范围在0.05到0.2之间。时间衰减用指数函数而不是硬切窗口这样不会出现“59分钟内的事件全保留61分钟前的事件全抹掉”的突兀跳变。检索结果拼进提示词时放在系统提示词后面模型看到上下文的顺序是系统规则、上下文信息、用户当前问题。这个顺序在实践中比把上下文放在用户问题后面更不容易被忽略。4. 上下文感知落地避坑指南上下文补齐后效果变差的5个原因4.1 盲目把整段历史塞进上下文token爆炸且模型开始复读现象接入上下文感知后多轮对话质量反而下降。模型开始复述用户早先说过的话或者答非所问。原因把整个会话原文按时间顺序全量塞进Prompt上下文动辄几千字模型注意力被中间冗余信息稀释。研究结论和工程直觉都指向同一个方向——上下文越长关键信息越容易淹没在无关细节里同时token成本直接拖慢响应速度。解决给上下文设预算。我一般把上下文总token控制在512到1024之间超过的部分做摘要只保留最近3轮原文。历史对话摘要可以存在Redis里过期的会话直接清掉不参与检索。检索事件同样设上限top_k默认给5宁可少给也不给全。4.2 过期上下文未设保鲜期让系统把昨天的结论当今天的事实现象用户上午问过“你们的退货政策”下午又故意问同一个问题系统直接沿用上午的答复哪怕政策已经在后台更新了。原因上下文采集和业务数据更新是两个异步流程上下文库里存的是上午的快照业务表已经变了但检索时没有校验数据的新鲜度。解决在检索逻辑里对业务上下文做强制保鲜。方案是给业务类上下文单独设置检查阈值库存、价格这类高频变化的数据禁止从上下文仓库读取必须实时调用业务接口只有用户偏好这类低频数据可以走缓存。我在系统里加了一个数据源标记字段拼装提示词时对sourcecache的条目做时间戳校验超过保留期直接丢弃并触发实时查询。4.3 上下文漂移导致用户信息串号现象用户A的对话历史出现在用户B的上下文里客服系统里表现为“这个用户明明是新客户系统却称呼他王先生”。原因session_id 全局唯一没做到位。有的实现把 session_id 生成放在前端用户清掉Cookie后生成了新ID但用户标识user_id没传系统用了当前请求的IP或设备指纹临时拼ID不同用户落在同一台设备上就串了。解决采集层强制以user_id为主维度session_id只做二级维度检索时两个条件必须同时匹配。写入时缺user_id的事件直接放弃不写默认值。另外要对上下文内容里的手机号、地址等字段做正则脱敏即使真的串号也不能把敏感信息原样带到模型侧。4.4 异步事件乱序导致上下文错位现象用户先下单后改地址两次操作间隔10秒但上下文库里改地址的事件排在前面系统在下单回复里引用了修改后的地址和业务单据对不上。原因中间件写入使用的是服务端接收时间arrival_time而不是业务事件发生时间。异步消息在消息队列里会堆积、重试接收时间完全不能代表事件的业务时序。解决事件结构里必须带业务侧时间字段event_time由业务模块在事件产生的瞬间生成采集层只负责透传。排序、保鲜、窗口截断全部基于event_time计算。同一条事件被重复投递时靠request_id做幂等后到的重复消息直接丢弃。给一个事件写入时的字段对照方便排查这类错位问题。字段含义来源注意事项event_time业务实际发生时间业务系统用UTC毫秒不要用接收时间request_id请求唯一标识网关生成重试时保持相同值received_time采集系统接收时间中间件只用于监控不参与排序4.5 检索topK和时效性窗口冲突相关但过期的内容挤掉了有效内容现象检索出来的上下文全部是昨天相似问题的处理记录今天刚发生的关键事件反而排在最后模型回答偏向陈旧。原因retrieve_context里时间衰减权重设得太低内容相关度词频计分直接把旧内容顶上来。内容含查询词次数越多分越高但业务场景里语义相关并不等于字面匹配一条旧工单里正好塞满了关键词也会得高分。解决把时间窗口和衰减放在排名第一阶段强制截断先过滤掉超出保鲜期的内容再做内容打分。我在项目里把衰减系数从0.1调到0.3后当日事件出现在第一位的情况明显变多。另一个做法是给事件类型加优先级业务状态变更类事件的权重高于普通对话事件因为这类信息决定回复的事实基础。5. 给上下文感知做一页纸评测用对照集验证你的改动真有效上下文感知的效果不能靠“感觉回答得更聪明了”来验收那样和拍脑袋没有区别。我建了一套只有一页纸的评测方法叫“上下文响应对照集”。核心思想是每个用例设两个上下文版本一个关掉上下文一个打开上下文期望行为必须有可见差异。比如场景“用户问退货时效”关上下文时模型回答“请咨询客服”打开上下文时模型回答“您所在地区支持7天无理由退货”。如果同样的输入在两种上下文的输出完全一致说明上下文链路没起作用这个用例不通过。建对照集时测试用例要覆盖四类信号多轮会话里的指代消解、用户画像带来的偏好调整、环境变化引起的答案差异、业务状态变更后的口径调整。每类至少3个用例合起来12个用例足够发现大部分回归问题。下面是一个只跑验证不接数据库的最小评测脚本结构架构上把布了上下文的链路和对照组都封装成函数测试代码逐条调用比对。def run_without_context(query: str, system_prompt: str) - str: return call_llm(system_promptsystem_prompt, user_messagequery) def run_with_context(query: str, unified_context: dict) - str: context_block fcontext{unified_context}/context return call_llm( system_promptSYSTEM_PROMPT \n context_block, user_messagequery, ) case { query: 今天适合跑步吗, without_context_expected_includes: [], with_context_expected_includes: [当前的空气质量指数], context: { environment: {region_code: SH, timezone: Asia/Shanghai}, business: {entity_type: weather, entity_id: live_air_quality}, }, } output_no_ctx run_without_context(case[query], SYSTEM_PROMPT) output_ctx run_with_context(case[query], case[context]) assert any(keyword in output_ctx for keyword in case[with_context_expected_includes]) assert not any(keyword in output_no_ctx for keyword in case[with_context_expected_includes])评测通过之后再跑一次成本检查对比打开上下文前后的输入token数量和响应时延确保上下文链路带来的收益大于成本。我在实际项目中做过的记录是打开上下文后回答准确率提升了三成左右但token消耗也攀升了接近一倍。如果发现有上下文和没上下文的输出差别不大优先查检索是否真的召回了正确条目而不是急着换更大的模型。评测集要纳入每次改动后的回归流程。新增一个业务字段、调整时间衰减系数、切换检索排序算法都要重新跑一遍对照集。我吃过一次亏优化了向量检索的召回参数测试集上效果好但上生产后用户多聊几轮就推到历史问题上去最后发现是会话窗口检索里user_id过滤条件被新代码漏写了。从那以后每个上下文检索函数都强制带用户维度断言评测集里也固定加一个“跨用户隔离”用例。这套一页纸的评测集改动不大但能兜住大多数回归问题希望帮到你。本文还有配套的精品资源点击获取