企业官网建设流程全解析

简介：《构建上下文感知的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就完成，非要做向量化，既慢又没必要。

给一张我常用的选型参考表。

落库之前的另一个决定是保留期。用户画像可以保留很久，会话详情最多留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_factory=list) 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_factory=list) recent_topics: List[str] = Field(default_factory=list) 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 f"<context>\n{self.model_dump_json(exclude_none=True)}\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_id=request_id, user_id=user_id, session_id=session_id, event_type="api_access", entity_type="", entity_id="", content=json.dumps(body_summary, ensure_ascii=False), event_time=event_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_type='health_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(key=lambda x: x[0], reverse=True) 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 过期上下文未设保鲜期：让系统把昨天的结论当今天的事实

现象：用户上午问过“你们的退货政策”，下午又故意问同一个问题，系统直接沿用上午的答复，哪怕政策已经在后台更新了。

原因：上下文采集和业务数据更新是两个异步流程，上下文库里存的是上午的快照，业务表已经变了，但检索时没有校验数据的新鲜度。

解决：在检索逻辑里对业务上下文做强制保鲜。方案是给业务类上下文单独设置检查阈值：库存、价格这类高频变化的数据，禁止从上下文仓库读取，必须实时调用业务接口；只有用户偏好这类低频数据可以走缓存。我在系统里加了一个数据源标记字段，拼装提示词时对source='cache'的条目做时间戳校验，超过保留期直接丢弃并触发实时查询。

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做幂等，后到的重复消息直接丢弃。

给一个事件写入时的字段对照，方便排查这类错位问题。

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_prompt=system_prompt, user_message=query) def run_with_context(query: str, unified_context: dict) -> str: context_block = f"<context>{unified_context}</context>" return call_llm( system_prompt=SYSTEM_PROMPT + "\n" + context_block, user_message=query, ) 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过滤条件被新代码漏写了。从那以后，每个上下文检索函数都强制带用户维度断言，评测集里也固定加一个“跨用户隔离”用例。这套一页纸的评测集改动不大，但能兜住大多数回归问题，希望帮到你。

本文还有配套的精品资源，点击获取