14天构建AI数字分身：基于RAG与Agent的实践指南

1. 项目概述：打造你的数字分身

最近在AI领域，基于个人数据的数字分身（Digital Twin）技术越来越受关注。想象一下，如果有一个AI能像你一样思考、说话，甚至记得你过去的所有经历，那会是什么体验？这就是我们今天要探讨的主题——通过你的聊天记录和日记，构建一个真正懂你的AI数字分身。

这个项目不同于普通的聊天机器人，它需要三个核心能力：

1. 深度理解你的个人语言风格和表达习惯
2. 准确回忆你过去的经历和观点
3. 在对话中展现出连贯的"人格"特征

我最近用14天时间完成了一个完整的实现方案，效果相当惊艳。当测试时问它"我上周三晚上做了什么"，它能准确从日记中找出相关内容并组织成自然回答；问"我对AI行业的看法是什么"，它能综合多篇日记的观点给出符合我个人风格的总结。

2. 工业级RAG链路重构（第1-5天）

2.1 语义切片与文档解析

传统RAG系统最大的问题在于粗暴的文本分割。假设你的日记写着："今天和Mary吃了火锅。聊到她的新工作，我觉得...（500字）...明天要去买新手机"，普通token splitter可能会在"我觉得"后面强行切断，导致语义断裂。

解决方案是语义切片(Semantic Chunking)。我测试了三种方案：

1. LangChain的RecursiveCharacterTextSplitter：简单但不够智能
2. LlamaIndex的SentenceSplitter：对中文支持更好
3. 自定义规则：结合标点、段落和语义连贯性

最终采用的方案：

from llama_index.core.node_parser import SentenceSplitter parser = SentenceSplitter( chunk_size=512, chunk_overlap=20, paragraph_separator="\n\n", secondary_chunking_regex="[^。？！]+[。？！]?" ) nodes = parser.get_nodes_from_documents(documents)

关键发现：对于中文日记，设置paragraph_separator="\n\n"特别重要，因为大多数人写日记会有自然分段习惯。

2.2 向量化与本地数据库

向量模型的选择直接影响检索质量。对比测试了以下模型：

最终选择bge-base-zh模型，虽然稍大但质量明显更好。存储使用ChromaDB的本地模式：

import chromadb from sentence_transformers import SentenceTransformer embedder = SentenceTransformer('BAAI/bge-base-zh') client = chromadb.PersistentClient(path="./vector_db") collection = client.create_collection("diary") # 批量插入优化 batch_size = 100 for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] embeddings = embedder.encode(batch) collection.add( ids=[str(j) for j in range(i, i+len(batch))], documents=batch, embeddings=embeddings.tolist() )

避坑指南：批量插入时控制batch_size在100左右，太大可能导致内存溢出，太小则速度慢。

2.3 混合检索策略

纯向量检索在特定名词查询上表现不佳。比如问"3月15日的日记"，关键词"3月15日"比语义更重要。实现混合检索：

from rank_bm25 import BM25Okapi import jieba # BM25初始化 tokenized_corpus = [list(jieba.cut(doc)) for doc in corpus] bm25 = BM25Okapi(tokenized_corpus) def hybrid_search(query, top_k=10): # 向量检索 vector_results = collection.query( query_embeddings=embedder.encode([query]).tolist(), n_results=top_k*2 ) # BM25检索 tokenized_query = list(jieba.cut(query)) bm25_scores = bm25.get_scores(tokenized_query) bm25_indices = np.argsort(bm25_scores)[-top_k*2:][::-1] # 结果融合 combined = [] seen_ids = set() # 优先取BM25结果 for idx in bm25_indices: if idx not in seen_ids: combined.append((corpus[idx], bm25_scores[idx], "bm25")) seen_ids.add(idx) # 补充向量结果 for doc, score in zip(vector_results['documents'], vector_results['distances']): if doc not in seen_ids: combined.append((doc, 1-score, "vector")) return sorted(combined, key=lambda x: x[1], reverse=True)[:top_k]

2.4 重排序优化

测试发现，直接给LLM喂20条检索结果会导致回答质量下降。解决方案是使用bge-reranker-base进行精排：

