引言Context Engineering 的崛起2025年AI应用正从单次交互转向复杂的智能体系统。传统Prompt Engineering提示工程的局限性日益凸显——它过于依赖单次指令的设计缺乏对多源数据流的整合能力。而Context Engineering上下文工程则提供了全新的思路不是优化单次指令而是构建动态上下文系统。这个转变对我们团队而言不是理论探讨而是业务倒逼的现实。随着知识库规模膨胀到数万篇文档、机器人日均问答量突破千次传统“检索→拼装→生成”的Naive RAG模式已经捉襟见肘。我们需要一套能够支撑业务落地的Context Infra——从团队知识库的供给管理到机器人对话入口层的Skill与工具产品开发再到Agent工作流的设计与实现。本文将完整分享我们在这一过程中的架构设计、技术选型与工程实践。文章涵盖从RAG基础架构到Agentic RAG的演进路径并附上可直接运行的代码示例。一、Context Infra 的三层架构在系统设计之初我们参考了行业对Context Infrastructure的通用定义它应该跨越协议层MCP、交付层RAG与上下文图谱和受治理的数据基座三个层次。基于这个框架我们将自己的Context Infra划分为三层数据层负责知识库的供给管理包括文档采集、清洗、分块、向量化与存储。这一层的核心目标是“让知识可检索”。检索层负责向量检索、混合检索、重排序与上下文组织。这一层的核心目标是“让检索更精准”。执行层负责Agent工作流编排、Skill与工具调用、多轮对话管理。这一层的核心目标是“让智能可编排”。三层之间通过标准化的数据接口串联每一层都可独立迭代优化。二、数据层从文档到向量2.1 技术选型在向量数据库的选型上我们遵循了一个务实的原则FAISS够用就别上Milvus。FAISS零依赖、速度最快50万条以内完全够用。我们的知识库当前规模在10万条级别FAISS完全能够满足需求且省去了维护分布式数据库的运维成本。Embedding模型选择了BAAI/bge-small-zh-v1.5——512维支持本地推理CPU也能跑编码速度约1000条/秒。中文场景下表现扎实性价比极高。2.2 核心代码实现以下是数据层核心的文档入库流程pythonimport osimport jsonfrom typing import List, Dictimport faissimport numpy as npfrom sentence_transformers import SentenceTransformerclass VectorStore:“”“基于FAISS的轻量级向量存储”“”def __init__(self, model_name: str BAAI/bge-small-zh-v1.5): self.model SentenceTransformer(model_name) self.dimension 512 self.index None self.documents: List[Dict] [] def embed(self, text: str) - np.ndarray: 将文本转换为向量注意添加检索前缀 # bge模型要求添加检索前缀以获得更好的检索效果[reference:6] text_with_prefix f为这个句子生成表示以用于检索{text} return self.model.encode(text_with_prefix, normalize_embeddingsTrue) def chunk_document(self, text: str, chunk_size: int 512, overlap: int 50) - List[str]: 语义分块按句子边界切分避免截断语义单元 sentences text.replace(\n, ).split(。) chunks [] current_chunk [] current_len 0 for sent in sentences: sent sent.strip() if not sent: continue # 简单起见按字符数控制chunk大小生产环境建议使用更精细的语义分块策略 if current_len len(sent) chunk_size and current_chunk: chunks.append(。.join(current_chunk) 。) # 保留overlap部分的句子以维持上下文连贯性 overlap_sents current_chunk[-2:] if len(current_chunk) 2 else current_chunk current_chunk overlap_sents.copy() current_len sum(len(s) for s in current_chunk) current_chunk.append(sent) current_len len(sent) if current_chunk: chunks.append(。.join(current_chunk) 。) return chunks def ingest(self, documents: List[Dict[str, str]]): 批量入库 documents: [{id: doc_001, text: ..., source: wiki/xxx}] all_embeddings [] for doc in documents: chunks self.chunk_document(doc[text]) for chunk in chunks: emb self.embed(chunk) all_embeddings.append(emb) self.documents.append({ id: doc.get(id, fchunk_{len(self.documents)}), text: chunk, source: doc.get(source, unknown), embedding: emb.tolist() }) if not all_embeddings: return embeddings_matrix np.vstack(all_embeddings).astype(float32) if self.index is None: # 使用内积相似度余弦相似度的等价形式因为向量已归一化 self.index faiss.IndexFlatIP(self.dimension) self.index.add(embeddings_matrix) print(f✅ 已索引 {len(self.documents)} 个文档块) def save(self, path: str): 持久化索引与文档元数据 if self.index is not None: faiss.write_index(self.index, f{path}/index.faiss) with open(f{path}/documents.json, w, encodingutf-8) as f: # 移除embedding字段以减小存储体积 docs_to_save [{k: v for k, v in d.items() if k ! embedding} for d in self.documents] json.dump(docs_to_save, f, ensure_asciiFalse, indent2) def load(self, path: str): 加载持久化的索引与文档 self.index faiss.read_index(f{path}/index.faiss) with open(f{path}/documents.json, r, encodingutf-8) as f: self.documents json.load(f)2.3 踩坑记录在实际落地中我们遇到了几个典型问题分块策略的粒度问题初期采用固定512字符切分导致大量语义单元被截断。例如“如何配置Nginx反向代理”这个完整知识点被切分到两个chunk中检索时召回率大幅下降。我们最终采用了按句子边界切分保留重叠的策略将召回率提升了约15%。Embedding前缀的重要性bge模型要求添加“为这个句子生成表示以用于检索”前缀。初期我们忽略了这一点导致检索效果比预期差20%以上。加上前缀后效果显著改善。三、检索层从向量检索到智能检索3.1 检索架构演进我们的检索层经历了三个阶段的演进第一阶段Naive RAG。用户提问→向量检索Top-K→拼接上下文→LLM生成。这是最基础的形态但问题也很明显检索精度低缺乏上下文过滤无自我校正能力。第二阶段Advanced RAG。引入了混合检索BM25向量检索、重排序Reranking和上下文压缩。这是当前生产环境的主力方案。第三阶段Agentic RAG。检索系统本身作为一个Agent具备自适应检索、工具调用、自我反思等能力。目前正在逐步推进中。3.2 混合检索与重排序以下是检索层的核心实现pythonimport faissimport numpy as npfrom typing import List, Tuplefrom rank_bm25 import BM25Okapiimport jiebaclass HybridRetriever:“”“混合检索器BM25关键词检索 FAISS向量检索”“”def __init__(self, vector_store: VectorStore): self.vector_store vector_store self.bm25_index None self.bm25_corpus [] def build_bm25_index(self): 构建BM25索引用于关键词检索 tokenized_corpus [] for doc in self.vector_store.documents: tokens list(jieba.cut(doc[text])) tokenized_corpus.append(tokens) self.bm25_corpus.append(doc[text]) self.bm25_index BM25Okapi(tokenized_corpus) def search(self, query: str, top_k: int 10, alpha: float 0.5) - List[Tuple[Dict, float]]: 混合检索BM25 向量检索加权融合 alpha: 向量检索的权重1-alpha为BM25的权重 # 1. 向量检索 query_vec self.vector_store.embed(query).reshape(1, -1).astype(float32) distances, indices self.vector_store.index.search(query_vec, top_k * 2) vector_results [] for idx, dist in zip(indices[0], distances[0]): if idx 0 and idx len(self.vector_store.documents): vector_results.append((self.vector_store.documents[idx], float(dist))) # 2. BM25关键词检索 if self.bm25_index is None: self.build_bm25_index() query_tokens list(jieba.cut(query)) bm25_scores self.bm25_index.get_scores(query_tokens) bm25_results [(self.vector_store.documents[i], float(bm25_scores[i])) for i in range(len(self.vector_store.documents)) if bm25_scores[i] 0] bm25_results.sort(keylambda x: x[1], reverseTrue) bm25_results bm25_results[:top_k * 2] # 3. 分数归一化与加权融合RRF - Reciprocal Rank Fusion # 使用更稳定的RRF融合策略[reference:12] doc_scores {} for rank, (doc, score) in enumerate(vector_results): doc_id doc[id] doc_scores[doc_id] doc_scores.get(doc_id, 0) alpha / (60 rank) for rank, (doc, score) in enumerate(bm25_results): doc_id doc[id] doc_scores[doc_id] doc_scores.get(doc_id, 0) (1 - alpha) / (60 rank) # 4. 排序并返回Top-K sorted_docs sorted(doc_scores.items(), keylambda x: x[1], reverseTrue) results [] for doc_id, score in sorted_docs[:top_k]: # 找到对应的文档 for doc in self.vector_store.documents: if doc[id] doc_id: results.append((doc, score)) break return resultsclass Reranker:“”“重排序器对检索结果进行精排”“”def __init__(self): # 生产环境可使用Cross-Encoder模型这里用TF-IDF作为轻量级替代[reference:13] pass def rerank(self, query: str, candidates: List[Tuple[Dict, float]]) - List[Tuple[Dict, float]]: 对候选结果进行重排序 实际生产环境建议使用Cross-Encoder精度更高但延迟也更高[reference:14] # 简单实现按原始分数排序实际生产环境会用专门的reranker模型 return sorted(candidates, keylambda x: x[1], reverseTrue)3.3 上下文压缩检索返回的Top-K文档可能包含大量冗余信息直接拼接会浪费宝贵的上下文窗口。我们引入了上下文压缩机制pythonfrom langchain.chains.combine_documents import create_stuff_documents_chainfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.documents import Documentclass ContextCompressor:“”“上下文压缩器提取检索结果中最相关的信息”“”def __init__(self, llm): self.llm llm def compress(self, query: str, documents: List[Dict], max_tokens: int 4000) - str: 将多个文档压缩为精简的上下文 使用LLM提取与query最相关的信息片段 # 构建压缩提示词 prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息提取专家。请根据用户的问题从以下文档中提取最相关的信息。 只保留直接相关的信息去除冗余和无关内容。用简洁的语言组织成连贯的段落。), (human, 用户问题{query} 文档内容 {context} 请提取相关信息) ]) # 将文档拼接为上下文注意控制总长度 context \n\n---\n\n.join([d[text] for d in documents[:5]]) chain create_stuff_documents_chain(self.llm, prompt) result chain.invoke({ query: query, context: [Document(page_contentcontext)] }) return result四、执行层Agent工作流编排4.1 从静态工作流到动态Agent检索层解决了“找什么”的问题执行层则要解决“怎么做”的问题。在Context Infra中执行层的核心是Agent工作流编排。我们采用LangGraph作为工作流引擎。LangGraph通过有向无环图DAG定义工作流每个节点是一个执行单元节点间通过状态传递数据。相比传统的线性ChainLangGraph支持条件分支、循环、并行等更复杂的编排模式。4.2 ReAct模式的Agent实现ReActReasoning Acting是Agent工作流中最经典的范式。Agent循环执行“思考→行动→观察”的闭环直到任务完成。以下是一个基于LangGraph实现的RAG Agentpythonfrom langgraph.graph import StateGraph, ENDfrom typing import TypedDict, List, Literalfrom langchain_core.messages import HumanMessage, AIMessagefrom langchain.tools import Toolclass AgentState(TypedDict):“”“Agent的状态管理”“”messages: List[dict]query: strretrieved_docs: List[Dict]context: stranswer: striteration: intmax_iterations: intclass RAGAgent:“”“基于LangGraph的RAG Agent”“”def __init__(self, retriever: HybridRetriever, llm, compressor: ContextCompressor): self.retriever retriever self.llm llm self.compressor compressor self.graph self._build_graph() def _build_graph(self): 构建Agent工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, self._retrieve_node) workflow.add_node(compress, self._compress_node) workflow.add_node(generate, self._generate_node) workflow.add_node(reflect, self._reflect_node) # 设置入口 workflow.set_entry_point(retrieve) # 添加边 workflow.add_edge(retrieve, compress) workflow.add_edge(compress, generate) workflow.add_conditional_edges( generate, self._should_continue, { reflect: reflect, end: END } ) workflow.add_edge(reflect, retrieve) return workflow.compile() def _retrieve_node(self, state: AgentState) - AgentState: 检索节点执行混合检索 query state[query] results self.retriever.search(query, top_k10) state[retrieved_docs] [doc for doc, _ in results] state[iteration] state.get(iteration, 0) 1 return state def _compress_node(self, state: AgentState) - AgentState: 压缩节点压缩检索结果 query state[query] docs state[retrieved_docs] state[context] self.compressor.compress(query, docs) return state def _generate_node(self, state: AgentState) - AgentState: 生成节点基于上下文生成回答 from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的知识助手。基于以下上下文回答用户的问题。 如果上下文中没有相关信息请诚实地说不知道不要编造答案。 上下文{context}), (human, {query}) ]) chain prompt | self.llm response chain.invoke({ context: state[context], query: state[query] }) state[answer] response.content state[messages] state.get(messages, []) [ {role: assistant, content: response.content} ] return state def _reflect_node(self, state: AgentState) - AgentState: 反思节点评估答案质量必要时触发重新检索 from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个质量评估专家。评估以下回答是否充分回答了用户的问题。 如果回答不够完整或存在明显遗漏返回need_retrieve否则返回satisfied。 只返回这两个词之一不要有其他内容。), (human, 用户问题{query} 当前回答{answer} 请评估) ]) chain prompt | self.llm result chain.invoke({ query: state[query], answer: state[answer] }) state[need_reflect] need_retrieve in result.content.lower() return state def _should_continue(self, state: AgentState) - Literal[reflect, end]: 决定是否继续迭代 # 如果达到最大迭代次数或答案已满足要求结束 if state[iteration] state.get(max_iterations, 3): return end if not state.get(need_reflect, True): return end return reflect def invoke(self, query: str) - str: 执行Agent initial_state { query: query, messages: [], retrieved_docs: [], context: , answer: , iteration: 0, max_iterations: 3 } result self.graph.invoke(initial_state) return result[answer]4.3 Skill与工具系统除了基础的RAG检索Agent还需要调用各种工具来完成复杂任务。我们设计了一套轻量级的Skill注册与调用机制pythonfrom typing import Callable, Dict, Anyimport functoolsclass SkillRegistry:“”“技能注册中心管理Agent可调用的工具”“”_skills: Dict[str, Dict] {} classmethod def register(cls, name: str, description: str): 装饰器注册技能 def decorator(func: Callable): cls._skills[name] { func: func, description: description, name: name } functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper return decorator classmethod def get_skill(cls, name: str) - Callable: return cls._skills.get(name, {}).get(func) classmethod def list_skills(cls) - List[Dict]: return [{name: k, description: v[description]} for k, v in cls._skills.items()]使用示例SkillRegistry.register(“query_knowledge_base”, “从知识库中检索相关信息”)def query_knowledge_base(query: str, top_k: int 5) - List[Dict]:# 调用检索层passSkillRegistry.register(“fetch_document”, “根据文档ID获取完整文档内容”)def fetch_document(doc_id: str) - str:# 获取完整文档passSkillRegistry.register(“generate_summary”, “生成文档摘要”)def generate_summary(doc_id: str) - str:# 调用LLM生成摘要pass五、知识质量评估与反馈飞轮Context Infra的最后一个关键环节是知识质量评估与反馈飞轮。没有持续优化的知识库最终会变成“垃圾进、垃圾出”的死循环。我们建立了三层反馈机制隐式反馈记录用户的追问、重新提问等行为作为检索质量的隐式信号。如果用户在同一主题下多次提问说明初次检索结果可能不够理想。显式反馈在问答界面提供“有帮助/无帮助”按钮收集用户对回答质量的直接评价。自动化评估定期运行RAGAS等评估框架对检索召回率、答案相关性等指标进行量化评测。pythonclass FeedbackCollector:“”“反馈收集与飞轮驱动”“”def __init__(self): self.feedback_log [] def record_feedback(self, query: str, doc_ids: List[str], answer: str, rating: int, user_action: str explicit): 记录反馈 rating: 1-5分 user_action: explicit(显式评分) / implicit(隐式行为) self.feedback_log.append({ query: query, doc_ids: doc_ids, answer: answer, rating: rating, user_action: user_action, timestamp: datetime.now().isoformat() }) def get_low_quality_queries(self, threshold: float 3.0) - List[Dict]: 识别低质量回答对应的查询 # 分析评分低于阈值的查询触发人工审核或自动优化 pass def trigger_retrieval_optimization(self): 基于反馈数据触发检索策略调整 # 例如调整混合检索的alpha权重、更新Embedding模型等 pass反馈数据驱动知识库的持续迭代低评分的问答对会被标记为“待优化”由人工审核或自动进行Query改写、文档补充等操作。这个闭环确保了知识库不是静态的“死文档”而是持续演进的活系统。六、总结与展望回顾整个Context Infra的建设历程我们有几个核心体会Context Engineering Prompt Engineering。 Prompt Engineering优化的是单次指令Context Engineering构建的是动态上下文系统。在Agent时代后者才是真正的核心竞争力。分层架构是工程化的基石。 数据层、检索层、执行层的清晰分离让每一层都可以独立优化和替换极大降低了系统的维护成本。从Naive RAG到Agentic RAG是必由之路。 单纯的向量检索LLM生成只能解决最基础的问答场景。要让AI真正“理解”业务、解决复杂问题必须引入Agent的工作流编排与自主决策能力。反馈飞轮决定系统的上限。 没有反馈优化的知识库注定会随着时间推移而退化。建立自动化的质量评估与反馈机制是让系统持续进化的关键。目前我们正在推进Agentic RAG的进一步落地——让Agent具备自适应检索、多步推理和工具调用的完整能力。同时多领域Agent的协同工作也在规划中未来将在项目管理、研发效能等多个场景中落地。Context Infra的建设是一场持久战但每一步的迭代都在让AI离“真正理解业务”更近一点。希望本文的分享能够为正在探索类似方向的团队提供一些参考。