From 31828a7be3de55a85086f9cbe2ffd8ff9bf59295 Mon Sep 17 00:00:00 2001 From: liubiren Date: Thu, 10 Sep 2026 20:54:14 +0800 Subject: [PATCH] 1 --- .vscode/settings.json | 3 +- agent/application/pages/conversation.py | 56 ++++- agent/application/states/conversation.py | 255 ++++++++++++++-------- agent/application/states/database.py | 67 +++--- agent/application/states/models.py | 57 +++-- agent/application/workshop/book_flight.py | 6 +- 6 files changed, 305 insertions(+), 139 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index 7b45f78..18802f9 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,4 +1,5 @@ { "python.languageServer": "None", - "python.analysis.extraPaths": ["${workspaceFolder}"] + "python.analysis.extraPaths": ["${workspaceFolder}"], + "python.REPL.enableREPLSmartSend": false } diff --git a/agent/application/pages/conversation.py b/agent/application/pages/conversation.py index 535ca45..c3a1f59 100644 --- a/agent/application/pages/conversation.py +++ b/agent/application/pages/conversation.py @@ -24,7 +24,7 @@ def conversation_history_item( # 高亮 is_highlight = ( - item.id == ConversationState.is_conversation_history_item_popover_shown + item.id == ConversationState.popover_shown_conversation_history_item_id ) | (item.id == ConversationState.actived_conversation_id) return rx.list.item( @@ -425,7 +425,7 @@ def tool_call(item: MessageHistoryItem) -> rx.Component: :return: Component """ return rx.vstack( - # 工具调用的标题栏 + # 标题栏 rx.hstack( rx.icon( "square-function", @@ -444,7 +444,7 @@ def tool_call(item: MessageHistoryItem) -> rx.Component: margin_bottom="8px", line_height="22px", ), - # 工具调用的内容 + # 内容 rx.markdown( item.content, width="100%", @@ -453,6 +453,54 @@ def tool_call(item: MessageHistoryItem) -> rx.Component: font_size="var(--prismui-font-size-1)", color="var(--prismui-color-3)", ), + # 审批 + rx.cond( + item.has_approval, + rx.hstack( + rx.input( + placeholder="请输入", + # 值绑定审批用户提示词 + value=item.approval_user_prompt, + # 值变化事件：设置审批用户提示词 + on_change=lambda user_prompt: ConversationState.set_approval_user_prompt( + item.id, user_prompt + ), + flex=1, + height="100%", + padding="4px", + border_radius="var(--prismui-border-radius-4)", + line_height="16px", + border="none !important", + ), + rx.button( + "发送", + # 点击事件：审批 + on_click=lambda: ConversationState.approve(item.id), + # 是否禁用绑定验证码发送禁用状态 + disabled=AuthState.is_captcha_code_sending_button_disabled, + margin="0 8px", + background_color="var(--prismui-background-color-1)", + line_height="16px", + # 若验证码发送禁用状态为是则颜色设置为 color-4，否则设置为 color-5（主题色） + color=rx.cond( + AuthState.is_captcha_code_sending_button_disabled, + "var(--prismui-color-4)", + "var(--prismui-color-5)", + ), + cursor=rx.cond( + AuthState.is_captcha_code_sending_button_disabled, + "not-allowed", + "pointer", + ), + ), + width="100%", + height="45px", + align_items="center", + border="var(--prismui-border-1)", + border_radius="var(--prismui-border-radius-4)", + ), + rx.fragment(), + ), ) @@ -772,7 +820,7 @@ def user_prompt_sending() -> rx.Component: # 不可点击状态：绑定用户提示词发送按钮不可点击状态 disabled=ConversationState.is_user_prompt_sending_button_disabled, # 点击事件：运行 - on_click=ConversationState.run, + on_click=lambda: ConversationState.run(), display="inline-flex", position="relative", justify_content="center", diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index b0e8b5e..1bd3e41 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -4,7 +4,12 @@ """ from datetime import datetime from typing import AsyncGenerator, cast -from pydantic_ai import DeferredToolRequests +from pydantic_ai import ( + DeferredToolRequests, + DeferredToolResults, + ToolApproved, + ToolDenied, +) from pydantic_ai.messages import ( FunctionToolCallEvent, @@ -67,13 +72,13 @@ class ConversationState(rx.State): # 当前用户的会话字典 conversations: dict[str, Conversation] = {} # 键为会话唯一标识，值为会话实例 - # 激活会话唯一标识 + # 激活的会话唯一标识 actived_conversation_id: str = "" # 显示会话历史，True 表示显示，False 表示隐藏 is_conversation_history_shown: bool = False - # 会话历史项显示悬停气泡 - is_conversation_history_item_popover_shown: str = "" + # 悬停气泡显示的会话历史项唯一标识（一个会话只允许存在一个悬停气泡会话历史项） + popover_shown_conversation_history_item_id: str = "" # 当前数据库状态（私有变量） # 私有变量：reflex 约定以 _ 开头的变量为私有变量，后端不序列化，前端不可使用 @@ -101,8 +106,8 @@ class ConversationState(rx.State): # 获取当前数据库状态 db_state = await self.get_db_state() - # 获取当前用户的会话字典 - self.conversations = await db_state.get_conversations(self.user_id) + # 查询当前用户的会话字典 + self.conversations = await db_state.read_conversations(self.user_id) # 若当前用户的会话字典为空则先创建会话记录再添加会话实例 if not self.conversations: self.conversations.update( @@ -136,15 +141,13 @@ class ConversationState(rx.State): @rx.event def set_conversation_history_item_popover_shown( - self, conversation_id: str, is_shown: bool + self, item_id: str, is_shown: bool ) -> None: """ 设置会话历史项悬停气泡显示 / 隐藏 :return: None """ - self.is_conversation_history_item_popover_shown = ( - conversation_id if is_shown else "" - ) + self.popover_shown_conversation_history_item_id = item_id if is_shown else "" @rx.event async def delete_conversation(self, conversation_id: str) -> None: @@ -188,10 +191,22 @@ class ConversationState(rx.State): conversation = self.conversations.get(self.actived_conversation_id) if not conversation: return [] - return [ - MessageHistoryItem(**message.model_dump()) - for message in list(conversation.messages.values()) - ] + message_history: list[MessageHistoryItem] = [] + has_approval = False + approval_user_prompt = "" + for message in conversation.messages.values(): + approval = message.approval + if isinstance(approval, Approval): + has_approval = True + approval_user_prompt = approval.user_prompt + message_history.append( + MessageHistoryItem( + **message.model_dump(), + has_approval=has_approval, + approval_user_prompt=approval_user_prompt, + ) + ) + return message_history @rx.event async def create_conversation(self) -> None: @@ -245,34 +260,16 @@ class ConversationState(rx.State): return True return conversation.is_running or not conversation.user_prompt - @rx.var - def is_running(self) -> bool: - """ - 当前会话正在运行 - :return: 当前会话正在运行（True 表示正在运行，False 表示运行完成） - """ - # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) - if not conversation: - return False - return conversation.is_running - - @rx.var - def awaiting_stream(self) -> bool: - """ - 当前会话正在等待流式输出 - :return: 当前会话正在等待流式输出（True 表示等待流式输出，False 表示已开始流式输出或已完成） - """ - # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) - if not conversation: - return False - return conversation.awaiting_stream - @rx.event - async def run(self) -> AsyncGenerator[None]: + async def run( + self, + approval: Approval | None = None, + deferred_tool_results: DeferredToolResults | None = None, + ) -> AsyncGenerator[None]: """ 运行 + :param approval: 审批实例 + :param deferred_tool_results: 延迟工具调用结果 :return: AsyncGenerator """ # 当前会话 @@ -280,36 +277,43 @@ class ConversationState(rx.State): if not conversation: return - # 若用户提示词为空则直接返回 - if not conversation.user_prompt: - return - - # 将正在运行设置为是 - conversation.is_running = True - # 将等待流式输出设置为是 - conversation.awaiting_stream = True - yield # 通知前端更新渲染 - # 获取数据库状态 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( - conversation_id=conversation.id, - message=Message( - type=MessageType.USER_PROMPT, - content=(user_prompt := conversation.user_prompt), - ), + + # 无审批实例 + if not approval: + # 若用户提示词为空则直接返回 + if not conversation.user_prompt: + return + + # 将正在运行设置为是 + conversation.is_running = True + # 将等待流式输出设置为是 + conversation.awaiting_stream = True + yield # 通知前端更新渲染 + + # 先创建消息记录再添加消息实例 + conversation.messages.update( + await db_state.create_message_record( + conversation_id=conversation.id, + message=Message( + type=MessageType.USER_PROMPT, + content=(user_prompt := conversation.user_prompt), + ), + ) ) - ) - # 清空用户提示词 - conversation.user_prompt = "" - yield # 通知前端更新渲染 + # 清空用户提示词 + conversation.user_prompt = "" + yield # 通知前端更新渲染 + + # 有审批实例 + else: + if not (user_prompt := approval.user_prompt): + return + + # 将等待流式输出设置为是 + conversation.awaiting_stream = True + yield # 通知前端更新渲染 # 初始化片段索引映射为消息实例唯一标识字典 index_map_to_message_id: dict[int, str] = {} @@ -328,10 +332,11 @@ class ConversationState(rx.State): stream_events = run_stream_events( deps=conversation.workflow.deps, user_prompt=user_prompt, - message_history=await db_state.get_message_history( + message_history=await db_state.read_message_history( conversation_id=self.actived_conversation_id ), usage=conversation.usage, + deferred_tool_results=deferred_tool_results, ) else: from application.workshop.talk import ( @@ -341,7 +346,7 @@ class ConversationState(rx.State): # 运行并流式输出事件 stream_events = run_stream_events( user_prompt=user_prompt, - message_history=await db_state.get_message_history( + message_history=await db_state.read_message_history( conversation_id=self.actived_conversation_id ), usage=conversation.usage, @@ -469,15 +474,17 @@ class ConversationState(rx.State): # ========== 智能体运行结果事件 ========== case AgentRunResultEvent(result=result): - if isinstance(result.output, DeferredToolRequests): - for approval in result.output.approvals: + if isinstance(output := result.output, DeferredToolRequests): + for approval_ in output.approvals: # 获取消息实例 message = conversation.messages[ - tool_name_map_to_message_id[approval.tool_name] + tool_name_map_to_message_id[approval_.tool_name] ] + message.content = output.metadata[ + tool_call_id := approval_.tool_call_id + ]["content"] message.approval = Approval( - tool_call_id=approval.tool_call_id, - interaction_type=approval.interaction_type, + tool_call_id=tool_call_id, ) messages.append(message) else: @@ -499,7 +506,7 @@ class ConversationState(rx.State): ) self.conversations[self.actived_conversation_id] = conversation - yield + yield # 通知前端更新渲染 # 批量创建消息记录 await db_state.create_message_records( @@ -507,6 +514,88 @@ class ConversationState(rx.State): messages=messages, ) + @rx.var + def is_running(self) -> bool: + """ + 当前会话正在运行 + :return: 当前会话正在运行（True 表示正在运行，False 表示运行完成） + """ + # 当前会话 + conversation = self.conversations.get(self.actived_conversation_id) + if not conversation: + return False + return conversation.is_running + + @rx.var + def awaiting_stream(self) -> bool: + """ + 当前会话正在等待流式输出 + :return: 当前会话正在等待流式输出（True 表示等待流式输出，False 表示已开始流式输出或已完成） + """ + # 当前会话 + conversation = self.conversations.get(self.actived_conversation_id) + if not conversation: + return False + return conversation.awaiting_stream + + @rx.event + def toggle_message_history_item_shown(self, item_id: str) -> None: + """ + 展示 / 隐藏消息历史项 + """ + # 当前会话 + conversation = self.conversations.get(self.actived_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: + """ + 设置审批用户提示词 + :param item_id: 消息历史项唯一标识 + :param user_prompt: 用户提示词 + """ + # 当前会话 + conversation = self.conversations.get(self.actived_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, result: bool) -> None: + """ + 审批消息历史项 + :param item_id: 消息历史项唯一标识 + :param result: 审批结果（True 表示审批通过，False 表示审批拒绝） + """ + # 当前会话 + conversation = self.conversations.get(self.actived_conversation_id) + if not conversation: + return + + # 获取审批实例 + if not (message := conversation.messages.get(item_id)) or not isinstance( + approval := message.approval, Approval + ): + return + + # 更新审批结果 + approval.result = result + # 运行 + await self.run( + approval, + DeferredToolResults( + approvals={ + approval.tool_call_id: ToolApproved() if result else ToolDenied() + } + ), + ) + @rx.event async def set_workflow(self, type: WorkflowType) -> None: """ @@ -529,14 +618,10 @@ class ConversationState(rx.State): ) conversation.workflow = workflow - - @rx.event - def toggle_message_history_item_shown(self, message_id: str) -> None: - """ - 展示 / 隐藏消息历史项 - """ - # 当前会话 - conversation = self.conversations.get(self.actived_conversation_id) - if not conversation: - return - conversation.messages[message_id].is_shown ^= True + # 获取数据库状态 + db_state = await self.get_db_state() + # 更新会话记录 + await db_state.update_conversation_record( + conversation.id, + workflow=conversation.workflow, + ) diff --git a/agent/application/states/database.py b/agent/application/states/database.py index c6bbf14..39ee00b 100644 --- a/agent/application/states/database.py +++ b/agent/application/states/database.py @@ -19,6 +19,8 @@ from application.states.models import ( usage_validate_json, workflow_dump_json, workflow_validate_json, + approval_dump_json, + approval_validate_json, ) @@ -101,7 +103,7 @@ class RunRecord(SQLModel, table=True): description="运行唯一标识", ) conversation_id: str = Field(..., index=True, description="会话唯一标识") - new_messages: str = Field(default="", sa_type=JSON, description="新增消息") + new_messages: str = Field(..., description="新增消息") class MessageRecord(SQLModel, table=True): @@ -118,6 +120,9 @@ class MessageRecord(SQLModel, table=True): type: MessageType = Field(..., description="消息类型") title: str = Field(default="", description="消息标题") content: str = Field(default="", description="消息内容") + approval: str = Field( + default="", description="审批" + ) # 原则上嵌套列表或字典若不涉及查询则以字符串储存 class DatabaseState(rx.State): @@ -197,9 +202,31 @@ class DatabaseState(rx.State): await session.refresh(record) return record.id - async def get_conversations(self, user_id: str) -> dict[str, Conversation]: + async def create_conversation_record(self, user_id: str) -> dict[str, Conversation]: """ - 获取当前用户的会话字典 + 创建会话记录 + :param user_id: 用户唯一标识 + :return: 会话实例 + """ + async with rx.asession() as session: + record = ConversationRecord(user_id=user_id) + session.add(record) + await session.commit() + await session.refresh(record) + return { + record.id: Conversation( + id=record.id, + description=record.description, + usage=usage_validate_json( + record.usage + ), # 将 record.usage 由 JSON 字符串反序列化为 RunUsage + created_at=record.created_at, + ) + } + + async def read_conversations(self, user_id: str) -> dict[str, Conversation]: + """ + 查询当前用户的会话字典 :param user_id: 用户唯一标识 :return: 当前用户的会话字典 """ @@ -241,33 +268,11 @@ class DatabaseState(rx.State): type=message_record.type, title=message_record.title, content=message_record.content, + approval=approval_validate_json(message_record.approval), ), ) return conversations - async def create_conversation_record(self, user_id: str) -> dict[str, Conversation]: - """ - 创建会话记录 - :param user_id: 用户唯一标识 - :return: 会话实例 - """ - async with rx.asession() as session: - record = ConversationRecord(user_id=user_id) - session.add(record) - await session.commit() - await session.refresh(record) - return { - record.id: Conversation( - id=record.id, - description=record.description, - usage=usage_validate_json( - record.usage - ), # 将 record.usage 由 JSON 字符串反序列化为 RunUsage - messages={}, # 新会话默认消息字典为空 - created_at=record.created_at, - ) - } - async def update_conversation_record( self, conversation_id: str, @@ -321,6 +326,7 @@ class DatabaseState(rx.State): type=message.type, title=message.title, content=message.content, + approval=approval_dump_json(message.approval), ) session.add(record) await session.commit() @@ -335,8 +341,8 @@ class DatabaseState(rx.State): """ 创建消息记录 :param conversation_id: 会话唯一标识 - :param message: 消息实例 - :return: 消息实例 + :param messages: 消息实例列表 + :return: None """ async with rx.asession() as session: for message in messages: @@ -346,13 +352,14 @@ class DatabaseState(rx.State): type=message.type, title=message.title, content=message.content, + approval=approval_dump_json(message.approval), ) session.add(record) await session.commit() - async def get_message_history(self, conversation_id: str) -> list[ModelMessage]: + async def read_message_history(self, conversation_id: str) -> list[ModelMessage]: """ - 获取消息历史列表 + 查询消息历史列表 :param conversation_id: 会话唯一标识 :return: 消息历史 """ diff --git a/agent/application/states/models.py b/agent/application/states/models.py index 28e782e..839164b 100644 --- a/agent/application/states/models.py +++ b/agent/application/states/models.py @@ -19,23 +19,6 @@ class NoResult(BaseModel): ... -class InteractionType(StrEnum): - """ - 交互类型枚举 - """ - - INPUT = "input" - - -class Approval(BaseModel): - """ - 审批类 - """ - - tool_call_id: str = Field(..., description="工具调用唯一标识") - interaction_type: InteractionType = Field(..., description="交互类型") - - class MessageType(StrEnum): """ 消息类型枚举 @@ -47,6 +30,19 @@ class MessageType(StrEnum): TEXT = "text" +class Approval(BaseModel): + """ + 审批类 + """ + + tool_call_id: str = Field(..., description="工具调用唯一标识") + user_prompt: str = Field(default="", description="用户提示词") + result: bool | None = Field( + default=None, + description="审批结果，True 表示审批通过，False 表示审批拒绝，None 表示待审批", + ) + + class Message(BaseModel): """ 消息类 @@ -118,6 +114,9 @@ class MessageHistoryItem(Message): 消息历史项类 """ + has_approval: bool = Field(default=False, description="包含审批实例") + approval_user_prompt: str = Field(default="", description="审批用户提示词") + # 工作流适配器 WorkflowAdapter = TypeAdapter(Any) @@ -161,3 +160,27 @@ def usage_dump_json(usage: RunUsage) -> str: :return: 字符串 """ return UsageAdapter.dump_json(usage).decode("utf-8") + + +# 审批适配器 +ApprovalAdapter = TypeAdapter(Approval) + + +def approval_validate_json(approval: str) -> Approval | None: + """ + 将审批反序列化 + """ + if not approval: + return None + return ApprovalAdapter.validate_json(approval) + + +def approval_dump_json(approval: Approval | None) -> str: + """ + 将审批序列化 + :param approval: 审批 + :return: 字符串 + """ + if approval is None: + return "" + return ApprovalAdapter.dump_json(approval).decode("utf-8") diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index fbc405f..16d3334 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -28,7 +28,6 @@ from pydantic_ai.run import AgentRunResult, AgentRunResultEvent from application.states.models import ( NoResult, - InteractionType, Approval, ) from application.workshop.models import ( @@ -207,7 +206,6 @@ async def search_flights(ctx: RunContext[Deps]) -> list[Flight]: @agent.tool( name="预定航班", - requires_approval=True, ) async def book_flight(ctx: RunContext[Deps]) -> Flight | NoResult: """ @@ -216,6 +214,9 @@ async def book_flight(ctx: RunContext[Deps]) -> Flight | NoResult: if (searched_flights := ctx.deps.searched_flights) is None: raise ModelRetry("必须先使用 search_flights 查询航班") + if not ctx.tool_call_approved: + raise ApprovalRequired(metadata={"content": "您要预定哪班航班？"}) + # 提取到的航班号 extracted_flight_number = ( await extraction_flight_number_agent.run( @@ -298,6 +299,7 @@ async def run_stream_events( deferred_tool_results=deferred_tool_results, ) as events: async for event in events: + print(event) match event: case PartStartEvent( part=part,