50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""聊天 Agent 共享工具：历史获取、格式化、存储"""
|
|
from datetime import datetime
|
|
from typing import Any, List
|
|
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from app.core.redis import redis_service
|
|
|
|
|
|
async def get_history_messages(conversation_id: str) -> List[Any]:
|
|
"""从 Redis 获取对话历史"""
|
|
history = await redis_service.get_conversation_history(conversation_id)
|
|
messages = []
|
|
for msg in history:
|
|
if msg["role"] == "human":
|
|
messages.append(HumanMessage(content=msg["content"]))
|
|
elif msg["role"] == "ai":
|
|
messages.append(AIMessage(content=msg["content"]))
|
|
return messages
|
|
|
|
|
|
def format_history_string(messages: List[Any]) -> str:
|
|
"""将消息列表格式化为 Human/Assistant 字符串"""
|
|
history_parts = []
|
|
for msg in messages:
|
|
if isinstance(msg, HumanMessage):
|
|
history_parts.append(f"Human: {msg.content}")
|
|
elif isinstance(msg, AIMessage):
|
|
history_parts.append(f"Assistant: {msg.content}")
|
|
return "\n\n".join(history_parts)
|
|
|
|
|
|
async def save_message(
|
|
conversation_id: str,
|
|
role: str,
|
|
content: str,
|
|
message_type: str = "text",
|
|
voice_session_id: str | None = None,
|
|
timestamp: datetime | str | int | None = None,
|
|
) -> None:
|
|
"""保存消息到 Redis"""
|
|
await redis_service.add_message(
|
|
conversation_id,
|
|
role,
|
|
content,
|
|
message_type=message_type,
|
|
voice_session_id=voice_session_id,
|
|
timestamp=timestamp.isoformat() if isinstance(timestamp, datetime) else timestamp,
|
|
)
|