diff --git a/agent/application/pages/conversation.py b/agent/application/pages/conversation.py index 5cd1cd0..a19e48a 100644 --- a/agent/application/pages/conversation.py +++ b/agent/application/pages/conversation.py @@ -25,7 +25,7 @@ def conversation_history_item( # 高亮 is_highlight = ( item.id == ConversationState.popover_shown_conversation_history_item_id - ) | (item.id == ConversationState.actived_conversation_id) + ) | (item.id == ConversationState.conversation_id) return rx.list.item( rx.vstack( @@ -63,7 +63,7 @@ def conversation_history_item( margin_bottom="8px", ), # 点击事件：设置激活会话 - on_click=lambda: ConversationState.set_actived_conversation(item.id), + on_click=lambda: ConversationState.set_conversation(item.id), width="100%", ), # 会话历史项的更多按钮 @@ -461,13 +461,12 @@ def tool_call(item: MessageHistoryItem) -> rx.Component: item.has_approval, rx.vstack( rx.text_area( - # 值绑定审批用户提示词 - value=item.approval_user_prompt, - # 值变化事件：设置审批用户提示词 - on_change=lambda user_prompt: ConversationState.set_approval_user_prompt( - item.id, user_prompt + # 值绑定审批内容 + value=item.approval_content, + # 值变化事件：设置审批内容 + on_change=lambda content: ConversationState.set_approval_content( + item.id, content ), - disabled=item.is_approval_user_prompt_input_disabled, placeholder="请输入", vertical_align="middle", width="100%", @@ -490,30 +489,16 @@ def tool_call(item: MessageHistoryItem) -> rx.Component: ), rx.hstack( rx.button( - rx.cond( - item.is_approval_user_prompt_input_disabled, - "已提交", - "提交", - ), + "提交", # 点击事件：审批 - on_click=lambda: ConversationState.approve(item.id), - disabled=item.is_approval_user_prompt_input_disabled, + on_click=lambda: ConversationState.run(True, item.id), padding="0 12px", - # 是否禁用绑定审批用户提示词不可输出状态 - background_color=rx.cond( - item.is_approval_user_prompt_input_disabled, - "var(--prismui-background-color-8)", - "var(--prismui-background-color-7)", - ), + background_color="var(--prismui-background-color-7)", border_radius="var(--prismui-border-radius-3)", line_height="18px", font_size="var(--prismui-font-size-1)", color="var(--prismui-color-9)", - cursor=rx.cond( - item.is_approval_user_prompt_input_disabled, - "not-allowed", - "pointer", - ), + cursor="pointer", ), justify_content="flex-end", width="100%", diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index 2da1660..f6411bc 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -32,7 +32,7 @@ from application.states.models import ( Workflow, WorkflowType, ) -from application.workshop.events import AgentPromptEvent +from application.workshop.events import AgentRuntimeEvent def format_conversation_created_at(created_at: datetime) -> str: @@ -61,9 +61,9 @@ class ConversationState(rx.State): # 当前用户的会话字典（私有变量） # 当前用户的会话字典 - conversations: dict[str, Conversation] = {} # 键为会话唯一标识，值为会话实例 - # 激活的会话唯一标识 - actived_conversation_id: str = "" + conversations: dict[str, Conversation] = {} # 键为会话唯一标识，值为会话 + # 当前会话唯一标识 + conversation_id: str = "" # 显示会话历史，True 表示显示，False 表示隐藏 is_conversation_history_shown: bool = False @@ -98,14 +98,25 @@ class ConversationState(rx.State): db_state = await self.get_db_state() # 查询当前用户的会话字典 self.conversations = await db_state.read_conversations(self.user_id) - # 若当前用户的会话字典为空则先创建会话记录再添加会话实例 + # 若当前用户的会话字典为空则先创建会话记录再添加会话 if not self.conversations: self.conversations.update( await db_state.create_conversation_record(self.user_id) ) - # 将最后一个会话的唯一标识作为激活会话唯一标识 - self.actived_conversation_id = next(reversed(self.conversations.keys())) - print(self.conversations[self.actived_conversation_id]) + # 将最后一个会话的唯一标识作为当前会话唯一标识 + self.conversation_id = next(reversed(self.conversations.keys())) + + # 当前会话 + conversation = self.conversations[self.conversation_id] + # 最后一个消息 + message = next(reversed(conversation.messages.values())) + # 若最后一个消息包含审批且审批结果为 None 则将会话正在运行设置为是、消息正在运行设置为是 + if ( + isinstance(approval := message.approval, Approval) + and approval.result is None + ): + conversation.is_running = True + message.is_running = True @rx.event def toggle_conversation_history_shown(self) -> None: @@ -159,18 +170,18 @@ class ConversationState(rx.State): await db_state.create_conversation_record(self.user_id) ) - # 删除后，若激活会话唯一标识不存在则将最后一个会话的唯一标识作为激活会话唯一标识 - if self.actived_conversation_id not in self.conversations: - self.actived_conversation_id = next(reversed(self.conversations.keys())) + # 删除后，若当前会话唯一标识不存在则将最后一个会话的唯一标识作为当前会话唯一标识 + if self.conversation_id not in self.conversations: + self.conversation_id = next(reversed(self.conversations.keys())) @rx.event - def set_actived_conversation(self, conversation_id: str) -> None: + def set_conversation(self, conversation_id: str) -> None: """ - 设置激活会话 + 设置当前会话 :param conversation_id: 会话唯一标识 :return: None """ - self.actived_conversation_id = conversation_id + self.conversation_id = conversation_id @rx.var def message_history(self) -> list[MessageHistoryItem]: @@ -179,7 +190,7 @@ class ConversationState(rx.State): :return: 当前会话的消息历史 """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return [] message_history: list[MessageHistoryItem] = [] @@ -187,13 +198,9 @@ class ConversationState(rx.State): message_history.append( MessageHistoryItem( **message.model_dump(), - has_approval=isinstance(approval := message.approval, Approval), - approval_user_prompt=getattr(approval, "user_prompt", ""), - is_approval_user_prompt_input_disabled=( - False - if (isinstance(approval, Approval) and approval.result is None) - else True - ), + has_approval=isinstance(approval := message.approval, Approval) + and approval.result is None, + approval_content=getattr(approval, "content", ""), ) ) return message_history @@ -206,12 +213,12 @@ class ConversationState(rx.State): """ # 获取当前数据库状态 db_state = await self.get_db_state() - # 创建会话记录再添加会话实例 + # 创建会话记录再添加会话 self.conversations.update( await db_state.create_conversation_record(self.user_id) ) - # 将最后一个会话的唯一标识作为激活会话唯一标识 - self.actived_conversation_id = next(reversed(self.conversations.keys())) + # 将最后一个会话的唯一标识作为当前会话唯一标识 + self.conversation_id = next(reversed(self.conversations.keys())) @rx.var def user_prompt(self) -> str: @@ -220,7 +227,7 @@ class ConversationState(rx.State): :return: 用户提示词 """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return "" return conversation.user_prompt @@ -233,9 +240,13 @@ class ConversationState(rx.State): :return: None """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return + + if not user_prompt.strip(): + return + conversation.user_prompt = user_prompt.strip() @rx.var @@ -245,7 +256,7 @@ class ConversationState(rx.State): :return: bool """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return True return conversation.is_running or not conversation.user_prompt @@ -253,27 +264,53 @@ class ConversationState(rx.State): @rx.event async def run( self, - approval: Approval | None = None, - deferred_tool_results: DeferredToolResults | None = None, + is_resume_run: bool = False, + message_id: str | None = None, ) -> AsyncGenerator[None]: """ 运行 - :param approval: 审批实例 - :param deferred_tool_results: 延迟工具调用结果 + :param is_resume_run: 恢复运行 + :param message_id: 消息唯一标识 :return: AsyncGenerator """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return - # 获取数据库状态 + # 获取当前数据库状态 db_state = await self.get_db_state() - if isinstance(approval, Approval): - if not (user_prompt := approval.user_prompt): + user_prompt = None # 用户提示词 + deferred_tool_results = None # 延迟工具结果 + # 若为恢复运行 + if is_resume_run: + if not message_id: return + # 指定消息唯一标识的消息 + message = conversation.messages.get(message_id) + if ( + not message + or not isinstance(approval := message.approval, Approval) + or not approval.content + ): + return + + # 将等待流式输出设置为是 + conversation.awaiting_stream = True + # 更新消息内容 + message.content = f"您输入的是：\n\n{approval.content}\n\n" + # 更新审批结果 + approval.result = True + yield # 通知前端更新渲染 + + # 构建延迟工具结果 + deferred_tool_results = DeferredToolResults( + approvals={approval.tool_call_id: ToolApproved()}, # 审批通过 + metadata={approval.tool_call_id: {"content": approval.content}}, + ) + else: # 若用户提示词为空则直接返回 if not (user_prompt := conversation.user_prompt): @@ -285,7 +322,7 @@ class ConversationState(rx.State): conversation.is_running = True # 将等待流式输出设置为是 conversation.awaiting_stream = True - # 先创建消息记录再添加消息实例 + # 先创建消息记录再添加消息 conversation.messages.update( await db_state.create_message_record( conversation_id=conversation.id, @@ -297,13 +334,15 @@ class ConversationState(rx.State): ) yield # 通知前端更新渲染 - # 初始化片段索引映射为消息实例唯一标识字典 + # 初始化片段索引映射为消息唯一标识字典 index_map_to_message_id: dict[int, str] = {} # 查询消息历史列表 message_history = await db_state.read_message_history( - conversation_id=self.actived_conversation_id + conversation_id=self.conversation_id ) + # 使用量 + usage = conversation.usage # 初始化工作流 match conversation.workflow.type: @@ -316,7 +355,7 @@ class ConversationState(rx.State): stream_events = run_stream_events( user_prompt=user_prompt, message_history=message_history, - usage=conversation.usage, + usage=usage, ) case WorkflowType.BOOK_FLIGHT: from application.workshop.book_flight import ( @@ -328,7 +367,7 @@ class ConversationState(rx.State): deps=conversation.workflow.deps, user_prompt=user_prompt, message_history=message_history, - usage=conversation.usage, + usage=usage, deferred_tool_results=deferred_tool_results, ) @@ -348,7 +387,7 @@ class ConversationState(rx.State): match part: # 思考分片 case ThinkingPart(content=content): - # 构建消息实例 + # 构建消息 message = Message( type=MessageType.THINKING, title="正在思考", @@ -357,7 +396,7 @@ class ConversationState(rx.State): ) # 添加至消息字典 conversation.messages[message.id] = message - # 将消息实例唯一标识与片段索引映射 + # 将消息唯一标识与片段索引映射 index_map_to_message_id[index] = message.id yield # 通知前端更新渲染 @@ -369,7 +408,7 @@ class ConversationState(rx.State): tool_call_id not in conversation.workflow.tool_call_id_map_to_message_id ): - # 构建消息实例 + # 构建消息 message = Message( type=MessageType.TOOL_CALL, title=f"正在{tool_name}", @@ -377,7 +416,7 @@ class ConversationState(rx.State): ) # 添加至消息字典 conversation.messages[message.id] = message - # 将消息实例唯一标识与工具调用唯一标识映射 + # 将消息唯一标识与工具调用唯一标识映射 conversation.workflow.tool_call_id_map_to_message_id[ tool_call_id ] = message.id @@ -385,11 +424,11 @@ class ConversationState(rx.State): # 文本分片 case TextPart(content=content): - # 构建消息实例 + # 构建消息 message = Message(type=MessageType.TEXT, content=content) # 添加至消息字典 conversation.messages[message.id] = message - # 将消息实例唯一标识与片段索引映射 + # 将消息唯一标识与片段索引映射 index_map_to_message_id[index] = message.id yield # 通知前端更新渲染 @@ -425,7 +464,7 @@ class ConversationState(rx.State): match part: # 思考分片 case ThinkingPart(): - # 获取消息实例 + # 获取消息 message = conversation.messages[ index_map_to_message_id[index] ] @@ -436,7 +475,7 @@ class ConversationState(rx.State): # 文本分片 case TextPart(): - # 获取消息实例 + # 获取消息 message = conversation.messages[ index_map_to_message_id[index] ] @@ -451,7 +490,7 @@ class ConversationState(rx.State): tool_name=tool_name, content=content, ): - # 获取消息实例 + # 获取消息 message = conversation.messages[ conversation.workflow.tool_call_id_map_to_message_id[ tool_call_id @@ -464,20 +503,30 @@ class ConversationState(rx.State): yield # 通知前端更新渲染 messages.append(message) - # ========== 智能体提示事件 ========== - case AgentPromptEvent(content=content): - # 构建消息实例 - message = Message(type=MessageType.TEXT, content=content) - # 添加至消息字典 - conversation.messages[message.id] = message - yield # 通知前端更新渲染 - messages.append(message) + # ========== 智能体运行时事件 ========== + case AgentRuntimeEvent(tool_call_id=tool_call_id, content=content): + # 若工具调用唯一标识为空则创建消息并添加至消息字典，否则若消息内容非空则根据工具调用唯一标识获取消息并更新消息内容 + if not tool_call_id: + message = Message(type=MessageType.TEXT, content=content) + conversation.messages[message.id] = message + yield # 通知前端更新渲染 + messages.append(message) + + else: + if content: + message = conversation.messages[ + conversation.workflow.tool_call_id_map_to_message_id[ + tool_call_id + ] + ] + message.content = content + yield # 通知前端更新渲染 # ========== 智能体运行结果事件 ========== case AgentRunResultEvent(result=result): if isinstance(output := result.output, DeferredToolRequests): for appr in output.approvals: - # 获取消息实例 + # 获取消息 message = conversation.messages[ conversation.workflow.tool_call_id_map_to_message_id[ tool_call_id := appr.tool_call_id @@ -524,7 +573,7 @@ class ConversationState(rx.State): :return: 当前会话正在运行（True 表示正在运行，False 表示运行完成） """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return False return conversation.is_running @@ -536,7 +585,7 @@ class ConversationState(rx.State): :return: 当前会话正在等待流式输出（True 表示等待流式输出，False 表示已开始流式输出或已完成） """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return False return conversation.awaiting_stream @@ -547,59 +596,32 @@ class ConversationState(rx.State): 展示 / 隐藏消息历史项 """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return conversation.messages[item_id].is_shown ^= True @rx.event - def set_approval_user_prompt(self, item_id: str, user_prompt: str) -> None: + def set_approval_content(self, item_id: str, content: str) -> None: """ - 设置审批用户提示词 + 设置审批内容 :param item_id: 消息历史项唯一标识 - :param user_prompt: 用户提示词 + :param content: 内容 """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return - # 获取消息实例 + + # 获取消息 message = conversation.messages.get(item_id) - if not message or not isinstance(message.approval, Approval): - return - message.approval.user_prompt = user_prompt - - @rx.event - async def approve(self, item_id: str) -> AsyncGenerator[None, None]: - """ - 审批消息历史项 - :param item_id: 消息历史项唯一标识 - """ - # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) - if not conversation: + if not message or not isinstance(approval := message.approval, Approval): return - if not (message := conversation.messages.get(item_id)) or not isinstance( - approval := message.approval, Approval - ): + if not content.strip(): return - # 若会话未正在运行则将正在运行设置为是 - if not conversation.is_running: - conversation.is_running = True - # 将等待流式输出设置为是 - conversation.awaiting_stream = True - # 更新审批结果为审批通过 - approval.result = True - yield # 通知前端更新渲染 - - # 运行 - async for _ in self.run( - approval, - DeferredToolResults(approvals={approval.tool_call_id: ToolApproved()}), - ): - ... + approval.content = content @rx.event async def set_workflow(self, type: WorkflowType) -> None: @@ -607,7 +629,7 @@ class ConversationState(rx.State): 设置工作流 """ # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) + conversation = self.conversations.get(self.conversation_id) if not conversation: return diff --git a/agent/application/states/database.py b/agent/application/states/database.py index d8460de..78756b8 100644 --- a/agent/application/states/database.py +++ b/agent/application/states/database.py @@ -22,6 +22,7 @@ from application.states.models import ( usage_validate_json, workflow_dump_json, workflow_validate_json, + Approval, ) @@ -368,6 +369,25 @@ class DatabaseState(rx.State): ) await session.commit() + + async def update_message_record( + self, + message_id: str, + approval: Approval | None = None, + ) -> None: + """ + 更新消息记录 + :param message_id: 消息唯一标识 + :param approval: 审批实例 + """ + async with rx.asession() as session: + record = await session.get(MessageRecord, message_id) + if not record: + return + if isinstance(approval, Approval): + record.approval = approval_dump_json(approval) + await session.commit() + async def read_message_history(self, conversation_id: str) -> list[ModelMessage]: """ 查询消息历史列表 diff --git a/agent/application/states/models.py b/agent/application/states/models.py index ea9ced2..bd8857b 100644 --- a/agent/application/states/models.py +++ b/agent/application/states/models.py @@ -11,14 +11,6 @@ from pydantic_ai import RunUsage from pydantic_ai._uuid import uuid7 -class NoResult(BaseModel): - """ - 无结果类 - """ - - ... - - class MessageType(StrEnum): """ 消息类型枚举 @@ -36,7 +28,7 @@ class Approval(BaseModel): """ tool_call_id: str = Field(..., description="工具调用唯一标识") - user_prompt: str = Field(default="", description="用户提示词") + content: str = Field(default="", description="审批内容") result: bool | None = Field( default=None, description="审批结果，True 表示审批通过，False 表示审批拒绝，None 表示待审批", @@ -119,10 +111,7 @@ class MessageHistoryItem(Message): """ has_approval: bool = Field(default=False, description="包含审批实例") - approval_user_prompt: str = Field(default="", description="审批用户提示词") - is_approval_user_prompt_input_disabled: bool = Field( - default=False, description="审批用户提示词不可输出状态" - ) + approval_content: str = Field(default="", description="审批内容") # 工作流适配器 diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index 5d3a4d6..cc1c57a 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -27,7 +27,7 @@ from pydantic_ai.messages import ( ) from pydantic_ai.run import AgentRunResultEvent -from application.workshop.events import AgentPromptEvent +from application.workshop.events import AgentRuntimeEvent from application.workshop.models import ( DEEPSEEK_V4_FLASH_MODEL, DEEPSEEK_V4_FLASH_MODEL_SETTINGS, @@ -172,7 +172,7 @@ async def search_flights(ctx: RunContext[Deps]) -> list[Flight]: **设计工具需先设计业务流程** 本示例按照 查询航班 -> 推荐航班 -> 预定航班 """ - if ctx.deps.searched_flights: + if ctx.deps.searched_flights is not None: return ctx.deps.searched_flights # 提取到的所有航班信息 @@ -216,13 +216,14 @@ async def book_flight(ctx: RunContext[Deps]) -> Flight: if (searched_flights := ctx.deps.searched_flights) is None: raise ModelRetry("必须先使用 search_flights 查询航班") + # 若工具调用未审批则抛出需要审批异常，否则解析审批内容并作为用户输入航班号 if not ctx.tool_call_approved: raise ApprovalRequired(metadata={"content": "请输入航班号：\n\n"}) # 提取到的航班号 extracted_flight_number = ( await extraction_flight_number_agent.run( - f"用户提示词：\n{cast(str, ctx.prompt).upper()}\n查询到的航班：\n{(flight_numbers:= [flight.number for flight in searched_flights])}", + f"用户提示词：\n{cast(str, ctx.tool_call_metadata["content"]).upper()}\n查询到的航班：\n{(flight_numbers:= [flight.number for flight in searched_flights])}", deps=ctx.deps, usage=ctx.usage, usage_limits=UsageLimits(request_limit=10), @@ -286,7 +287,7 @@ deps = Deps( async def run_stream_events( deps: Any, - user_prompt: str, + user_prompt: str | None, message_history: list[ModelMessage], usage: RunUsage, usage_limits: UsageLimits | None = None, @@ -323,6 +324,7 @@ async def run_stream_events( case FunctionToolResultEvent(part): match part: case ToolReturnPart( + tool_call_id=tool_call_id, tool_name=tool_name, content=content, ): @@ -330,28 +332,31 @@ async def run_stream_events( case "查询航班": # 查询到的航班 searched_flights = cast(list[Flight], content) - event.part.content = f"已查询到 **{len(searched_flights)}** 班航班：\n\n" + # 构造智能体运行时事件 + agent_runtime_event = AgentRuntimeEvent( + tool_call_id=tool_call_id, + content=f"已查询到 **{len(searched_flights)}** 班航班：\n\n", + ) for searched_flight in searched_flights: - event.part.content += f"**{searched_flight.number}** | 💰 {searched_flight.airfare} | {searched_flight.date.strftime('%Y-%m-%d')} | {searched_flight.origin_airport_code} ✈️ {searched_flight.destination_airport_code}\n\n" - yield event + agent_runtime_event.content += f"**{searched_flight.number}** | 💰 {searched_flight.airfare} | {searched_flight.date.strftime('%Y-%m-%d')} | {searched_flight.origin_airport_code} ✈️ {searched_flight.destination_airport_code}\n\n" + + yield agent_runtime_event case "预定航班": - if isinstance(deps.booked_flight, Flight): - event.part.content = ( - "请输入航班号：\n\n" # 成功预定内容维持提示 - ) - else: - event.part.content = ( - "未提取到航班号，请检查后重试\n\n" - ) - yield event + # 构造智能体运行时事件 + agent_runtime_event = AgentRuntimeEvent( + tool_call_id=tool_call_id, + ) + yield agent_runtime_event + + yield event # 智能体运行时依赖 event，原则上不可修改 case AgentRunResultEvent(result=result): match result: case AgentRunResult(output=output): match output: case Flight(): - yield AgentPromptEvent( + yield AgentRuntimeEvent( content=f"---\n\n✅ 预定机票成功：\n\n{output.number} | 💰 {output.airfare} | {output.date.strftime('%Y-%m-%d')} | {output.origin_airport_code} ✈️ {output.destination_airport_code}\n\n本次会话结束。" ) yield event diff --git a/agent/application/workshop/events.py b/agent/application/workshop/events.py index 5d70436..e3b5e57 100644 --- a/agent/application/workshop/events.py +++ b/agent/application/workshop/events.py @@ -8,9 +8,9 @@ from pydantic_ai.messages import CustomEvent @dataclass(kw_only=True) -class AgentPromptEvent(CustomEvent): +class AgentRuntimeEvent(CustomEvent): """ - 智能体提示事件 + 智能体运行时事件 """ - content: str + content: str = "" diff --git a/agent/application/workshop/talk.py b/agent/application/workshop/talk.py index 0352ee1..0637ac6 100644 --- a/agent/application/workshop/talk.py +++ b/agent/application/workshop/talk.py @@ -34,7 +34,7 @@ agent = Agent( async def run_stream_events( - user_prompt: str, + user_prompt: str | None, message_history: list[ModelMessage], usage: RunUsage, ) -> AsyncGenerator[AgentStreamEvent | AgentRunResultEvent, None]: diff --git a/agent/database.db b/agent/database.db index 94864f0..18d31a9 100644 Binary files a/agent/database.db and b/agent/database.db differ