diff --git a/agent/application/agent/models.py b/agent/application/agent/models.py index 1c20db5..1a72be8 100644 --- a/agent/application/agent/models.py +++ b/agent/application/agent/models.py @@ -25,5 +25,3 @@ DEEPSEEK_V4_FLASH_RESPONSE_MODEL = OpenAIResponsesModel( api_key="sk-D9Y1mCe8VlvNqLuSC4mAjqEwxJ2nW4C0h8a7EPn8kg9RLsHq", ), ) - -"sk-4a79ec5c0af74680997ef56758604850" \ No newline at end of file diff --git a/agent/application/agent/runtime.py b/agent/application/agent/runtime.py index 5bf6fe7..9044cc5 100644 --- a/agent/application/agent/runtime.py +++ b/agent/application/agent/runtime.py @@ -2,7 +2,7 @@ """ 智能体运行时模块 """ -from asyncio import Queue, Task, Event, Lock +from asyncio import Queue, Task, Event, Lock, create_task from typing import AsyncGenerator, Optional @@ -11,145 +11,227 @@ from pydantic_ai.messages import AgentStreamEvent from pydantic_ai.run import AgentRunResultEvent from pydantic import BaseModel, Field from application.states.models import CapabilityComponentType -from pydantic_ai import DeferredToolRequests, DeferredToolResults, ToolApproved +from pydantic_ai import ( + DeferredToolRequests, + DeferredToolResults, + ToolApproved, + Agent, + ModelMessage, + ModelMessagesTypeAdapter, +) +from pydantic_ai.models.openai import OpenAIChatModel, OpenAIResponsesModel +from pydantic_ai.providers.openai import OpenAIProvider +from aiosqlite import connect, Row +from pathlib import Path + +# 构建数据库路径 +DATABASE_PATH = Path(__file__).parent.parent / "database.db" -class TaskRuntime(BaseModel): - queue: Queue[AgentStreamEvent | AgentRunResultEvent | None] = Field( - default_factory=lambda: Queue(maxsize=256), description="任务运行时队列" - ) - worker: Task | None = Field(default=None, description="任务运行时工作器") - done: Event = Field(default_factory=Event, description="任务运行时完成状态") - - -class TaskRuntimeRegistrar: +async def get_message_history(task_id: str) -> list[ModelMessage]: """ - 任务运行时注册器 + 获取消息历史 + """ + async with connect(DATABASE_PATH) as connection: + # 设置为行模式 + connection.row_factory = Row + cursor = await connection.execute( + "SELECT new_messages FROM message_history WHERE task_id = ? ORDER BY id ASC", + (task_id,), + ) + rows = await cursor.fetchall() + message_history: list[ModelMessage] = [] + for row in rows: + new_messages = ModelMessagesTypeAdapter.validate_json(row["new_messages"]) + message_history.extend(new_messages) + return message_history + + +class Runtime(BaseModel): + + task: Task | None = Field(default=None, description="运行时任务") + event: Event = Field(default_factory=Event, description="运行时通知信号") + queue: Queue[AgentStreamEvent | AgentRunResultEvent | None] = Field( + default_factory=lambda: Queue(maxsize=256), + description="运行时队列,默认最大长度为256", + ) + + +class Deps(BaseModel): + """ + 依赖类 + """ + + ... + + +instruction = """ +# 角色 +专业友好AI助手,结构化解答各类问题。 + +# 输出硬性规则 +1. 全文强制标准Markdown,禁止纯文本;不要额外说明排版格式,直接输出内容; +2. 层级使用 `#/##/###`,列表用 `-` 无序列表或数字有序列表; +3. 代码块用 ```语言名``` 包裹; +4. 重点内容标注 **粗体**/*斜体*; +5. 思考、工具日志仅输出文本,适配前端折叠面板,禁止输出HTML标签; +6. 内容分点拆分,排版整洁适配前端Markdown渲染。 + +# 行文要求 +语言通俗,逻辑完整简洁,无多余废话。 +""" + + +class RuntimeManager: + """ + 运行时管理器 """ def __init__(self): # 异步互斥锁 self.lock = Lock() - # 任务运行时字典,键为任务唯一标识,值为任务运行时实例 - self.task_runtimes: dict[str, TaskRuntime] = {} + # 运行时字典,键为任务唯一标识,值为运行时实例 + self.runtimes: dict[str, Runtime] = {} - async def get(self, task_id: str) -> TaskRuntime: + async def create( + self, + task_id: str, + capability_component_type: CapabilityComponentType, + user_prompt: str | None, + ) -> Runtime: """ - 获取任务运行时实例 + 创建运行时实例 :param task_id: 任务唯一标识 - :return: 任务运行时实例 + :param capability_component_type: 能力组件类型 + :param user_prompt: 用户提示词 + :return: 运行时实例 """ + + async def run( + capability_component_type: CapabilityComponentType, + user_prompt: str | None, + usage: dict, + queue: Queue[AgentStreamEvent | AgentRunResultEvent | None], + ): + """ + 运行 + """ + try: + # 根据能力组件的类型匹配流式输出方法 + match capability_component_type: + # 预定航班 + case CapabilityComponentType.BOOK_FLIGHT: + from application.agent.book_flight import ( + run_stream_events, + ) + + # 运行并流式输出事件 + stream_events = run_stream_events( + deps=deps, + user_prompt=user_prompt, + message_history=message_history, + usage=usage, + deferred_tool_results=deferred_tool_results, + ) + + # 银行客户服务 + case CapabilityComponentType.BANK_CUSTOMER_SERVICE: + from application.agent.bank_customer_service import ( + run_stream_events, + ) + from application.agent.database import connection_pool + + # 若连接池为空则直接返回 + if not connection_pool: + return + + # 运行并流式输出事件 + stream_events = run_stream_events( + deps=deps, + user_prompt=user_prompt, + message_history=message_history, + usage=usage, + ) + + # 联网检索 + case CapabilityComponentType.WEB_RETRIEVE: + from application.agent.web_retrieve import ( + run_stream_events, + ) + + # 运行并流式输出事件 + stream_events = run_stream_events( + deps=deps, + user_prompt=user_prompt, + message_history=message_history, + usage=usage, + ) + + # 知识库检索 + case CapabilityComponentType.KNOWLEDGE_BASE_RETRIEVE: + from application.agent.knowledge_base_retrieve import ( + run_stream_events, + ) + + # 运行并流式输出事件 + stream_events = run_stream_events( + deps=deps, + user_prompt=user_prompt, + message_history=message_history, + usage=usage, + ) + + # 兜底:闲聊 + case _: + ... + + # 初始化智能体 + agent = Agent( + model=OpenAIResponsesModel( + model_name="deepseek/deepseek-flash", + provider=OpenAIProvider( + base_url="https://tokenhub.tencentmaas.com/v1", + api_key="sk-D9Y1mCe8VlvNqLuSC4mAjqEwxJ2nW4C0h8a7EPn8kg9RLsHq", + ), + ), + instructions=instruction, + ) + async with agent.run_stream_events( + user_prompt=user_prompt, + message_history=await get_message_history(task_id), + usage=usage, + ) as events: + async for event in events: + # 将事件放进队列 + await queue.put(event) + # 哨兵:正常结束 + await queue.put(None) + + except Exception as exception: + await queue.put(exception) + async with self.lock: # 若任务唯一标识不存在则创建任务运行时实例 - if task_id not in self.task_runtimes: - self.task_runtimes[task_id] = TaskRuntime() - return self.task_runtimes[task_id] + if task_id not in self.runtimes: + self.runtimes[task_id] = (runtime := Runtime()) + runtime.task = create_task( + run( + capability_component_type=capability_component_type, + user_prompt=user_prompt, + message_history=message_history, + usage=usage, + queue=runtime.queue, + ) + ) + + return self.runtimes[task_id] async def cancel(self, task_id: str): """ - 取消任务运行时实例 + 取消运行时实例 :param task_id: 任务唯一标识 """ async with self.lock: - task_runtime = self.task_runtimes.pop(task_id, None) - if task_runtime and task_runtime.worker and not task_runtime.worker.done(): - task_runtime.worker.cancel() - - -async def worker( - task_id: str, - user_prompt: str | None, - message_history, - usage, - capability_component_type: CapabilityComponentType, - queue: Queue, - deps=None, - deferred_tool_results: DeferredToolResults | None = None, -): - """ - 工作器 - """ - try: - # 根据能力组件的类型匹配流式输出方法 - match capability_component_type: - case CapabilityComponentType.TALK: - from application.agent.talk import ( - run_stream_events, - ) - - # 运行并流式输出事件 - stream_events = run_stream_events( - user_prompt=user_prompt, - message_history=message_history, - usage=usage, - ) - - # 预定航班 - case CapabilityComponentType.BOOK_FLIGHT: - from application.agent.book_flight import ( - run_stream_events, - ) - - # 运行并流式输出事件 - stream_events = run_stream_events( - deps=deps, - user_prompt=user_prompt, - message_history=message_history, - usage=usage, - deferred_tool_results=deferred_tool_results, - ) - - # 银行客户服务 - case CapabilityComponentType.BANK_CUSTOMER_SERVICE: - from application.agent.bank_customer_service import ( - run_stream_events, - ) - from application.agent.database import connection_pool - - # 若连接池为空则直接返回 - if not connection_pool: - return - - # 运行并流式输出事件 - stream_events = run_stream_events( - deps=deps, - user_prompt=user_prompt, - message_history=message_history, - usage=usage, - ) - - # 联网检索 - case CapabilityComponentType.WEB_RETRIEVE: - from application.agent.web_retrieve import ( - run_stream_events, - ) - - # 运行并流式输出事件 - stream_events = run_stream_events( - deps=deps, - user_prompt=user_prompt, - message_history=message_history, - usage=usage, - ) - - # 知识库检索 - case CapabilityComponentType.KNOWLEDGE_BASE_RETRIEVE: - from application.agent.knowledge_base_retrieve import ( - run_stream_events, - ) - - # 运行并流式输出事件 - stream_events = run_stream_events( - deps=deps, - user_prompt=user_prompt, - message_history=message_history, - usage=usage, - ) - - async for event in stream_events: - await queue.put(event) - # 哨兵:正常结束 - await queue.put(None) - except Exception as e: - # 异常丢入队列给消费端处理 - await queue.put(e) + runtime = self.runtimes.pop(task_id, None) + if runtime and runtime.task and not runtime.task.done(): + runtime.task.cancel() diff --git a/agent/database.db b/agent/database.db index a4403dc..e948ab4 100644 Binary files a/agent/database.db and b/agent/database.db differ diff --git a/短视频自动合成_新/caches.db b/短视频自动合成_新/caches.db index a903f55..26d17a8 100644 Binary files a/短视频自动合成_新/caches.db and b/短视频自动合成_新/caches.db differ diff --git a/短视频自动合成_新/drafts_manager.py b/短视频自动合成_新/drafts_manager.py index 0fe4081..deea6d3 100644 --- a/短视频自动合成_新/drafts_manager.py +++ b/短视频自动合成_新/drafts_manager.py @@ -8,7 +8,6 @@ import hashlib import json from pathlib import Path from random import choice -import shutil import sys from pyJianYingDraft import DraftFolder, VideoMaterial @@ -86,7 +85,8 @@ DRAFT_GENERATE_SCHEMES = { { "clip_settings": [ { - "rotation": 90, + "scale_x": 1.0, + "scale_y": 1.0, }, ], # 剪辑设置 }, @@ -98,7 +98,8 @@ DRAFT_GENERATE_SCHEMES = { { "clip_settings": [ { - "rotation": 90, + "scale_x": 1.0, + "scale_y": 1.0, }, ], # 剪辑设置 }, @@ -126,13 +127,13 @@ class DraftsManager: # 项目名称 self.project_name = project_name + # 草稿生成方案 self.draft_generate_scheme: list[list[tuple[str, dict]]] = ( DRAFT_GENERATE_SCHEMES.get(self.project_name, []) ) if not self.draft_generate_scheme: raise RuntimeError(f"未就 {self.project_name} 配置草稿生成方案") - # 构建此项目素材仓库路径 self.material_vault_path = Path(rf"E:\jianying\materials\{project_name}") # 检查此项目素材仓库是否存在,若不存在则抛出异常 @@ -196,15 +197,10 @@ class DraftsManager: def batch_create( self, draft_counts: int, - video_width: int = 1080, - video_height: int = 1920, - video_fps: int = 30, ) -> None: """ 批量创建草稿 :param draft_counts: 草稿数 - :param video_width: 视频宽度(单位为像素),默认为 1080 - :param video_height: 视频高度(单位为像素),默认为 1920 :param video_fps: 视频帧率(单位为帧/秒),默认为 30 :return: 无 """ @@ -268,6 +264,11 @@ class DraftsManager: print(f"正在创建草稿 {draft_name}({index}/{draft_counts})...") + if "横版" in self.project_name: + video_width, video_height = 1920, 1080 + else: + video_width, video_height = 1080, 1920 + # 创建剪映草稿 draft = Drafts( materials_vault_path=self.material_vault_path, @@ -275,7 +276,7 @@ class DraftsManager: draft_name=draft_name, video_width=video_width, video_height=video_height, - video_fps=video_fps, + video_fps=30, # 视频帧率(单位为帧/秒),默认为 30 video_duration=video_duration, ) for segment_generate_scheme in draft_generate_scheme: