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

Merge branch 'refs/heads/develop' into feature/agent-tool_xjn

Browse Source
This commit is contained in:
Timebomb2018
2026-04-24 19:41:23 +08:00
parent 141fd94513 a9413f57d1
commit 15b352d16b
52 changed files with 1752 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
api/app/services/api_key_service.py
View File

@@ -65,6 +65,11 @@ class ApiKeyService:
BizCode.BAD_REQUEST
)
if data.resource_id:
app = db.get(App, data.resource_id)
if not app or not app.current_release_id:
raise BusinessException("该应用未发布", BizCode.APP_NOT_PUBLISHED)
# 生成 API Key
api_key = generate_api_key(data.type)

284
api/app/services/app_log_service.py
View File

@@ -1,16 +1,17 @@
"""应用日志服务层"""
import uuid
import datetime as dt
from typing import Optional, Tuple
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.logging_config import get_business_logger
from app.models.app_model import AppType
from app.models.conversation_model import Conversation, Message
from app.models.workflow_model import WorkflowExecution
from app.repositories.conversation_repository import ConversationRepository, MessageRepository
from app.schemas.app_log_schema import AppLogNodeExecution
from app.schemas.app_log_schema import AppLogMessage, AppLogNodeExecution
logger = get_business_logger()
@@ -83,51 +84,40 @@ class AppLogService:
self,
app_id: uuid.UUID,
conversation_id: uuid.UUID,
workspace_id: uuid.UUID
) -> Tuple[Conversation, dict[str, list[AppLogNodeExecution]]]:
workspace_id: uuid.UUID,
app_type: str = AppType.AGENT
) -> Tuple[Conversation, list, dict[str, list[AppLogNodeExecution]]]:
"""
查询会话详情（包含消息和工作流节点执行记录）
Args:
app_id: 应用 ID
conversation_id: 会话 ID
workspace_id: 工作空间 ID
查询会话详情
Returns:
Tuple[Conversation, dict[str, list[AppLogNodeExecution]]]:
(包含消息的会话对象, 按消息ID分组的节点执行记录)
Raises:
ResourceNotFoundException: 当会话不存在时
Tuple[Conversation, list[AppLogMessage|Message], dict[str, list[AppLogNodeExecution]]]
"""
logger.info(
"查询应用日志会话详情",
extra={
"app_id": str(app_id),
"conversation_id": str(conversation_id),
"workspace_id": str(workspace_id)
"workspace_id": str(workspace_id),
"app_type": app_type
}
)
# 查询会话
conversation = self.conversation_repository.get_conversation_for_app_log(
conversation_id=conversation_id,
app_id=app_id,
workspace_id=workspace_id
)
# 查询消息（按时间正序）
messages = self.message_repository.get_messages_by_conversation(
conversation_id=conversation_id
)
# 将消息附加到会话对象
conversation.messages = messages
# 查询工作流节点执行记录（按消息分组）
_, node_executions_map = self._get_workflow_node_executions_with_map(
conversation_id, messages
)
if app_type == AppType.WORKFLOW:
messages, node_executions_map = self._get_workflow_messages_and_nodes(conversation_id)
else:
messages = self.message_repository.get_messages_by_conversation(
conversation_id=conversation_id
)
node_executions_map = self._get_workflow_node_executions_with_map(
conversation_id, messages
)
logger.info(
"查询应用日志会话详情成功",
@@ -139,13 +129,129 @@ class AppLogService:
}
)
return conversation, node_executions_map
return conversation, messages, node_executions_map
def _get_workflow_messages_and_nodes(
self,
conversation_id: uuid.UUID,
) -> Tuple[list[AppLogMessage], dict[str, list[AppLogNodeExecution]]]:
"""
工作流应用专用：从 workflow_executions 构建 messages 和节点日志。
每条 WorkflowExecution 对应一轮对话：
- user message：来自 execution.input_data（content 取 message 字段，files 放 meta_data）
- assistant message：来自 execution.output_data（失败时内容为错误信息）
开场白的 suggested_questions 合并到第一条 assistant message 的 meta_data 里。
Returns:
(messages 列表, node_executions_map)
"""
stmt = (
select(WorkflowExecution)
.where(
WorkflowExecution.conversation_id == conversation_id,
WorkflowExecution.status.in_(["completed", "failed"])
)
.order_by(WorkflowExecution.started_at.asc())
)
executions = list(self.db.scalars(stmt).all())
# 查开场白：Message 表里 meta_data 含 suggested_questions 的第一条 assistant 消息
opening_stmt = (
select(Message)
.where(
Message.conversation_id == conversation_id,
Message.role == "assistant",
)
.order_by(Message.created_at.asc())
.limit(10)
)
early_messages = list(self.db.scalars(opening_stmt).all())
suggested_questions: list = []
for m in early_messages:
if isinstance(m.meta_data, dict) and "suggested_questions" in m.meta_data:
suggested_questions = m.meta_data.get("suggested_questions") or []
break
messages: list[AppLogMessage] = []
node_executions_map: dict[str, list[AppLogNodeExecution]] = {}
# 如果有开场白，作为第一条 assistant 消息插入
if suggested_questions or early_messages:
opening_msg = next(
(m for m in early_messages
if isinstance(m.meta_data, dict) and "suggested_questions" in m.meta_data),
None
)
if opening_msg:
messages.append(AppLogMessage(
id=opening_msg.id,
conversation_id=conversation_id,
role="assistant",
content=opening_msg.content,
status=None,
meta_data={"suggested_questions": suggested_questions},
created_at=opening_msg.created_at,
))
for execution in executions:
started_at = execution.started_at or dt.datetime.now()
completed_at = execution.completed_at or started_at
# assistant message 的 id，同时作为 node_executions_map 的 key
assistant_msg_id = uuid.uuid5(execution.id, "assistant")
# --- user message（输入）---
input_data = execution.input_data or {}
input_content = input_data.get("message") or _extract_text(input_data)
# 跳过没有用户输入的 execution（如开场白触发的记录）
if not input_content or not input_content.strip():
continue
files = input_data.get("files") or []
user_msg = AppLogMessage(
id=uuid.uuid5(execution.id, "user"),
conversation_id=conversation_id,
role="user",
content=input_content,
meta_data={"files": files} if files else None,
created_at=started_at,
)
messages.append(user_msg)
# --- assistant message（输出）---
if execution.status == "completed":
output_content = _extract_text(execution.output_data)
meta = {"usage": execution.token_usage or {}, "elapsed_time": execution.elapsed_time}
else:
output_content = _extract_text(execution.output_data) or ""
meta = {"error": execution.error_message, "error_node_id": execution.error_node_id}
assistant_msg = AppLogMessage(
id=assistant_msg_id,
conversation_id=conversation_id,
role="assistant",
content=output_content,
status=execution.status,
meta_data=meta,
created_at=completed_at,
)
messages.append(assistant_msg)
# --- 节点执行记录，从 workflow_executions.output_data["node_outputs"] 读取 ---
execution_nodes = _build_nodes_from_output_data(execution.output_data)
if execution_nodes:
node_executions_map[str(assistant_msg_id)] = execution_nodes
return messages, node_executions_map
def _get_workflow_node_executions_with_map(
self,
conversation_id: uuid.UUID,
messages: list[Message]
) -> Tuple[list[AppLogNodeExecution], dict[str, list[AppLogNodeExecution]]]:
) -> dict[str, list[AppLogNodeExecution]]:
"""
从 workflow_executions 表中提取节点执行记录，并按 assistant message 分组
@@ -157,13 +263,12 @@ class AppLogService:
Tuple[list[AppLogNodeExecution], dict[str, list[AppLogNodeExecution]]]:
(所有节点执行记录列表, 按 message_id 分组的节点执行记录字典)
"""
node_executions = []
node_executions_map: dict[str, list[AppLogNodeExecution]] = {}
# 查询该会话关联的所有工作流执行记录（按时间正序）
stmt = select(WorkflowExecution).where(
WorkflowExecution.conversation_id == conversation_id,
WorkflowExecution.status == "completed"
WorkflowExecution.status.in_(["completed", "failed"])
).order_by(WorkflowExecution.started_at.asc())
executions = self.db.scalars(stmt).all()
@@ -188,10 +293,18 @@ class AppLogService:
used_message_ids: set[str] = set()
for execution in executions:
if not execution.output_data:
# 构建节点执行记录列表，从 workflow_executions.output_data["node_outputs"] 读取
execution_nodes = _build_nodes_from_output_data(execution.output_data)
if not execution_nodes:
continue
# 找到该 execution 对应的 assistant message
# 失败的执行没有 assistant message，直接用 execution id 作为 key
if execution.status == "failed":
node_executions_map[f"execution_{str(execution.id)}"] = execution_nodes
continue
# completed：通过时序匹配关联到对应的 assistant message
# 逻辑：找 execution.started_at 之后最近的、未使用的 assistant message
best_msg = None
best_dt = None
@@ -200,9 +313,9 @@ class AppLogService:
if msg_id_str in used_message_ids:
continue
if msg.created_at and msg.created_at >= execution.started_at:
dt = (msg.created_at - execution.started_at).total_seconds()
if best_dt is None or dt < best_dt:
best_dt = dt
delta = (msg.created_at - execution.started_at).total_seconds()
if best_dt is None or delta < best_dt:
best_dt = delta
best_msg = msg
if not best_msg:
@@ -210,31 +323,76 @@ class AppLogService:
msg_id_str = str(best_msg.id)
used_message_ids.add(msg_id_str)
node_executions_map[msg_id_str] = execution_nodes
# 提取节点输出
output_data = execution.output_data
if isinstance(output_data, dict):
node_outputs = output_data.get("node_outputs", {})
execution_nodes = []
for node_id, node_data in node_outputs.items():
if not isinstance(node_data, dict):
continue
node_execution = AppLogNodeExecution(
node_id=node_data.get("node_id", node_id),
node_type=node_data.get("node_type", "unknown"),
node_name=node_data.get("node_name"),
status=node_data.get("status", "unknown"),
error=node_data.get("error"),
input=node_data.get("input"),
process=node_data.get("process"),
output=node_data.get("output"),
elapsed_time=node_data.get("elapsed_time"),
token_usage=node_data.get("token_usage"),
)
node_executions.append(node_execution)
execution_nodes.append(node_execution)
return node_executions_map
# 将节点记录关联到 message_id
node_executions_map[msg_id_str] = execution_nodes
return node_executions, node_executions_map
def _extract_text(data: Optional[dict]) -> str:
"""从 workflow execution 的 input_data / output_data 中提取可读文本。
优先取 'text'、'content'、'output' 字段；若都没有则 JSON 序列化整个 dict。
"""
if not data:
return ""
for key in ("message", "text", "content", "output", "result", "answer"):
if key in data and isinstance(data[key], str):
return data[key]
import json
return json.dumps(data, ensure_ascii=False)
def _build_nodes_from_output_data(output_data: Optional[dict]) -> list[AppLogNodeExecution]:
"""从 workflow_executions.output_data["node_outputs"] 构建节点执行记录列表。
output_data 结构：
{
"node_outputs": {
"<node_id>": {
"node_type": ...,
"node_name": ...,
"status": ...,
"input": ...,
"output": ...,
"elapsed_time": ...,
"token_usage": ...,
"error": ...,
"cycle_items": [...],
...
}
},
"error": ...,
...
}
"""
if not output_data:
return []
node_outputs: dict = output_data.get("node_outputs") or {}
result = []
for node_id, node_data in node_outputs.items():
if not isinstance(node_data, dict):
continue
output = dict(node_data)
cycle_items = output.pop("cycle_items", None)
# 把已知的顶层字段剥离，剩余的作为 output
node_type = output.pop("node_type", "unknown")
node_name = output.pop("node_name", None)
status = output.pop("status", "completed")
error = output.pop("error", None)
inp = output.pop("input", None)
elapsed_time = output.pop("elapsed_time", None)
token_usage = output.pop("token_usage", None)
result.append(AppLogNodeExecution(
node_id=node_id,
node_type=node_type,
node_name=node_name,
status=status,
error=error,
input=inp,
process=None,
output=output if output else None,
cycle_items=cycle_items,
elapsed_time=elapsed_time,
token_usage=token_usage,
))
return result

