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

Merge branch 'refs/heads/develop' into fix/memory_bug_fix

Browse Source
This commit is contained in:
lixinyue
2026-01-19 18:04:29 +08:00
parent 5a0d3df689 9d25b08641
commit 17504b1b9c
102 changed files with 3013 additions and 1540 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
api/app/core/config.py
View File

@@ -38,6 +38,7 @@ class Settings:
REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
REDIS_DB: int = int(os.getenv("REDIS_DB", "1"))
REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD", "")
# ElasticSearch configuration
ELASTICSEARCH_HOST: str = os.getenv("ELASTICSEARCH_HOST", "https://127.0.0.1")

29
api/app/core/rag/llm/cv_model.py
View File

@@ -243,6 +243,33 @@ class QWenCV(GptV4):
tmp_path = tmp.name
video_path = f"file://{tmp_path}"
prompt_ch = """
你是一名专业的视频转录助手，能够将视频文件的内容转写为文本，并**精确标记每句话或每个段落对应的时间戳**（开始时间-结束时间）。\n
**任务要求**：
1.输入是MP4等视频文件,解析带时间戳的文本。
2.时间戳格式为 `[HH:MM:SS.mmm]`（毫秒可选），例如 `[00:01:23.456]`。
3.时间戳需尽可能贴近实际视频的起止时间，误差不超过1秒。
4.如果无法确定具体时间，请根据上下文合理估算。
5.最后总结:这段视频的内容是什么?,并用恰当的句子总结这个视频。
**示例输出**：
[00:00:00.000] 今天天气真好，
[00:00:02.500] 我们一起去公园散步吧。
[00:00:05.800] 公园里的花开得非常漂亮。
这段视频的内容是关于如何在CREAMS系统中进行楼宇管理集合的编辑或删除操作。视频演示了 ..."""
prompt_en = """
You are a professional video transcription assistant, capable of transcribing the content of video files into text and **precisely marking the timestamp (start time-end time) corresponding to each sentence or paragraph**.
**Task requirements**:
1. Input is MP4 or other video files, and parse the text with timestamps.
2. The timestamp format is `[HH:MM:SS.mmm]` (milliseconds are optional), for example, `[00:01:23.456]`.
3. The timestamp should be as close as possible to the actual start and end time of the video, with an error not exceeding 1 second.
4. If the specific time cannot be determined, please make a reasonable estimation based on the context.
5. Final summary: What is the content of this video? Summarize this video in an appropriate sentence.
**Example output**:
[00:00:00.000] The weather is really nice today, [00:00:02.500] let's go for a walk in the park together.
[00:00:05.800] The flowers in the park are blooming beautifully.
The content of this video is about how to edit or delete building management collections in the CREAMS system. The video demonstrates .."""
messages = [
{
"role": "user",
@@ -252,7 +279,7 @@ class QWenCV(GptV4):
"fps": 2,
},
{
"text": "视频的内容是什么?,并且，请用恰当的句子总结这个视频。" if self.lang.lower() == "chinese" else "What is the content of the video? And please summarize this video in proper sentences.",
"text": prompt_ch if self.lang.lower() == "chinese" else prompt_en,
},
],
}

30
api/app/core/rag/llm/sequence2txt_model.py
View File

@@ -60,6 +60,34 @@ class QWenSeq2txt(Base):
from dashscope import MultiModalConversation
audio_path = f"file://{audio_path}"
prompt_ch = """
你是一名专业的音频转录助手，能够将MP3音频文件的内容转写为文本，并**精确标记每句话或每个段落对应的时间戳**（开始时间-结束时间）。\n
**任务要求**：
1.输入是MP3,解析带时间戳的文本。
2.时间戳格式为 `[HH:MM:SS.mmm]`（毫秒可选），例如 `[00:01:23.456]`。
3.时间戳需尽可能贴近实际语音的起止时间，误差不超过1秒。
4.如果无法确定具体时间，请根据上下文合理估算。
5.最后总结:这段音频在说什么?
**示例输出**：
[00:00:00.000] 今天天气真好，
[00:00:02.500] 我们一起去公园散步吧。
[00:00:05.800] 公园里的花开得非常漂亮。
这段音频讲述的是一个关于**“吃水不忘挖井人”**的感人故事，主 ..."""
prompt_en = """
You are a professional audio transcription assistant, capable of transcribing the content of MP3 audio files into text and **precisely marking the timestamps (start time - end time) corresponding to each sentence or paragraph**.
**Task requirements**:
1. Input is MP3, parse text with timestamps.
2. The timestamp format is `[HH:MM:SS.mmm]` (milliseconds are optional), for example, `[00:01:23.456]`.
3. The timestamp should be as close as possible to the actual start and end time of the voice, with an error not exceeding 1 second.
4. If a specific time cannot be determined, please make a reasonable estimation based on the context.
5. Final summary: What is this audio talking about?
**Example Output**:
[00:00:00.000] The weather is really nice today,
[00:00:02.500] let's go for a walk in the park together.
[00:00:05.800] The flowers in the park are blooming beautifully.
This audio tells a touching story about **"Remembering the one who dug the well when drinking water"** .."""
messages = [
{
"role": "user",
@@ -68,7 +96,7 @@ class QWenSeq2txt(Base):
"audio": audio_path
},
{
"text": "这段音频在说什么?" if self.lang.lower() == "chinese" else "What is this audio saying?",
"text": prompt_ch if self.lang.lower() == "chinese" else prompt_en,
},
],
}

