From 606ad0135212898fe4fc79a57cdc6d6ca0c20793 Mon Sep 17 00:00:00 2001 From: liubiren Date: Wed, 9 Sep 2026 20:02:02 +0800 Subject: [PATCH] 1 --- agent/application/states/conversation.py | 79 ++++++++++++----------- agent/application/states/models.py | 14 +--- agent/application/workshop/book_flight.py | 27 +------- 3 files changed, 46 insertions(+), 74 deletions(-) diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index e7e93d2..b0e8b5e 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -4,6 +4,7 @@ """ from datetime import datetime from typing import AsyncGenerator, cast +from pydantic_ai import DeferredToolRequests from pydantic_ai.messages import ( FunctionToolCallEvent, @@ -20,12 +21,11 @@ from pydantic_ai.messages import ( ToolCallPart, ToolSearchCallPart, ) -from pydantic_ai.run import AgentRunResultEvent +from pydantic_ai.run import AgentRunResultEvent, AgentRunResult import reflex as rx from application.states.database import DatabaseState from application.states.models import ( - AgentRunOutputEvent, Conversation, ConversationHistoryItem, Message, @@ -33,6 +33,7 @@ from application.states.models import ( MessageType, Workflow, WorkflowType, + Approval, workflow_dump_json, workflow_validate_json, usage_validate_json, @@ -291,6 +292,11 @@ class ConversationState(rx.State): # 获取数据库状态 db_state = await self.get_db_state() + # 更新会话记录 + await db_state.update_conversation_record( + conversation.id, + workflow=conversation.workflow, + ) # 先创建消息记录再添加消息实例 conversation.messages.update( await db_state.create_message_record( @@ -461,45 +467,46 @@ class ConversationState(rx.State): yield # 通知前端更新渲染 messages.append(message) - # ========== 智能体运行输出事件 ========== - case AgentRunOutputEvent(content=content): - # 构建消息实例 - message = Message( - type=MessageType.TEXT, - content=content, - ) - # 添加至消息字典 - conversation.messages[message.id] = message - yield # 通知前端更新渲染 - messages.append(message) - # ========== 智能体运行结果事件 ========== case AgentRunResultEvent(result=result): - # 更新会话记录 - await db_state.update_conversation_record( - conversation.id, - usage=result.usage, - workflow=conversation.workflow, - ) - # 创建运行记录 - await db_state.create_run_record( + if isinstance(result.output, DeferredToolRequests): + for approval in result.output.approvals: + # 获取消息实例 + message = conversation.messages[ + tool_name_map_to_message_id[approval.tool_name] + ] + message.approval = Approval( + tool_call_id=approval.tool_call_id, + interaction_type=approval.interaction_type, + ) + messages.append(message) + else: + # 将工作流设置为空 + conversation.workflow = None + # 将正在运行设置为否 + conversation.is_running = False + + # 更新会话记录 + await db_state.update_conversation_record( + conversation.id, + usage=result.usage, + workflow=conversation.workflow, + ) + # 创建运行记录 + await db_state.create_run_record( + conversation_id=conversation.id, + new_messages=result.new_messages(), + ) + + self.conversations[self.actived_conversation_id] = conversation + yield + + # 批量创建消息记录 + await db_state.create_message_records( conversation_id=conversation.id, - new_messages=result.new_messages(), + messages=messages, ) - # 将正在运行设置为否 - conversation.is_running = False - # 将等待流式输出设置为否 - conversation.awaiting_stream = False - self.conversations[self.actived_conversation_id] = conversation - yield - - # 批量创建消息记录 - await db_state.create_message_records( - conversation_id=conversation.id, - messages=messages, - ) - @rx.event async def set_workflow(self, type: WorkflowType) -> None: """ diff --git a/agent/application/states/models.py b/agent/application/states/models.py index c21a930..28e782e 100644 --- a/agent/application/states/models.py +++ b/agent/application/states/models.py @@ -33,22 +33,9 @@ class Approval(BaseModel): """ tool_call_id: str = Field(..., description="工具调用唯一标识") - tool_name: str = Field(..., description="工具名称") interaction_type: InteractionType = Field(..., description="交互类型") -class AgentRunOutputEvent(BaseModel): - """ - 智能体运行输出事件类 - """ - - content: str = Field( - default="", - description="智能体运行内容", - ) - approvals: list[Approval] = Field(default_factory=list, description="审批列表") - - class MessageType(StrEnum): """ 消息类型枚举 @@ -69,6 +56,7 @@ class Message(BaseModel): type: MessageType = Field(..., description="消息类型") title: str = Field(default="", description="消息标题") content: str = Field(default="", description="消息内容") + approval: Approval | None = Field(default=None, description="审批") is_running: bool = Field( default=False, description="正在运行，True 表示正在运行， False 表示运行结束" ) diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index 571abae..fbc405f 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -27,7 +27,6 @@ from pydantic_ai.messages import ( from pydantic_ai.run import AgentRunResult, AgentRunResultEvent from application.states.models import ( - AgentRunOutputEvent, NoResult, InteractionType, Approval, @@ -284,7 +283,6 @@ async def run_stream_events( ) -> AsyncGenerator[ AgentStreamEvent | AgentRunResultEvent[Flight | NoResult | DeferredToolRequests] - | AgentRunOutputEvent | None, ]: """ @@ -336,26 +334,5 @@ async def run_stream_events( event.part.content = f"已预定航班：\n{content.number} {content.airfare}$ 于 {content.date.strftime('%Y-%m-%d')} 从 {content.origin_airport_code} 到 {content.destination_airport_code}\n" yield event - case AgentRunResultEvent( - result=result, - ): - match result: - case AgentRunResult(output=output): - match output: - case Flight() | NoResult(): - yield event - - case DeferredToolRequests(approvals=approvals): - yield AgentRunOutputEvent( - content="请输入航班号", - approvals=[ - Approval( - tool_call_id=approval.tool_call_id, - tool_name=approval.tool_name, - interaction_type=InteractionType.INPUT, - ) - for approval in approvals - if approval.tool_name == "预定航班" - ], - ) - yield event + case AgentRunResultEvent: + yield event