fix: 修复部分场景下历史记录保存异常的问题。回退到使用 AsyncSaver的双线备份消息模式

This commit is contained in:
Wenjie Zhang 2025-10-12 16:04:46 +08:00
parent 2ac690eb81
commit 3600c7f9b7
5 changed files with 42 additions and 14 deletions

View File

@ -166,7 +166,7 @@ async def chat_agent(
existing_count = len(existing_messages) existing_count = len(existing_messages)
# 只保存新增的消息 # 只保存新增的消息
new_messages = messages new_messages = messages[existing_count :]
for msg in new_messages: for msg in new_messages:
msg_dict = msg.model_dump() if hasattr(msg, "model_dump") else {} msg_dict = msg.model_dump() if hasattr(msg, "model_dump") else {}

View File

@ -26,10 +26,8 @@ class ChatbotAgent(BaseAgent):
def __init__(self, **kwargs): def __init__(self, **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
self.graph = None self.graph = None
self.checkpointer = InMemorySaver() self.checkpointer = None
self.context_schema = Context self.context_schema = Context
self.workdir = Path(sys_config.save_dir) / "agents" / self.module_name
self.workdir.mkdir(parents=True, exist_ok=True)
self.agent_tools = None self.agent_tools = None
def get_tools(self): def get_tools(self):
@ -103,21 +101,14 @@ class ChatbotAgent(BaseAgent):
builder.add_edge("tools", "chatbot") builder.add_edge("tools", "chatbot")
builder.add_edge("chatbot", END) builder.add_edge("chatbot", END)
self.checkpointer = await self._get_checkpointer()
graph = builder.compile(checkpointer=self.checkpointer, name=self.name) graph = builder.compile(checkpointer=self.checkpointer, name=self.name)
self.graph = graph self.graph = graph
return graph return graph
def main(): def main():
agent = ChatbotAgent(Context) pass
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
from src.agents.utils import agent_cli
agent_cli(agent, config)
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@ -1,10 +1,14 @@
from __future__ import annotations from __future__ import annotations
import os
from pathlib import Path
from abc import abstractmethod from abc import abstractmethod
from langgraph.graph.state import CompiledStateGraph from langgraph.graph.state import CompiledStateGraph
from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver, aiosqlite
from src import config as sys_config
from src.agents.common.context import BaseContext from src.agents.common.context import BaseContext
from src.utils import logger from src.utils import logger
@ -19,8 +23,10 @@ class BaseAgent:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.graph = None # will be covered by get_graph self.graph = None # will be covered by get_graph
self.checkpointer = InMemorySaver() self.checkpointer = None
self.context_schema = BaseContext self.context_schema = BaseContext
self.workdir = Path(sys_config.save_dir) / "agents" / self.module_name
self.workdir.mkdir(parents=True, exist_ok=True)
@property @property
def module_name(self) -> str: def module_name(self) -> str:
@ -103,3 +109,26 @@ class BaseAgent:
例如: graph = workflow.compile(checkpointer=sqlite_checkpointer) 例如: graph = workflow.compile(checkpointer=sqlite_checkpointer)
""" """
pass pass
async def _get_checkpointer(self):
# 创建数据库连接并确保设置 checkpointer
checkpointer = None
try:
checkpointer = AsyncSqliteSaver(await self.get_async_conn())
except Exception as e:
logger.error(f"构建 Graph 设置 checkpointer 时出错: {e}, 尝试使用内存存储")
checkpointer = InMemorySaver()
return checkpointer
async def get_async_conn(self) -> aiosqlite.Connection:
"""获取异步数据库连接"""
return await aiosqlite.connect(os.path.join(self.workdir, "aio_history.db"))
async def get_aio_memory(self) -> AsyncSqliteSaver:
"""获取异步存储实例"""
return AsyncSqliteSaver(await self.get_async_conn())

View File

@ -35,7 +35,9 @@ class ReActAgent(BaseAgent):
return self.graph return self.graph
available_tools = get_buildin_tools() available_tools = get_buildin_tools()
self.checkpointer = await self._get_checkpointer()
# 创建 ReActAgent
graph = create_react_agent(model, tools=available_tools, prompt=prompt, checkpointer=self.checkpointer) graph = create_react_agent(model, tools=available_tools, prompt=prompt, checkpointer=self.checkpointer)
self.graph = graph self.graph = graph
logger.info("ReActAgent 使用内存 checkpointer 构建成功") logger.info("ReActAgent 使用内存 checkpointer 构建成功")

View File

@ -83,6 +83,12 @@ class Message(Base):
"tool_calls": [tc.to_dict() for tc in self.tool_calls] if self.tool_calls else [], "tool_calls": [tc.to_dict() for tc in self.tool_calls] if self.tool_calls else [],
} }
def to_simple_dict(self):
return {
"role": self.role,
"content": self.content,
}
class ToolCall(Base): class ToolCall(Base):
"""ToolCall table - stores tool invocations""" """ToolCall table - stores tool invocations"""