38
api/app/core/workflow/executor.py
View File

@@ -8,6 +8,7 @@ import logging
import uuid
from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.graph.state import CompiledStateGraph
from app.core.workflow.graph_builder import GraphBuilder
@@ -53,11 +54,11 @@ class WorkflowExecutor:
self.edges = workflow_config.get("edges", [])
self.execution_config = workflow_config.get("execution_config", {})
self.checkpoint_config = {
"configurable": {
self.checkpoint_config = RunnableConfig(
configurable={
"thread_id": uuid.uuid4(),
}
}
)
def _prepare_initial_state(self, input_data: dict[str, Any]) -> WorkflowState:
"""准备初始状态（注入系统变量和会话变量）
@@ -214,13 +215,13 @@ class WorkflowExecutor:
return {
"status": "completed",
"output": final_output,
"variables": result.get("variables", {}),
"node_outputs": node_outputs,
"messages": result.get("messages", []),
"conversation_id": conversation_id,
"elapsed_time": elapsed_time,
"token_usage": token_usage,
"error": result.get("error"),
"variables": result.get("variables", {}),
}
def build_graph(self, stream=False) -> CompiledStateGraph:
@@ -326,11 +327,10 @@ class WorkflowExecutor:
}
# 1. 构建图
graph = self.build_graph(True)
graph = self.build_graph(stream=True)
# 2. 初始化状态（自动注入系统变量）
initial_state = self._prepare_initial_state(input_data)
# 3. Execute workflow
try:
chunk_count = 0
@@ -346,14 +346,16 @@ class WorkflowExecutor:
mode, data = event
else:
# Unexpected format, log and skip
logger.warning(f"[STREAM] Unexpected event format: {type(event)}, value: {event}")
logger.warning(f"[STREAM] Unexpected event format: {type(event)}, value: {event}"
f"- execution_id: {self.execution_id}")
continue
if mode == "custom":
# Handle custom streaming events (chunks from nodes via stream writer)
chunk_count += 1
event_type = data.get("type", "node_chunk") # "message" or "node_chunk"
logger.info(f"[CUSTOM] ✅ 收到 {event_type} #{chunk_count} from {data.get('node_id')}")
logger.info(f"[CUSTOM] ✅ 收到 {event_type} #{chunk_count} from {data.get('node_id')}"
f"- execution_id: {self.execution_id}")
yield {
"event": event_type, # "message" or "node_chunk"
"data": {
@@ -380,7 +382,8 @@ class WorkflowExecutor:
variables_sys = variables.get("sys", {})
conversation_id = input_data.get("conversation_id")
execution_id = variables_sys.get("execution_id")
logger.info(f"[DEBUG] Node starts execution: {node_name}")
logger.info(f"[NODE-START] Node starts execution: {node_name} "
f"- execution_id: {self.execution_id}")
yield {
"event": "node_start",
@@ -399,7 +402,8 @@ class WorkflowExecutor:
variables_sys = variables.get("sys", {})
conversation_id = input_data.get("conversation_id")
execution_id = variables_sys.get("execution_id")
logger.info(f"[DEBUG] Node execution completed: {node_name}")
logger.info(f"[NODE-END] Node execution completed: {node_name} "
f"- execution_id: {self.execution_id}")
yield {
"event": "node_end",
@@ -407,13 +411,15 @@ class WorkflowExecutor:
"node_id": node_name,
"conversation_id": conversation_id,
"execution_id": execution_id,
"timestamp": data.get("timestamp")
"timestamp": data.get("timestamp"),
"state": result.get("node_outputs", {}).get(node_name),
}
}
elif mode == "updates":
# Handle state updates - store final state
logger.debug(f"[UPDATES] 收到 state 更新 from {list(data.keys())}")
logger.debug(f"[UPDATES] 收到 state 更新 from {list(data.keys())} "
f"- execution_id: {self.execution_id}")
# 计算耗时
end_time = datetime.datetime.now()
@@ -421,7 +427,7 @@ class WorkflowExecutor:
result = graph.get_state(self.checkpoint_config).values
logger.info(
f"Workflow execution completed (streaming), "
f"total chunks: {chunk_count}, elapsed: {elapsed_time:.2f}s"
f"total chunks: {chunk_count}, elapsed: {elapsed_time:.2f}s, execution_id: {self.execution_id}"
)
# 发送 workflow_end 事件
@@ -449,7 +455,8 @@ class WorkflowExecutor:
}
}
def _extract_final_output(self, node_outputs: dict[str, Any]) -> str | None:
@staticmethod
def _extract_final_output(node_outputs: dict[str, Any]) -> str | None:
"""从节点输出中提取最终输出
优先级：
@@ -473,7 +480,8 @@ class WorkflowExecutor:
return None
def _aggregate_token_usage(self, node_outputs: dict[str, Any]) -> dict[str, int] | None:
@staticmethod
def _aggregate_token_usage(node_outputs: dict[str, Any]) -> dict[str, int] | None:
"""聚合所有节点的 token 使用情况
Args:

2
api/app/core/workflow/nodes/base_node.py
View File

@@ -25,7 +25,7 @@ class WorkflowState(TypedDict):
The state object passed between nodes in a workflow, containing messages, variables, node outputs, etc.
"""
# List of messages (append mode)
messages: list[dict[str, str]]
messages: Annotated[list[dict[str, str]], lambda x, y: y]
# Set of loop node IDs, used for assigning values in loop nodes
cycle_nodes: list

16
api/app/core/workflow/nodes/cycle_graph/iteration.py
View File

@@ -21,6 +21,7 @@ class IterationRuntime:
optional parallel execution, flattening of output, and loop control via
the workflow state.
"""
def __init__(
self,
graph: CompiledStateGraph,
@@ -87,6 +88,7 @@ class IterationRuntime:
self.result.append(output)
if not result["looping"]:
self.looping = False
return result
def _create_iteration_tasks(self, array_obj, idx):
"""
@@ -124,7 +126,7 @@ class IterationRuntime:
array_obj = VariablePool(self.state).get(input_expression)
if not isinstance(array_obj, list):
raise RuntimeError("Cannot iterate over a non-list variable")
child_state = []
idx = 0
if self.typed_config.parallel:
# Execute iterations in parallel batches
@@ -132,15 +134,14 @@ class IterationRuntime:
tasks = self._create_iteration_tasks(array_obj, idx)
logger.info(f"Iteration node {self.node_id}: running, concurrency {len(tasks)}")
idx += self.typed_config.parallel_count
await asyncio.gather(*tasks)
logger.info(f"Iteration node {self.node_id}: execution completed")
return self.result
child_state.extend(await asyncio.gather(*tasks))
else:
# Execute iterations sequentially
while idx < len(array_obj) and self.looping:
logger.info(f"Iteration node {self.node_id}: running")
item = array_obj[idx]
result = await self.graph.ainvoke(self._init_iteration_state(item, idx))
child_state.append(result)
output = VariablePool(result).get(self.output_value)
if isinstance(output, list) and self.typed_config.flatten:
self.result.extend(output)
@@ -150,5 +151,8 @@ class IterationRuntime:
self.looping = False
idx += 1
logger.info(f"Iteration node {self.node_id}: execution completed")
return self.result
logger.info(f"Iteration node {self.node_id}: execution completed")
return {
"output": self.result,
"__child_state": child_state
}

13
api/app/core/workflow/nodes/cycle_graph/loop.py
View File

@@ -67,7 +67,9 @@ class LoopRuntime:
variables=pool.get_all_conversation_vars(),
node_outputs=pool.get_all_node_outputs(),
system_vars=pool.get_all_system_vars(),
) if variable.input_type == ValueInputType.VARIABLE else TypeTransformer.transform(variable.value, variable.type)
)
if variable.input_type == ValueInputType.VARIABLE
else TypeTransformer.transform(variable.value, variable.type)
for variable in self.typed_config.cycle_vars
}
self.state["node_outputs"][self.node_id] = {
@@ -76,7 +78,9 @@ class LoopRuntime:
variables=pool.get_all_conversation_vars(),
node_outputs=pool.get_all_node_outputs(),
system_vars=pool.get_all_system_vars(),
) if variable.input_type == ValueInputType.VARIABLE else TypeTransformer.transform(variable.value, variable.type)
)
if variable.input_type == ValueInputType.VARIABLE
else TypeTransformer.transform(variable.value, variable.type)
for variable in self.typed_config.cycle_vars
}
loopstate = WorkflowState(
@@ -171,10 +175,11 @@ class LoopRuntime:
"""
loopstate = self._init_loop_state()
loop_time = self.typed_config.max_loop
child_state = []
while self.evaluate_conditional(loopstate) and loopstate["looping"] and loop_time > 0:
logger.info(f"loop node {self.node_id}: running")
await self.graph.ainvoke(loopstate)
child_state.append(await self.graph.ainvoke(loopstate))
loop_time -= 1
logger.info(f"loop node {self.node_id}: execution completed")
return loopstate["runtime_vars"][self.node_id]
return loopstate["runtime_vars"][self.node_id] | {"__child_state": child_state}

7
api/app/core/workflow/nodes/knowledge/node.py
View File

@@ -10,9 +10,8 @@ from app.core.workflow.nodes.base_node import BaseNode, WorkflowState
from app.core.workflow.nodes.knowledge import KnowledgeRetrievalNodeConfig
from app.db import get_db_read
from app.models import knowledge_model, knowledgeshare_model, ModelType
from app.repositories import knowledge_repository
from app.repositories import knowledge_repository, knowledgeshare_repository
from app.schemas.chunk_schema import RetrieveType
from app.services import knowledge_service, knowledgeshare_service
from app.services.model_service import ModelConfigService
logger = logging.getLogger(__name__)
@@ -96,7 +95,7 @@ class KnowledgeRetrievalNode(BaseNode):
filters = self._build_kb_filter(kb_ids, knowledge_model.PermissionType.Share)
share_ids = knowledge_service.knowledge_repository.get_chunked_knowledgeids(
share_ids = knowledge_repository.get_chunked_knowledgeids(
db=db,
filters=filters
)
@@ -105,7 +104,7 @@ class KnowledgeRetrievalNode(BaseNode):
filters = [
knowledgeshare_model.KnowledgeShare.target_kb_id.in_(kb_ids)
]
items = knowledgeshare_service.knowledgeshare_repository.get_source_kb_ids_by_target_kb_id(
items = knowledgeshare_repository.get_source_kb_ids_by_target_kb_id(
db=db,
filters=filters
)

2
api/app/core/workflow/nodes/llm/config.py
View File

@@ -66,7 +66,7 @@ class LLMNodeConfig(BaseNodeConfig):
)
memory: MemoryWindowSetting = Field(
...,
default_factory=MemoryWindowSetting,
description="对话上下文窗口"
)

20
api/app/core/workflow/nodes/llm/node.py
View File

@@ -85,6 +85,7 @@ class LLMNode(BaseNode):
"""
# 1. 处理消息格式（优先使用 messages）
self.typed_config = LLMNodeConfig(**self.config)
messages_config = self.typed_config.messages
if messages_config:
@@ -167,7 +168,7 @@ class LLMNode(BaseNode):
Returns:
LLM 响应消息
"""
self.typed_config = LLMNodeConfig(**self.config)
# self.typed_config = LLMNodeConfig(**self.config)
llm, prompt_or_messages = self._prepare_llm(state, True)
logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（非流式）")
@@ -269,12 +270,16 @@ class LLMNode(BaseNode):
chunk_count = 0
# 调用 LLM（流式，支持字符串或消息列表）
async for chunk in llm.astream(prompt_or_messages):
last_meta_data = {}
async for chunk in llm.astream(prompt_or_messages, stream_usage=True):
# 提取内容
if hasattr(chunk, 'content'):
content = chunk.content
else:
content = str(chunk)
if hasattr(chunk, 'response_metadata'):
if chunk.response_metadata:
last_meta_data = chunk.response_metadata
# 只有当内容不为空时才处理
if content:
@@ -288,13 +293,10 @@ class LLMNode(BaseNode):
logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(full_response)}, 总 chunks: {chunk_count}")
# 构建完整的 AIMessage（包含元数据）
if isinstance(last_chunk, AIMessage):
final_message = AIMessage(
content=full_response,
response_metadata=last_chunk.response_metadata if hasattr(last_chunk, 'response_metadata') else {}
)
else:
final_message = AIMessage(content=full_response)
final_message = AIMessage(
content=full_response,
response_metadata=last_meta_data
)
# yield 完成标记
yield {"__final__": True, "result": final_message}