1. 项目概述当大模型学会“认识”人最近在折腾一些AI应用发现一个挺有意思的痛点我们总想让大语言模型LLM记住我们是谁了解我们的背景、偏好和习惯从而提供更个性化的服务。但现实是每次对话都像初次见面你得反复自我介绍。jongomez/llmpeople这个项目就是为了解决这个问题而生的。简单来说它就是一个“人物信息管理器”专门设计用来让LLM应用能够持久化、结构化地存储和调用关于“人”的信息。想象一下你开发了一个智能助手它应该知道你的用户张三是个程序员喜欢用Python对机器学习感兴趣最近在做一个推荐系统项目。有了这些信息助手在回答技术问题、推荐学习资料甚至闲聊时都能更有针对性。llmpeople就是帮你把这些“人物画像”管起来的工具库。它不是一个独立的应用而是一个可以轻松集成到你的LLM应用比如基于LangChain、LlamaIndex或直接调用OpenAI/Anthropic API的项目中的组件。它的核心价值在于将原本可能散落在对话历史、临时变量或简陋配置文件中的用户信息进行标准化、向量化存储并在需要时高效检索从而让AI的“记忆力”变得可用、可控。这个项目适合任何正在构建需要长期记忆或用户画像功能的LLM应用开发者。无论你是在做客服机器人、个性化内容推荐、智能导师还是企业内部的知识助手只要涉及到“理解并记住用户”llmpeople提供的思路和工具都值得参考。接下来我会结合自己的集成经验拆解它的设计思路、核心用法并分享一些实战中踩过的坑和优化技巧。2. 核心设计思路与架构拆解2.1 为什么需要专门的人物信息管理在深入代码之前我们先想想为什么不能直接用数据库存个JSON了事。LLM应用中对“人”的信息使用有几个特殊需求动态与结构化人的信息不是一成不变的。兴趣会变项目会更新每次对话都可能产生新的信息片段例如用户这次提到“我最近开始学吉他”。这些信息需要能被动态添加、更新并以一种LLM容易理解的结构比如属性-值对、标签、自由文本描述来组织。语义化检索当AI需要回忆关于用户的某个方面时它不太可能用精确的关键词去数据库查询。更常见的场景是根据当前对话的上下文一段自然语言去“联想”出相关的用户信息。例如用户问“关于我之前那个项目有什么建议”系统需要从存储的信息中找到最相关的“项目”描述。这需要向量检索能力。信息融合与摘要来自不同时间、不同对话的信息可能是冗余或互补的。系统需要能自动合并相似信息比如两次都提到“喜欢Python”或者将零散信息综合成一段连贯的人物描述再喂给LLM。这涉及到信息去重、优先级排序和文本摘要。隐私与粒度控制不是所有信息都应该在每次对话中被无条件唤起。有些是公开偏好有些是敏感数据。系统需要能根据上下文或规则决定哪些信息可以暴露给LLM。这要求存储层有基本的分类或标记能力。llmpeople的设计正是围绕这些需求展开的。它没有重新发明所有轮子而是巧妙地利用了现有的工具链如Pydantic用于数据验证、SQLite/PostgreSQL用于持久化、向量数据库用于语义检索提供了一个轻量级的抽象层。2.2 项目架构与核心模块浏览其代码库可以看到几个核心部分Person数据模型这是核心。通常用一个Pydantic模型来定义包含一些基本字段如id、name以及一个关键字段——attributes或profile。这个字段可能是一个字典或是一个列表用于存放动态的属性如{occupation: software engineer, hobbies: [hiking, reading]}。好的设计会为这些属性定义子模型以支持类型验证和更复杂的结构。存储后端项目会抽象一个存储接口然后提供多种实现。内存存储用于开发和测试数据在运行时保存在内存中。SQL数据库存储使用SQLAlchemy或类似ORM将Person模型和其属性序列化后存入SQLite或PostgreSQL。这是最常用的持久化方案。向量存储集成这是实现语义检索的关键。项目可能会将人物的文本描述如将所有属性拼接成一段文字或关键的属性值通过嵌入模型embedding model转换成向量然后存入像Chroma、Weaviate、Pinecone或Qdrant这样的向量数据库中。当需要检索时将当前查询也转换成向量进行相似度搜索。管理器/服务类提供高级API如add_person、update_person、find_similar_people、get_relevant_context等。它封装了底层存储逻辑并可能包含一些业务逻辑比如信息融合的策略当新增信息与旧信息冲突时如何处理。与LLM框架的集成提供与LangChain或LlamaIndex工具链便捷集成的组件。例如一个PeopleRetriever可以作为一个Retriever对象被加入到LangChain的检索链中或者一个PeopleMemory类可以作为对话记忆模块的一部分。注意llmpeople的具体实现可能随时间变化但其核心思想是通用的。下面的实操将基于这个通用思想并假设一个典型的实现方式为你展示如何从零开始构建或集成这样一个系统。3. 从零到一构建你的第一个“人物记忆体”3.1 环境准备与基础模型定义我们首先创建一个干净的Python环境并安装核心依赖。这里假设我们使用SQLite进行基础存储并使用chromadb作为向量存储同时用openai的嵌入模型。# 创建并激活虚拟环境可选但推荐 python -m venv venv_llmpeople source venv_llmpeople/bin/activate # Linux/macOS # venv_llmpeople\Scripts\activate # Windows # 安装依赖 pip install pydantic sqlalchemy chromadb openai tiktoken接下来定义我们的Person模型。我们将使用Pydantic V2因为它性能更好功能更强大。from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any from uuid import uuid4, UUID from datetime import datetime class PersonAttribute(BaseModel): 定义一个属性条目包含键、值、来源和置信度 key: str Field(..., description属性名称如 occupation, favorite_food) value: Any Field(..., description属性值可以是字符串、数字、列表等) source: str Field(defaultconversation, description信息来源如 conversation_20240415, user_profile) confidence: float Field(default1.0, ge0.0, le1.0, description信息置信度由提取模型或规则给出) created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) class Config: json_encoders { datetime: lambda v: v.isoformat() } class Person(BaseModel): 核心人物模型 id: UUID Field(default_factoryuuid4, description唯一标识符) name: str Field(..., description人物名称) # 使用一个列表来存储多个属性条目允许同一属性有多个来源或历史值 attributes: List[PersonAttribute] Field(default_factorylist) # 一个综合的文本描述可用于生成向量嵌入 summary: Optional[str] Field(None, description自动或手动生成的人物摘要) created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) def get_attribute(self, key: str) - List[Any]: 获取某个键的所有值按时间或置信度排序 values [attr.value for attr in self.attributes if attr.key key] return values def update_attribute(self, key: str, value: Any, source: str, confidence: float 1.0): 更新或添加一个属性 # 简单的实现直接追加新条目。更复杂的实现可以合并、去重。 new_attr PersonAttribute(keykey, valuevalue, sourcesource, confidenceconfidence) self.attributes.append(new_attr) self.updated_at datetime.now() def to_text(self, for_embedding: bool False) - str: 将人物信息转换为文本用于LLM上下文或生成嵌入向量 lines [fName: {self.name}] # 按属性键分组展示最新或最高置信度的值 attr_dict {} for attr in sorted(self.attributes, keylambda x: (x.confidence, x.updated_at), reverseTrue): if attr.key not in attr_dict: # 只取每个键最高优先级的一个 attr_dict[attr.key] attr.value for key, value in attr_dict.items(): if isinstance(value, list): value_str , .join(str(v) for v in value) else: value_str str(value) lines.append(f{key}: {value_str}) if self.summary: lines.append(fSummary: {self.summary}) text \n.join(lines) if for_embedding: # 为了嵌入模型可能需要进行一些清洗或截断 text text.replace(\n, ) return text这个模型设计有几个小心思属性列表化attributes是一个列表而不是字典。这允许我们记录同一个属性的多个值可能来自不同对话并附带来源和置信度。这在处理冲突或不确定信息时非常有用。分离存储与视图原始数据完整存储而to_text方法负责生成用于LLM或向量化的字符串。我们可以根据需要改变生成策略而不影响底层数据。使用UUIDid使用UUID便于分布式系统或未来数据迁移。3.2 实现存储层SQL 向量数据库双写单纯用SQL存储文本是可以的但无法实现语义检索。因此我们需要一个混合存储策略用SQL数据库存储完整的、结构化的Person数据同时用向量数据库存储其文本表示的嵌入向量用于快速相似度查找。第一步SQL存储实现使用SQLAlchemyfrom sqlalchemy import create_engine, Column, String, JSON, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import json from datetime import datetime Base declarative_base() class PersonModel(Base): __tablename__ people id Column(String(36), primary_keyTrue) # UUID as string name Column(String(255), nullableFalse) attributes Column(JSON, defaultlist) # 存储为JSON数组 summary Column(Text, nullableTrue) created_at Column(DateTime, defaultdatetime.utcnow) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) class SQLPersonStorage: def __init__(self, db_url: str sqlite:///people.db): self.engine create_engine(db_url) Base.metadata.create_all(self.engine) self.SessionLocal sessionmaker(bindself.engine) def save(self, person: Person): session self.SessionLocal() try: # 检查是否存在 existing session.query(PersonModel).filter_by(idstr(person.id)).first() person_dict person.dict() # 将UUID和datetime转换为字符串以便JSON序列化 person_dict[id] str(person_dict[id]) person_dict[created_at] person_dict[created_at].isoformat() person_dict[updated_at] person_dict[updated_at].isoformat() for attr in person_dict[attributes]: attr[created_at] attr[created_at].isoformat() attr[updated_at] attr[updated_at].isoformat() if existing: for key, value in person_dict.items(): setattr(existing, key, value) else: new_person PersonModel(**person_dict) session.add(new_person) session.commit() except Exception as e: session.rollback() raise e finally: session.close() def load(self, person_id: UUID) - Optional[Person]: session self.SessionLocal() try: model session.query(PersonModel).filter_by(idstr(person_id)).first() if model: # 将字符串转换回datetime对象 data {c.name: getattr(model, c.name) for c in model.__table__.columns} return Person.parse_obj(data) return None finally: session.close()第二步向量存储集成使用ChromaDBimport chromadb from chromadb.config import Settings from openai import OpenAI import hashlib class VectorPersonStorage: def __init__(self, openai_api_key: str, collection_name: str people_profiles): self.client chromadb.Client(Settings(persist_directory./chroma_db, anonymized_telemetryFalse)) self.collection self.client.get_or_create_collection(namecollection_name) self.openai_client OpenAI(api_keyopenai_api_key) self.embedding_model text-embedding-3-small # 可根据需要更换 def _get_embedding(self, text: str) - List[float]: 调用OpenAI API获取文本嵌入向量 response self.openai_client.embeddings.create( modelself.embedding_model, inputtext ) return response.data[0].embedding def index_person(self, person: Person): 将一个人物索引到向量数据库 text_for_embedding person.to_text(for_embeddingTrue) embedding self._get_embedding(text_for_embedding) # 使用人物ID和文本的哈希作为唯一ID避免重复 doc_id f{person.id}_{hashlib.md5(text_for_embedding.encode()).hexdigest()[:8]} metadata { person_id: str(person.id), name: person.name, source: llmpeople_index } self.collection.add( embeddings[embedding], documents[text_for_embedding], metadatas[metadata], ids[doc_id] ) def find_similar(self, query_text: str, n_results: int 3) - List[Dict]: 根据查询文本查找相似的人物 query_embedding self._get_embedding(query_text) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # 结果格式转换 people_info [] if results[ids]: for i in range(len(results[ids][0])): people_info.append({ person_id: results[metadatas][0][i][person_id], document: results[documents][0][i], distance: results[distances][0][i] }) return people_info第三步创建统一的管理器现在我们把两者结合起来创建一个PeopleManager它负责协调SQL存储和向量存储。class PeopleManager: def __init__(self, sql_storage: SQLPersonStorage, vector_storage: VectorPersonStorage): self.sql_store sql_storage self.vector_store vector_storage def add_or_update_person(self, person: Person, reindex: bool True): 添加或更新人物并可选地重新索引到向量库 self.sql_store.save(person) if reindex: # 注意这里简单删除旧索引并新增。生产环境可能需要更复杂的增量更新逻辑。 # 我们可以先删除该person_id的所有旧向量再添加新的。 self.vector_store.index_person(person) def get_person(self, person_id: UUID) - Optional[Person]: return self.sql_store.load(person_id) def search_by_context(self, context_query: str, threshold: float 0.3) - List[Person]: 根据一段上下文查询相关人物 similar_docs self.vector_store.find_similar(context_query) persons [] for doc_info in similar_docs: # 可以根据距离阈值过滤 if doc_info[distance] threshold: # Chroma默认使用余弦距离越小越相似 person_id UUID(doc_info[person_id]) person self.get_person(person_id) if person: persons.append(person) return persons def extract_and_update_from_conversation(self, person_id: UUID, conversation_text: str, llm_client): 一个高级功能使用LLM从对话文本中提取人物属性并更新 # 这里简化为一个示例提示词。实际应用中你需要设计更精细的提示和解析逻辑。 prompt f 从以下对话中提取关于人物“{self.get_person(person_id).name if self.get_person(person_id) else 该用户}的新信息或属性。 对话内容 {conversation_text} 请以JSON格式输出包含一个“attributes”列表每个元素是一个对象包含“key”属性名和“value”属性值。 只提取确实在对话中明确提到或强烈暗示的信息。 # 调用LLM这里用模拟响应 # response llm_client.chat.completions.create(...) # extracted_data json.loads(response.choices[0].message.content) # 模拟提取结果 extracted_data { attributes: [ {key: current_project, value: 重构用户画像系统}, {key: mood, value: 感到有些挑战但兴奋} ] } person self.get_person(person_id) if person: for attr in extracted_data[attributes]: person.update_attribute( keyattr[key], valueattr[value], sourcefextracted_from_conversation_{datetime.now().date()}, confidence0.8 # 给一个置信度 ) self.add_or_update_person(person)这个管理器提供了基本的核心功能持久化存储、语义检索甚至初步的信息提取。你可以看到通过组合SQL的精确查询和向量的语义搜索我们构建了一个既能精确管理又能模糊联想的人物信息库。4. 实战集成将人物记忆注入你的LLM应用4.1 与LangChain集成假设我们正在构建一个基于LangChain的聊天机器人。我们希望机器人在每次回复时都能参考当前用户的历史信息。首先我们基于上面的PeopleManager创建一个LangChain的Retriever。from langchain.schema import BaseRetriever, Document from typing import List class PeopleRetriever(BaseRetriever): LangChain检索器用于从人物库中查找相关人物信息 def __init__(self, people_manager: PeopleManager, user_id: UUID): self.manager people_manager self.user_id user_id # 当前对话的用户ID def get_relevant_documents(self, query: str) - List[Document]: # 1. 首先获取当前用户的最新信息作为首要上下文 current_user self.manager.get_person(self.user_id) primary_doc None if current_user: primary_doc Document( page_contentcurrent_user.to_text(), metadata{source: current_user_profile, person_id: str(self.user_id)} ) # 2. 根据查询语义查找其他相关人物例如用户提到了“我的同事小明” similar_people self.manager.search_by_context(query, n_results2) other_docs [] for person in similar_people: if person.id ! self.user_id: # 排除自己 other_docs.append(Document( page_contentperson.to_text(), metadata{source: similar_person, person_id: str(person.id), name: person.name} )) # 组合文档当前用户信息放在最前面 all_docs [] if primary_doc: all_docs.append(primary_doc) all_docs.extend(other_docs) return all_docs async def aget_relevant_documents(self, query: str) - List[Document]: # 异步实现略 return self.get_relevant_documents(query)然后在你的对话链中使用这个检索器。例如构建一个RetrievalQA链from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 或 ChatOpenAI from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma # 这里仅作示例我们用自己的Retriever # 初始化组件 llm OpenAI(temperature0, openai_api_keyyour_key) people_manager PeopleManager(...) # 初始化你的管理器 user_id UUID(...) # 从会话中获取当前用户ID # 创建自定义检索器 retriever PeopleRetriever(people_managerpeople_manager, user_iduser_id) # 构建QA链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 简单地将所有检索到的文档塞入上下文 retrieverretriever, return_source_documentsTrue, chain_type_kwargs{ prompt: YOUR_CUSTOM_PROMPT # 建议使用自定义提示词明确告诉LLM如何使用人物信息 } ) # 使用链进行问答 response qa_chain.run(我之前提到的那个项目现在进展如何) print(response[result]) # LLM的上下文现在包含了当前用户的profile其中记录了项目信息因此它能做出个性化回答。自定义提示词示例你是一个智能助手并且你了解以下关于对话涉及人员的信息 {context} 请根据以上信息回答用户的问题。如果信息不足可以询问更多细节。 用户问题{question} 助手回答4.2 与LlamaIndex集成LlamaIndex对结构化数据有很好的支持。我们可以将Person对象视为一种“文档”并利用LlamaIndex的索引和查询引擎。from llama_index.core import Document as LlamaDocument from llama_index.core import VectorStoreIndex, StorageContext from llama_index.vector_stores.chroma import ChromaVectorStore from llama_index.core.schema import TextNode # 假设我们有一个PeopleManager实例 manager PeopleManager(...) # 1. 将人物库中的所有人物转换为LlamaIndex的Document llama_documents [] all_person_ids [...] # 你需要一个方法从SQL存储中获取所有ID for pid in all_person_ids: person manager.get_person(pid) if person: text person.to_text() # 创建Document并将人物ID存入元数据 doc LlamaDocument( texttext, metadata{person_id: str(person.id), name: person.name, type: person_profile}, excluded_llm_metadata_keys[person_id], # 不让LLM看到ID excluded_embed_metadata_keys[person_id] ) llama_documents.append(doc) # 2. 构建索引这里复用之前的Chroma客户端 chroma_client manager.vector_store.client chroma_collection chroma_client.get_or_create_collection(llamaindex_people) vector_store ChromaVectorStore(chroma_collectionchroma_collection) storage_context StorageContext.from_defaults(vector_storevector_store) index VectorStoreIndex.from_documents( llama_documents, storage_contextstorage_context, show_progressTrue ) # 3. 创建查询引擎 query_engine index.as_query_engine( similarity_top_k3, # 可以添加自定义的响应合成器来优化输出格式 ) # 4. 查询时可以附加过滤器例如只查询特定用户 from llama_index.core.vector_stores import MetadataFilter, MetadataFilters def query_with_user_context(question: str, current_user_id: UUID): # 构建过滤器优先获取当前用户的信息同时允许检索其他相关信息 filters MetadataFilters(filters[ MetadataFilter(keyperson_id, valuestr(current_user_id)) ]) # 注意LlamaIndex的查询引擎可能不支持在查询时动态改变过滤器。 # 更常见的做法是在检索后对结果进行后处理或者为每个用户创建独立的索引/检索器。 # 这里展示一个后处理的思路 response query_engine.query(question) # 对response.source_nodes进行过滤和排序优先展示当前用户的信息 # ... return response与LangChain相比LlamaIndex更专注于索引和检索本身给了你更多的控制权。你可以精细地调整索引结构、检索策略和响应合成过程。5. 进阶技巧与避坑指南在实际使用llmpeople这类模式时我积累了一些经验教训这里分享给你。5.1 信息冲突与融合策略当同一个属性从不同来源例如两次对话获得不同值时如何处理我们简单的update_attribute只是追加这可能导致信息混乱。解决方案实现一个属性融合器。可以基于置信度、时间戳和来源可靠性进行决策。def merge_attributes(self, key: str): 合并同一个key下的所有属性条目 relevant_attrs [attr for attr in self.attributes if attr.key key] if not relevant_attrs: return None # 策略1选择置信度最高的 best_attr max(relevant_attrs, keylambda x: x.confidence) # 策略2如果置信度接近选择最新的 # 策略3对于列表类型的值如hobbies可以合并去重 # 这里采用策略1并删除其他条目 self.attributes [attr for attr in self.attributes if attr.key ! key] self.attributes.append(best_attr) # 放回最好的那个 return best_attr.value更复杂的策略可以引入投票机制或者使用一个小型LLM来判断哪个值更可能正确。5.2 向量检索的优化该索引什么不是所有人物信息都适合做向量检索。将整个to_text()结果拿去索引可能包含大量低频或无关词汇稀释了重要信息的语义。建议索引摘要而非全文为每个人物维护一个由LLM生成的、浓缩核心特征的summary字段专门用于向量化。多向量索引为不同类别的属性创建不同的向量索引。例如一个索引针对“专业技能”另一个针对“个人兴趣”。查询时根据问题分类选择对应的索引或者进行多路检索后再融合。关键词过滤前置在向量检索前先用简单的关键词匹配过滤出可能相关的人物子集再进行昂贵的向量相似度计算这能大幅提升性能。5.3 性能与扩展性考量批量操作当需要初始化或更新大量人物数据时确保你的index_person方法支持批量嵌入生成和批量插入向量数据库这比单条处理快几个数量级。缓存对于高频访问的当前用户信息可以缓存在内存中并设置过期时间避免每次对话都查询数据库。数据库连接池确保你的SQL存储使用了连接池如SQLAlchemy的QueuePool以应对Web服务的高并发请求。向量数据库的选择Chroma轻量适合原型和中小规模。生产环境若数据量大10万条应考虑Qdrant、Weaviate或Pinecone它们支持分布式、更丰富的过滤条件和更好的性能。5.4 隐私与安全这是重中之重。人物信息可能包含敏感数据。字段级加密对高度敏感的属性如邮箱、电话在存入数据库前进行加密。访问日志记录谁哪个系统/管理员在什么时候访问或修改了哪些人物信息。LLM上下文隔离在将人物信息注入LLM提示词时务必进行二次检查。可以设计一个“信息暴露过滤器”根据对话场景决定哪些属性可以发送给LLM。例如在客服场景中可以暴露“产品偏好”但绝不能暴露“账户余额”。用户知情与控制如果面向C端用户提供透明性让用户能看到并管理AI所记住的关于他们的信息并提供“忘记我”的选项。5.5 测试策略测试这类系统不能只测Happy Path。属性提取准确性测试准备一系列包含用户信息的模拟对话测试你的信息提取逻辑如果实现了的话能否正确抓取和归类。检索相关性测试构造各种查询人工评估返回的人物信息是否相关。可以计算一个简单的召回率RecallK。冲突处理测试故意输入矛盾信息检查系统融合后的结果是否符合预期策略。性能压测模拟大量并发用户更新和查询观察数据库和向量服务的响应时间。6. 总结与展望构建一个健壮的人物信息管理系统远不止是定义几个模型和调用API。它涉及数据建模、存储设计、检索算法、LLM集成、隐私安全和性能优化等多个层面。jongomez/llmpeople项目提供了一个很好的起点和设计模式的参考。从我自己的实践来看最大的挑战往往不在技术实现而在产品逻辑上到底应该记住什么记忆的粒度多细信息过期了怎么办如何平衡个性化与隐私这些问题没有标准答案需要根据你的具体应用场景反复权衡。一个实用的建议是从最小可行产品MVP开始。初期可以只实现核心的存储和基于关键词的检索快速验证用户对“记忆”功能的需求和反馈。然后再逐步引入向量检索、自动信息提取、冲突解决等高级功能。这样既能控制复杂度也能确保每一步都走在解决真实用户痛点的方向上。最后这个领域发展很快新的向量数据库、更高效的嵌入模型、以及LLM本身对长上下文处理能力的提升都会影响这类系统的设计。保持关注并准备好随时重构你的“记忆”模块。毕竟让AI真正理解并记住我们这场旅程才刚刚开始。