GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-07-07 15:51:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(agent): 支持图片消息处理

Browse Source
This commit is contained in:
jxxghp
2026-03-29 09:56:53 +08:00
parent e3fee39043
commit b349aa2693
6 changed files with 745 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
app/agent/__init__.py
View File

@@ -10,7 +10,7 @@ from langchain.agents.middleware import (
SummarizationMiddleware,
LLMToolSelectorMiddleware,
)
from langchain_core.messages import (
from langchain_core.messages import ( # noqa: F401
HumanMessage,
BaseMessage,
)
@@ -42,12 +42,12 @@ class MoviePilotAgent:
"""
def __init__(
self,
session_id: str,
user_id: str = None,
channel: str = None,
source: str = None,
username: str = None,
self,
session_id: str,
user_id: str = None,
channel: str = None,
source: str = None,
username: str = None,
):
self.session_id = session_id
self.user_id = user_id
@@ -92,10 +92,10 @@ class MoviePilotAgent:
if block.get("thought"):
continue
if block.get("type") in (
"thinking",
"reasoning_content",
"reasoning",
"thought",
"thinking",
"reasoning_content",
"reasoning",
"thought",
):
continue
if block.get("type") == "text":
@@ -174,20 +174,32 @@ class MoviePilotAgent:
logger.error(f"创建 Agent 失败: {e}")
raise e
async def process(self, message: str) -> str:
async def process(self, message: str, images: List[str] = None) -> str:
"""
处理用户消息，流式推理并返回 Agent 回复
"""
try:
logger.info(f"Agent推理: session_id={self.session_id}, input={message}")
logger.info(
f"Agent推理: session_id={self.session_id}, input={message}, images={len(images) if images else 0}"
)
# 获取历史消息
messages = memory_manager.get_agent_messages(
session_id=self.session_id, user_id=self.user_id
)
# 增加用户消息
messages.append(HumanMessage(content=message))
# 构建用户消息内容
if images:
from langchain_core.messages import HumanMessage
content = []
if message:
content.append({"type": "text", "text": message})
for img in images:
content.append({"type": "image_url", "image_url": {"url": img}})
messages.append(HumanMessage(content=content))
else:
messages.append(HumanMessage(content=message))
# 执行推理
await self._execute_agent(messages)
@@ -199,7 +211,7 @@ class MoviePilotAgent:
return error_message
async def _stream_agent_tokens(
self, agent, messages: dict, config: dict, on_token: Callable[[str], None]
self, agent, messages: dict, config: dict, on_token: Callable[[str], None]
):
"""
流式运行智能体，过滤工具调用token和思考内容，将模型生成的内容通过回调输出。
@@ -212,18 +224,18 @@ class MoviePilotAgent:
buffer = ""
async for chunk in agent.astream(
messages,
stream_mode="messages",
config=config,
subgraphs=False,
version="v2",
messages,
stream_mode="messages",
config=config,
subgraphs=False,
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["data"]
if (
token
and hasattr(token, "tool_call_chunks")
and not token.tool_call_chunks
token
and hasattr(token, "tool_call_chunks")
and not token.tool_call_chunks
):
# 跳过模型思考/推理内容（如 DeepSeek R1 的 reasoning_content）
additional = getattr(token, "additional_kwargs", None)
@@ -241,7 +253,7 @@ class MoviePilotAgent:
if start_idx > 0:
on_token(buffer[:start_idx])
in_think_tag = True
buffer = buffer[start_idx + 7:]
buffer = buffer[start_idx + 7 :]
else:
# 检查是否以 <think> 的前缀结尾
partial_match = False
@@ -259,7 +271,7 @@ class MoviePilotAgent:
end_idx = buffer.find("</think>")
if end_idx != -1:
in_think_tag = False
buffer = buffer[end_idx + 8:]
buffer = buffer[end_idx + 8 :]
else:
# 检查是否以 </think> 的前缀结尾
partial_match = False
@@ -421,6 +433,7 @@ class _MessageTask:
session_id: str
user_id: str
message: str
images: Optional[List[str]] = None
channel: Optional[str] = None
source: Optional[str] = None
username: Optional[str] = None
@@ -467,13 +480,14 @@ class AgentManager:
self.active_agents.clear()
async def process_message(
self,
session_id: str,
user_id: str,
message: str,
channel: str = None,
source: str = None,
username: str = None,
self,
session_id: str,
user_id: str,
message: str,
images: List[str] = None,
channel: str = None,
source: str = None,
username: str = None,
) -> str:
"""
处理用户消息：将消息放入会话队列，按顺序依次处理。
@@ -483,6 +497,7 @@ class AgentManager:
session_id=session_id,
user_id=user_id,
message=message,
images=images,
channel=channel,
source=source,
username=username,
@@ -497,8 +512,8 @@ class AgentManager:
# 如果队列中已有等待的消息，通知用户消息已排队
if queue_size > 0 or (
session_id in self._session_workers
and not self._session_workers[session_id].done()
session_id in self._session_workers
and not self._session_workers[session_id].done()
):
logger.info(
f"会话 {session_id} 有任务正在处理，消息已排队等待 "
@@ -510,8 +525,8 @@ class AgentManager:
# 确保该会话有一个worker在运行
if (
session_id not in self._session_workers
or self._session_workers[session_id].done()
session_id not in self._session_workers
or self._session_workers[session_id].done()
):
self._session_workers[session_id] = asyncio.create_task(
self._session_worker(session_id)
@@ -552,8 +567,8 @@ class AgentManager:
self._session_workers.pop(session_id, None) # noqa
# 如果队列为空，清理队列
if (
session_id in self._session_queues
and self._session_queues[session_id].empty()
session_id in self._session_queues
and self._session_queues[session_id].empty()
):
self._session_queues.pop(session_id, None)
@@ -584,7 +599,7 @@ class AgentManager:
if task.username:
agent.username = task.username
return await agent.process(task.message)
return await agent.process(task.message, images=task.images)
async def clear_session(self, session_id: str, user_id: str):
"""