from transformers import AutoModelForSequenceClassification, AutoTokenizer reranker = AutoModelForSequenceClassification.from_pretrained('BAAI/bge-reranker-base') tokenizer = AutoTokenizer.from_pretrained('BAAI/bge-reranker-base') def rerank(query, passages): pairs = [[query, passage] for passage in passages] inputs = tokenizer(pairs, padding=True, truncation=True, return_tensors='pt', max_length=512) with torch.no_grad(): scores = reranker(**inputs).logits.view(-1).float() return [passages[i] for i in scores.argsort(descending=True)[:3]]

性能优化：在4060上使用半精度(fp16)推理，速度提升40%：

reranker = reranker.half().cuda()

2.5 全链路封装

最终封装为可复用的Pipeline类：

class AdvancedRAGPipeline: def __init__(self, embed_model, db_path): self.embedder = embed_model self.client = chromadb.PersistentClient(path=db_path) self.collection = self.client.get_collection("diary") self.reranker = AutoModelForSequenceClassification.from_pretrained( 'BAAI/bge-reranker-base').half().cuda() def search(self, query, top_k=3): # 混合检索获取候选 candidates = self.hybrid_search(query, top_k*5) # 重排序 reranked = self.rerank(query, [doc for doc,_,_ in candidates]) return reranked def __call__(self, query): start_time = time.time() results = self.search(query) print(f"检索耗时：{time.time()-start_time:.2f}s") return self.format_prompt(query, results)

3. Agent工程与记忆注入（第6-9天）

3.1 ReAct框架实现

让LLM学会自主调用工具是关键突破点。使用Qwen-7B-Chat模型，设计如下system prompt：

