adai/MemoryBear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[fix] Revising the judgment method for the interest analysis tags

Browse Source
This commit is contained in:
lanceyq
2026-03-03 23:30:54 +08:00
parent 66c153f1ad
commit 8466c8e019
5 changed files with 210 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
api/app/controllers/memory_agent_controller.py
View File

@@ -661,34 +661,38 @@ async def get_knowledge_type_stats_api(
return fail(BizCode.INTERNAL_ERROR, "获取知识库类型统计失败", str(e))
@router.get("/analytics/hot_memory_tags/by_user", response_model=ApiResponse)
async def get_hot_memory_tags_by_user_api(
end_user_id: Optional[str] = Query(None, description="用户ID（可选）"),
limit: int = Query(20, description="返回标签数量限制"),
@router.get("/analytics/interest_distribution/by_user", response_model=ApiResponse)
async def get_interest_distribution_by_user_api(
end_user_id: Optional[str] = Query(None, description="用户ID（必填）"),
limit: int = Query(5, le=5, description="返回兴趣标签数量限制，最多5个"),
language_type: str = Header(default=None, alias="X-Language-Type"),
current_user: User = Depends(get_current_user),
db: Session=Depends(get_db),
db: Session = Depends(get_db),
):
"""
获取指定用户的热门记忆标签
获取指定用户的兴趣分布标签
注意：标签语言由写入时的 X-Language-Type 决定，查询时不进行翻译
与热门标签不同，此接口专注于识别用户的兴趣活动（运动、爱好、学习、创作等），
过滤掉纯物品、工具、地点等不代表用户主动参与活动的名词。
返回格式：
[
{"name": "标签名", "frequency": 频次},
{"name": "兴趣活动名", "frequency": 频次},
...
]
"""
api_logger.info(f"Hot memory tags by user requested: end_user_id={end_user_id}")
language = get_language_from_header(language_type)
api_logger.info(f"Interest distribution by user requested: end_user_id={end_user_id}, language={language}")
try:
result = await memory_agent_service.get_hot_memory_tags_by_user(
result = await memory_agent_service.get_interest_distribution_by_user(
end_user_id=end_user_id,
limit=limit
limit=limit,
language=language
)
return success(data=result, msg="获取热门记忆标签成功")
return success(data=result, msg="获取兴趣分布标签成功")
except Exception as e:
api_logger.error(f"Hot memory tags by user failed: {str(e)}")
return fail(BizCode.INTERNAL_ERROR, "获取热门记忆标签失败", str(e))
api_logger.error(f"Interest distribution by user failed: {str(e)}")
return fail(BizCode.INTERNAL_ERROR, "获取兴趣分布标签失败", str(e))
@router.get("/analytics/user_profile", response_model=ApiResponse)

112
api/app/core/memory/analytics/hot_memory_tags.py
View File

@@ -16,6 +16,10 @@ class FilteredTags(BaseModel):
"""用于接收LLM筛选后的核心标签列表的模型。"""
meaningful_tags: List[str] = Field(..., description="从原始列表中筛选出的具有核心代表意义的名词列表。")
class InterestTags(BaseModel):
"""用于接收LLM筛选后的兴趣活动标签列表的模型。"""
interest_tags: List[str] = Field(..., description="从原始列表中筛选出的代表用户兴趣活动的标签列表。")
async def filter_tags_with_llm(tags: List[str], end_user_id: str) -> List[str]:
"""
使用LLM筛选标签列表，仅保留具有代表性的核心名词。
@@ -89,6 +93,70 @@ async def filter_tags_with_llm(tags: List[str], end_user_id: str) -> List[str]:
# 在LLM失败时返回原始标签，确保流程继续
return tags
async def filter_interests_with_llm(tags: List[str], end_user_id: str, language: str = "zh") -> List[str]:
"""
使用LLM从标签列表中筛选出代表用户兴趣活动的标签。
与 filter_tags_with_llm 不同，此函数专注于识别"活动/行为"类兴趣，
过滤掉纯物品、工具、地点等不代表用户主动参与活动的名词。
Args:
tags: 原始标签列表
end_user_id: 用户ID，用于获取LLM配置
Returns:
筛选后的兴趣活动标签列表
"""
try:
with get_db_context() as db:
from app.services.memory_agent_service import (
get_end_user_connected_config,
)
connected_config = get_end_user_connected_config(end_user_id, db)
config_id = connected_config.get("memory_config_id")
workspace_id = connected_config.get("workspace_id")
if not config_id and not workspace_id:
raise ValueError(
f"No memory_config_id found for end_user_id: {end_user_id}."
)
config_service = MemoryConfigService(db)
memory_config = config_service.load_memory_config(
config_id=config_id,
workspace_id=workspace_id
)
if not memory_config.llm_model_id:
raise ValueError(
f"No llm_model_id found in memory config {config_id}."
)
factory = MemoryClientFactory(db)
llm_client = factory.get_llm_client(memory_config.llm_model_id)
tag_list_str = ", ".join(tags)
from app.core.memory.utils.prompt.prompt_utils import render_interest_filter_prompt
rendered_prompt = render_interest_filter_prompt(tag_list_str, language=language)
messages = [
{
"role": "user",
"content": rendered_prompt
}
]
structured_response = await llm_client.response_structured(
messages=messages,
response_model=InterestTags
)
return structured_response.interest_tags
except Exception as e:
print(f"兴趣标签LLM筛选过程中发生错误: {e}")
return tags
async def get_raw_tags_from_db(
connector: Neo4jConnector,
end_user_id: str,
@@ -183,3 +251,47 @@ async def get_hot_memory_tags(end_user_id: str, limit: int = 10, by_user: bool =
finally:
# 确保关闭连接
await connector.close()
async def get_interest_distribution(end_user_id: str, limit: int = 10, by_user: bool = False, language: str = "zh") -> List[Tuple[str, int]]:
"""
获取用户的兴趣分布标签。
与 get_hot_memory_tags 不同，此函数使用专门针对"活动/行为"的LLM prompt，
过滤掉纯物品、工具、地点等，只保留能代表用户兴趣爱好的活动类标签。
Args:
end_user_id: 必需参数。如果by_user=False，则为end_user_id；如果by_user=True，则为user_id
limit: 最终返回的标签数量限制（默认10）
by_user: 是否按user_id查询（默认False，按end_user_id查询）
Raises:
ValueError: 如果end_user_id未提供或为空
"""
if not end_user_id or not end_user_id.strip():
raise ValueError(
"end_user_id is required. Please provide a valid end_user_id or user_id."
)
connector = Neo4jConnector()
try:
# 查询更多原始标签，给LLM提供充足上下文
query_limit = 40
raw_tags_with_freq = await get_raw_tags_from_db(connector, end_user_id, query_limit, by_user=by_user)
if not raw_tags_with_freq:
return []
raw_tag_names = [tag for tag, freq in raw_tags_with_freq]
# 使用兴趣活动专用prompt进行筛选
interest_tag_names = await filter_interests_with_llm(raw_tag_names, end_user_id, language=language)
# 保留原始频率，按兴趣筛选结果过滤
final_tags = [
(tag, freq)
for tag, freq in raw_tags_with_freq
if tag in interest_tag_names
]
return final_tags[:limit]
finally:
await connector.close()

17
api/app/core/memory/utils/prompt/prompt_utils.py
View File

@@ -548,3 +548,20 @@ async def render_ontology_extraction_prompt(
})
return rendered_prompt
def render_interest_filter_prompt(tag_list: str, language: str = "zh") -> str:
"""
Renders the interest filter prompt using the interest_filter.jinja2 template.
Args:
tag_list: Comma-separated string of raw tags to filter
language: Output language ("zh" for Chinese, "en" for English)
Returns:
Rendered prompt content as string
"""
template = prompt_env.get_template("interest_filter.jinja2")
rendered_prompt = template.render(tag_list=tag_list, language=language)
log_prompt_rendering('interest filter', rendered_prompt)
return rendered_prompt

47
api/app/core/memory/utils/prompt/prompts/interest_filter.jinja2 Normal file
View File

@@ -0,0 +1,47 @@
{% if language == "zh" %}
You are a user interest analysis expert. Your task is to identify activity-based tags from a tag list that represent the user's hobbies and interests. Please output the results in Chinese.
**Keep Rules** (keep if any condition is met):
- Tags representing sports or physical activities the user actively participates in (e.g., '攀岩', '篮球', '游泳', '跑步')
- Tags representing cultural or entertainment hobbies (e.g., '读书', '看电影', '听音乐', '摄影')
- Tags representing learning or creative activities (e.g., '编程', '绘画', '写作', '烹饪')
- Tags representing specific interest domains or hobby categories (e.g., '历史', '天文', '园艺')
**Filter Rules** (remove if any condition is met):
- Pure object or tool names that do not represent an activity (e.g., '篮球鞋', '相机', '书桌')
- Pure location or venue names (e.g., '篮球场', '图书馆', '健身房')
- Abstract concepts or quality descriptions (e.g., '核心力量', '团队合作', '专注力')
- Person names, brand names, or proper nouns (e.g., '乔丹', 'Nike')
**Merge Rules**: For semantically similar tags, keep only the most representative one.
For example: keep '篮球' over '打篮球'; keep '读书' over '阅读'.
**Example**:
Input: ['攀岩', '篮球场', '篮球鞋', '篮球', '《三体》', '历史', '核心力量', '烹饪', '菜刀']
Output: ['攀岩', '篮球', '历史', '烹饪']
Please filter the following tag list and return only the tags that represent user interest activities in Chinese: {{ tag_list }}
{% else %}
You are a user interest analysis expert. Your task is to identify activity-based tags from a tag list that represent the user's hobbies and interests. Please output the results in English.
**Keep Rules** (keep if any condition is met):
- Tags representing sports or physical activities the user actively participates in (e.g., 'rock climbing', 'basketball', 'swimming', 'running')
- Tags representing cultural or entertainment hobbies (e.g., 'reading', 'watching movies', 'listening to music', 'photography')
- Tags representing learning or creative activities (e.g., 'programming', 'painting', 'writing', 'cooking')
- Tags representing specific interest domains or hobby categories (e.g., 'history', 'astronomy', 'gardening')
**Filter Rules** (remove if any condition is met):
- Pure object or tool names that do not represent an activity (e.g., 'basketball shoes', 'camera', 'desk')
- Pure location or venue names (e.g., 'basketball court', 'library', 'gym')
- Abstract concepts or quality descriptions (e.g., 'core strength', 'teamwork', 'focus')
- Person names, brand names, or proper nouns (e.g., 'Jordan', 'Nike')
**Merge Rules**: For semantically similar tags, keep only the most representative one.
For example: keep 'basketball' over 'playing basketball'; keep 'reading' over 'reading books'.
**Example**:
Input: ['rock climbing', 'basketball court', 'basketball shoes', 'basketball', 'The Three-Body Problem', 'history', 'core strength', 'cooking', 'kitchen knife']
Output: ['rock climbing', 'basketball', 'history', 'cooking']
Please filter the following tag list and return only the tags that represent user interest activities in English: {{ tag_list }}
{% endif %}

32
api/app/services/memory_agent_service.py
View File

@@ -36,7 +36,7 @@ from app.core.memory.agent.utils.messages_tools import (
)
from app.core.memory.agent.utils.type_classifier import status_typle
from app.core.memory.agent.utils.write_tools import write # 新增：直接导入 write 函数
from app.core.memory.analytics.hot_memory_tags import get_hot_memory_tags
from app.core.memory.analytics.hot_memory_tags import get_hot_memory_tags, get_interest_distribution
from app.core.memory.utils.llm.llm_utils import MemoryClientFactory
from app.db import get_db_context
from app.models.knowledge_model import Knowledge, KnowledgeType
@@ -890,36 +890,36 @@ class MemoryAgentService:
return result
async def get_hot_memory_tags_by_user(
async def get_interest_distribution_by_user(
self,
end_user_id: Optional[str] = None,
limit: int = 20
limit: int = 5,
language: str = "zh"
) -> List[Dict[str, Any]]:
"""
获取指定用户的热门记忆标签
获取指定用户的兴趣分布标签。
与热门标签不同，此接口专注于识别用户的兴趣活动（运动、爱好、学习等），
过滤掉纯物品、工具、地点等不代表用户主动参与活动的名词。
参数：
- end_user_id: 用户ID（可选），对应Neo4j中的end_user_id字段
- end_user_id: 用户ID（必填）
- limit: 返回标签数量限制
- language: 输出语言（"zh" 中文, "en" 英文）
返回格式：
[
{"name": "标签名", "frequency": 频次},
{"name": "兴趣活动名", "frequency": 频次},
...
]
注意：标签语言由写入时的 X-Language-Type 决定，查询时不进行翻译
"""
try:
# by_user=False 表示按 end_user_id 查询（在Neo4j中，end_user_id就是用户维度）
tags = await get_hot_memory_tags(end_user_id, limit=limit, by_user=False)
payload = []
for tag, freq in tags:
payload.append({"name": tag, "frequency": freq})
return payload
tags = await get_interest_distribution(end_user_id, limit=limit, by_user=False, language=language)
return [{"name": tag, "frequency": freq} for tag, freq in tags]
except Exception as e:
logger.error(f"热门记忆标签查询失败: {e}")
raise Exception(f"热门记忆标签查询失败: {e}")
logger.error(f"兴趣分布标签查询失败: {e}")
raise Exception(f"兴趣分布标签查询失败: {e}")
async def get_user_profile(