30
api/app/services/memory_api_service.py
View File

@@ -10,6 +10,7 @@ from typing import Any, Dict, Optional
from sqlalchemy.orm import Session
from app.celery_task_scheduler import scheduler
from app.core.error_codes import BizCode
from app.core.exceptions import BusinessException, ResourceNotFoundException
from app.core.logging_config import get_logger
@@ -166,20 +167,31 @@ class MemoryAPIService:
# Convert to message list format expected by write_message_task
messages = message if isinstance(message, list) else [{"role": "user", "content": message}]
from app.tasks import write_message_task
task = write_message_task.delay(
# from app.tasks import write_message_task
# task = write_message_task.delay(
# end_user_id,
# messages,
# config_id,
# storage_type,
# user_rag_memory_id or "",
# )
task_id = scheduler.push_task(
"app.core.memory.agent.write_message",
end_user_id,
messages,
config_id,
storage_type,
user_rag_memory_id or "",
{
"end_user_id": end_user_id,
"message": messages,
"config_id": config_id,
"storage_type": storage_type,
"user_rag_memory_id": user_rag_memory_id or ""
}
)
logger.info(f"Memory write task submitted: task_id={task.id}, end_user_id={end_user_id}")
logger.info(f"Memory write task submitted, task_id={task_id} end_user_id={end_user_id}")
return {
"task_id": task.id,
"status": "PENDING",
"task_id": task_id,
"status": "QUEUED",
"end_user_id": end_user_id,
}

207
api/app/services/memory_explicit_service.py
View File

@@ -4,7 +4,7 @@
处理显性记忆相关的业务逻辑，包括情景记忆和语义记忆的查询。
"""
from typing import Any, Dict
from typing import Any, Dict, Optional
from app.core.logging_config import get_logger
from app.services.memory_base_service import MemoryBaseService
@@ -104,7 +104,7 @@ class MemoryExplicitService(MemoryBaseService):
e.description AS core_definition
ORDER BY e.name ASC
"""
semantic_result = await self.neo4j_connector.execute_query(
semantic_query,
end_user_id=end_user_id
@@ -146,6 +146,209 @@ class MemoryExplicitService(MemoryBaseService):
logger.error(f"获取显性记忆总览时出错: {str(e)}", exc_info=True)
raise
async def get_episodic_memory_list(
self,
end_user_id: str,
page: int,
pagesize: int,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
episodic_type: str = "all",
) -> Dict[str, Any]:
"""
获取情景记忆分页列表
Args:
end_user_id: 终端用户ID
page: 页码
pagesize: 每页数量
start_date: 开始时间戳（毫秒），可选
end_date: 结束时间戳（毫秒），可选
episodic_type: 情景类型筛选
Returns:
{
"total": int, # 该用户情景记忆总数（不受筛选影响）
"items": [...], # 当前页数据
"page": {
"page": int,
"pagesize": int,
"total": int, # 筛选后总数
"hasnext": bool
}
}
"""
try:
logger.info(
f"情景记忆分页查询: end_user_id={end_user_id}, "
f"start_date={start_date}, end_date={end_date}, "
f"episodic_type={episodic_type}, page={page}, pagesize={pagesize}"
)
# 1. 查询情景记忆总数（不受筛选条件限制）
total_all_query = """
MATCH (s:MemorySummary)
WHERE s.end_user_id = $end_user_id
RETURN count(s) AS total
"""
total_all_result = await self.neo4j_connector.execute_query(
total_all_query, end_user_id=end_user_id
)
total_all = total_all_result[0]["total"] if total_all_result else 0
# 2. 构建筛选条件
where_clauses = ["s.end_user_id = $end_user_id"]
params = {"end_user_id": end_user_id}
# 时间戳筛选（毫秒时间戳转为 UTC ISO 字符串，使用 Neo4j datetime() 精确比较）
if start_date is not None and end_date is not None:
from datetime import datetime, timezone
start_dt = datetime.fromtimestamp(start_date / 1000, tz=timezone.utc)
end_dt = datetime.fromtimestamp(end_date / 1000, tz=timezone.utc)
# 开始时间取当天 UTC 00:00:00，结束时间取当天 UTC 23:59:59.999999
start_iso = start_dt.strftime("%Y-%m-%dT") + "00:00:00.000000"
end_iso = end_dt.strftime("%Y-%m-%dT") + "23:59:59.999999"
where_clauses.append("datetime(s.created_at) >= datetime($start_iso) AND datetime(s.created_at) <= datetime($end_iso)")
params["start_iso"] = start_iso
params["end_iso"] = end_iso
# 类型筛选下推到 Cypher（兼容中英文）
if episodic_type != "all":
type_mapping = {
"conversation": "对话",
"project_work": "项目/工作",
"learning": "学习",
"decision": "决策",
"important_event": "重要事件"
}
chinese_type = type_mapping.get(episodic_type)
if chinese_type:
where_clauses.append(
"(s.memory_type = $episodic_type OR s.memory_type = $chinese_type)"
)
params["episodic_type"] = episodic_type
params["chinese_type"] = chinese_type
else:
where_clauses.append("s.memory_type = $episodic_type")
params["episodic_type"] = episodic_type
where_str = " AND ".join(where_clauses)
# 3. 查询筛选后的总数
count_query = f"""
MATCH (s:MemorySummary)
WHERE {where_str}
RETURN count(s) AS total
"""
count_result = await self.neo4j_connector.execute_query(count_query, **params)
filtered_total = count_result[0]["total"] if count_result else 0
# 4. 查询分页数据
skip = (page - 1) * pagesize
data_query = f"""
MATCH (s:MemorySummary)
WHERE {where_str}
RETURN elementId(s) AS id,
s.name AS title,
s.memory_type AS memory_type,
s.content AS content,
s.created_at AS created_at
ORDER BY s.created_at DESC
SKIP $skip LIMIT $limit
"""
params["skip"] = skip
params["limit"] = pagesize
result = await self.neo4j_connector.execute_query(data_query, **params)
# 5. 处理结果
items = []
if result:
for record in result:
raw_created_at = record.get("created_at")
created_at_timestamp = self.parse_timestamp(raw_created_at)
items.append({
"id": record["id"],
"title": record.get("title") or "未命名",
"memory_type": record.get("memory_type") or "其他",
"content": record.get("content") or "",
"created_at": created_at_timestamp
})
# 6. 构建返回结果
return {
"total": total_all,
"items": items,
"page": {
"page": page,
"pagesize": pagesize,
"total": filtered_total,
"hasnext": (page * pagesize) < filtered_total
}
}
except Exception as e:
logger.error(f"情景记忆分页查询出错: {str(e)}", exc_info=True)
raise
async def get_semantic_memory_list(
self,
end_user_id: str
) -> list:
"""
获取语义记忆全量列表
Args:
end_user_id: 终端用户ID
Returns:
[
{
"id": str,
"name": str,
"entity_type": str,
"core_definition": str
}
]
"""
try:
logger.info(f"语义记忆列表查询: end_user_id={end_user_id}")
semantic_query = """
MATCH (e:ExtractedEntity)
WHERE e.end_user_id = $end_user_id
AND e.is_explicit_memory = true
RETURN elementId(e) AS id,
e.name AS name,
e.entity_type AS entity_type,
e.description AS core_definition
ORDER BY e.name ASC
"""
result = await self.neo4j_connector.execute_query(
semantic_query, end_user_id=end_user_id
)
items = []
if result:
for record in result:
items.append({
"id": record["id"],
"name": record.get("name") or "未命名",
"entity_type": record.get("entity_type") or "未分类",
"core_definition": record.get("core_definition") or ""
})
logger.info(f"语义记忆列表查询成功: end_user_id={end_user_id}, total={len(items)}")
return items
except Exception as e:
logger.error(f"语义记忆列表查询出错: {str(e)}", exc_info=True)
raise
async def get_explicit_memory_details(
self,
end_user_id: str,

7
api/app/services/tool_service.py
View File

@@ -815,11 +815,12 @@ class ToolService:
"default": param_info.get("default")
})
# 请求体参数
# 请求体参数 — _extract_request_body 返回 {"schema": {...}, "required": bool, ...}
request_body = operation.get("request_body")
if request_body:
schema_props = request_body.get("schema", {}).get("properties", {})
required_props = request_body.get("schema", {}).get("required", [])
body_schema = request_body.get("schema", {})
schema_props = body_schema.get("properties", {})
required_props = body_schema.get("required", [])
for prop_name, prop_schema in schema_props.items():
parameters.append({

25
api/app/services/workflow_service.py
View File

@@ -17,8 +17,9 @@ from app.core.workflow.executor import execute_workflow, execute_workflow_stream
from app.core.workflow.nodes.enums import NodeType
from app.core.workflow.validator import validate_workflow_config
from app.db import get_db
from sqlalchemy import select
from app.models import App
from app.models.workflow_model import WorkflowConfig, WorkflowExecution
from app.models.workflow_model import WorkflowConfig, WorkflowExecution, WorkflowNodeExecution
from app.repositories import knowledge_repository
from app.repositories.workflow_repository import (
WorkflowConfigRepository,
@@ -918,6 +919,7 @@ class WorkflowService:
input_data["conv_messages"] = conv_messages
init_message_length = len(input_data.get("conv_messages", []))
message_id = uuid.uuid4()
_cycle_items: dict[str, list] = {}
# 新会话时写入开场白
is_new_conversation = init_message_length == 0
@@ -948,6 +950,15 @@ class WorkflowService:
memory_storage_type=storage_type,
user_rag_memory_id=user_rag_memory_id
):
event_type = event.get("event")
event_data = event.get("data", {})
if event_type == "cycle_item":
cycle_id = event_data.get("cycle_id")
if cycle_id not in _cycle_items:
_cycle_items[cycle_id] = []
_cycle_items[cycle_id].append(event_data)
if event.get("event") == "workflow_end":
status = event.get("data", {}).get("status")
token_usage = event.get("data", {}).get("token_usage", {}) or {}
@@ -1019,6 +1030,18 @@ class WorkflowService:
)
else:
logger.error(f"unexpect workflow run status, status: {status}")
# 把积累的 cycle_item 写入 workflow_executions.output_data["node_outputs"]
if _cycle_items and execution.output_data:
import copy
new_output_data = copy.deepcopy(execution.output_data)
node_outputs = new_output_data.setdefault("node_outputs", {})
for cycle_node_id, items in _cycle_items.items():
if cycle_node_id in node_outputs:
node_outputs[cycle_node_id]["cycle_items"] = items
else:
node_outputs[cycle_node_id] = {"cycle_items": items}
execution.output_data = new_output_data
self.db.commit()
elif event.get("event") == "workflow_start":
event["data"]["message_id"] = str(message_id)
event = self._emit(public, event)