你是一个数字分身助手，需要根据用户问题决定是否查询日记库。 当需要查询时，请严格按以下JSON格式输出： { "action": "search_diary", "query": "要搜索的关键词或问题" } 你的记忆分为三种： 1. 短期记忆：记住当前对话内容 2. 摘要记忆：对长时间对话进行总结 3. 日记记忆：通过search_diary动作查询 """ 示例对话： 用户：我上周三做了什么？ 助手：{ "action": "search_diary", "query": "上周三的活动" } [等待RAG返回结果...] 根据你的日记，上周三你参加了AI研讨会并在晚上与朋友聚餐。

实现代码：

def chat_loop(): memory = ConversationBufferMemory() while True: user_input = input("你：") # 记忆处理 memory.save_context({"input": user_input}, {"output": ""}) # 获取LLM响应 response = qwen.generate( prompt_template.format(history=memory.load_memory_variables(), query=user_input) ) try: # 尝试解析JSON动作 action = json.loads(response) if action["action"] == "search_diary": results = rag_pipeline(action["query"]) memory.save_context({"observation": results}, {}) continue except: pass print(f"助手：{response}") memory.save_context({"output": response}, {})

3.2 三重记忆架构

完整的记忆系统实现：

class MemorySystem: def __init__(self, rag_pipeline): self.buffer = ConversationBufferWindowMemory(k=5) self.summary = ConversationSummaryMemory(llm=qwen) self.vector = rag_pipeline def update(self, user_input, ai_response): self.buffer.save_context({"input": user_input}, {"output": ai_response}) if len(self.buffer.load_memory_variables()["history"]) > 1000: summary = self.summary.load_memory_variables({"summary": True}) self.summary.save_context({"input": "对话摘要"}, {"output": summary}) self.buffer.clear() def query(self, question): # 先从buffer查近期记忆 buffer_mem = self.buffer.load_memory_variables() if question in buffer_mem["history"]: return buffer_mem["history"] # 再从摘要查 summary_mem = self.summary.load_memory_variables() if question in summary_mem["history"]: return summary_mem["history"] # 最后查日记 return self.vector(question)

3.3 LangGraph工作流

实现循环决策的工作流：

from langgraph.graph import Graph workflow = Graph() # 定义节点 def should_search(state): llm_output = qwen.generate(f"是否需要查询日记来回答：{state['question']}") return "yes" in llm_output.lower() def search_action(state): query = qwen.generate(f"根据问题'{state['question']}'生成搜索查询") return {"results": rag_pipeline(query)} def respond(state): if "results" in state: return qwen.generate(f"根据以下信息回答问题：{state['question']}\n信息：{state['results']}") return qwen.generate(state["question"]) # 构建图 workflow.add_node("check_search", should_search) workflow.add_node("perform_search", search_action) workflow.add_node("generate_response", respond) # 定义边 workflow.add_conditional_edges( "check_search", lambda x: "search" if x["needs_search"] else "respond", {"search": "perform_search", "respond": "generate_response"} ) workflow.add_edge("perform_search", "generate_response") # 编译 app = workflow.compile()

4. 多模态扩展（第10-11天）

4.1 视觉模型部署

在4060上部署Qwen-VL-Chat的4bit量化版：

from transformers import AutoModelForCausalLM, AutoTokenizer model = AutoModelForCausalLM.from_pretrained( "Qwen/Qwen-VL-Chat", device_map="auto", quantization_config=bnb_4bit_config ) tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-VL-Chat") def describe_image(image_path): query = tokenizer.from_list_format([ {'image': image_path}, {'text': '详细描述这张图片的内容'} ]) outputs = model.generate(query) return tokenizer.decode(outputs[0])

4.2 多模态RAG实现

使用CLIP模型构建图文联合检索：

import clip import torch device = "cuda" if torch.cuda.is_available() else "cpu" clip_model, preprocess = clip.load("ViT-B/32", device=device) def encode_image(image_path): image = preprocess(Image.open(image_path)).unsqueeze(0).to(device) with torch.no_grad(): return clip_model.encode_image(image).cpu().numpy() def add_to_mm_db(image_path, description): image_embed = encode_image(image_path) text_embed = clip_model.encode_text(clip.tokenize(description).to(device)).cpu().numpy() # 存入ChromaDB mm_collection.add( embeddings=[(image_embed + text_embed).tolist()], # 合并特征 documents=[description], metadatas=[{"type": "image", "path": image_path}] ) def mm_search(query, top_k=3): text_input = clip.tokenize(query).to(device) with torch.no_grad(): query_embed = clip_model.encode_text(text_input).cpu().numpy() results = mm_collection.query( query_embeddings=[query_embed.tolist()], n_results=top_k ) return results

5. 系统封装与优化（第12-14天）

5.1 FastAPI服务封装

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): message: str user_id: str @app.post("/chat") async def chat_endpoint(request: ChatRequest): # 加载用户专属的agent agent = load_user_agent(request.user_id) # 处理消息 response = agent.process(request.message) return { "response": response, "status": "success" } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

5.2 边界情况处理

实现了一套错误处理机制：

1. JSON解析失败：用正则兜底

import re def extract_json(text): # 尝试找json块 match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group()) except: pass return None

2. RAG无结果：设计fallback策略

def safe_search(query): results = rag_pipeline(query) if not results: return "我没有找到相关记忆" # 验证结果相关性 relevance = qwen.generate( f"判断以下内容是否与问题相关：\n问题：{query}\n内容：{results[:500]}\n" "只回答'相关'或'不相关'" ) return results if "相关" in relevance else "我的记忆中没有相关信息"

5.3 性能优化技巧

在4060上实现的关键优化：

1. 模型量化：

from transformers import BitsAndBytesConfig bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_use_double_quant=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.float16 )

2. 显存管理：

from accelerate import infer_auto_device_map device_map = infer_auto_device_model( model, max_memory={0: "10GiB", "cpu": "20GiB"}, no_split_module_classes=["LlamaDecoderLayer"] )

3. 批处理优化：

def batch_inference(texts, model, batch_size=8): outputs = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] inputs = tokenizer(batch, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): out = model(**inputs.to(model.device)) outputs.extend(out) return outputs

6. 项目总结与心得

经过这14天的密集开发，我总结出几个关键经验：

1. 数据质量决定上限：发现日记中如果有大量碎片化记录（如"今天很累"），会显著降低回答质量。后来增加了预处理步骤，过滤掉少于50字的条目。

2. 混合检索的黄金比例：经过反复测试，BM25和向量检索结果按6:4比例融合效果最佳，既能抓住关键词，又不失语义连贯性。

3. 记忆系统的温度参数：短期记忆(buffer)的温度设为0.7保持创造性，而日记记忆的温度设为0.3确保准确性。

4. 显存不足的创意解法：当需要同时加载多个模型时，采用"接力"方式——使用完后立即清空显存，再加载下一个模型。

这个项目最让我惊喜的是，当系统第一次准确回忆起我三个月前某次旅行的细节时，那种"它真的懂我"的感觉。现在这个数字分身已经能处理我70%的日常查询，比如"我去年读过哪些AI论文"、"我和Alice上次见面聊了什么"等。