diff --git a/智能体/application/models.py b/智能体/application/models.py index 019f623..ecdb853 100644 --- a/智能体/application/models.py +++ b/智能体/application/models.py @@ -74,7 +74,7 @@ class Kind(StrEnum): THINKING = "thinking" TOOL_SEARCH = "tool-search" - CAPABILITY_LOAD = "capability-load" + LOAD_CAPABILITY = "load-capability" TOOL_CALL = "tool-call" TEXT = "text" TOOL_RETURN = "tool-return" diff --git a/智能体/application/state/conversation.py b/智能体/application/state/conversation.py index 682de0d..935cbcb 100644 --- a/智能体/application/state/conversation.py +++ b/智能体/application/state/conversation.py @@ -21,6 +21,7 @@ from pydantic_ai.messages import ( ToolCallPart, ToolSearchCallPart, ) +from pydantic_ai.run import AgentRunResultEvent from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.providers.openai import OpenAIProvider import reflex as rx @@ -193,8 +194,8 @@ class ConversationState(rx.State): conversation_id=self.conversation_id ) - # 初始化工具调用唯一标识与片段索引映射字典 - tool_call_ids : Dict[int, str] = {} + # 初始化工具调用唯一标识和片段索引映射字典 + tool_call_ids: Dict[str, int] = {} # 创建运行 conversation.runs[str(uuid7())] = Run(user_prompt=user_prompt) @@ -202,7 +203,7 @@ class ConversationState(rx.State): run_id, run = next(reversed(conversation.runs.items())) # 将运行状态设置为正在运行 run.is_running = True - yield # 通知前端渲染运行状态 + yield # 通知前端渲染 async with self.agent.run_stream_events( conversation_id=self.conversation_id, @@ -211,7 +212,7 @@ class ConversationState(rx.State): ) as events: async for event in events: match event: - # ========== 分片开始事件 ========== + # ========== 开始事件 ========== case PartStartEvent( index=index, part=part, @@ -219,82 +220,79 @@ class ConversationState(rx.State): ): match part: # 思考分片开始事件 - case ThinkingPart( - content=content - ): + case ThinkingPart(content=content): # 若上一分片种类为空则将推理状态设置为正在推理 if not previous_part_kind: run.is_reasoning = True - # 初始化推理 - run.reasonings[index] = Reasoning(kind=Kind.THINKING, content=content) - yield # 通知前端渲染推理 + run.reasonings[index] = Reasoning( + kind=Kind.THINKING, content=content + ) + yield - case ToolSearchCallPart( - tool_call_id=tool_call_id - ): + # 工具检索分片开始事件 + case ToolSearchCallPart(tool_call_id=tool_call_id): # 创建工具调用唯一标识与片段索引映射 - tool_call_ids[index] = tool_call_id + tool_call_ids[tool_call_id] = index - # 初始化推理 - run.reasonings[index] = Reasoning(kind=Kind.TOOL_SEARCH, content="正在检索") - yield # 通知前端渲染推理 + run.reasonings[index] = Reasoning( + kind=Kind.TOOL_SEARCH, content="正在检索" + ) + yield + + # 能力加载分片开始事件 + case LoadCapabilityCallPart(tool_call_id=tool_call_id): + tool_call_ids[tool_call_id] = index + + run.reasonings[index] = Reasoning( + kind=Kind.LOAD_CAPABILITY, content="正在加载能力" + ) + yield + + # 工具调用分片开始事件 + case ToolCallPart(tool_call_id=tool_call_id): + tool_call_ids[tool_call_id] = index + + run.reasonings[index] = Reasoning( + kind=Kind.TOOL_CALL, content="正在调用工具" + ) + yield # 文本分片开始事件 - case TextPart( - content=content - ): - # 初始化回复正文 + case TextPart(content=content): run.assistant_content = content - yield # 通知前端渲染回复正文 + yield - case _: - continue - - # ========== 分片增量事件（前端仅就文本片段实现打字机效果） ========== - case PartDeltaEvent( - index=index, delta=delta - ): + # ========== 增量事件 ========== + case PartDeltaEvent(index=index, delta=delta): match delta: # 思考分片增量事件 case ThinkingPartDelta( content_delta=content_delta, ): run.reasonings[index].content += content_delta or "" - yield # 通知前端渲染推理 + yield # 文本分片增量事件 case TextPartDelta( content_delta=content_delta, ): run.assistant_content += content_delta - yield # 通知前端渲染回复正文 + yield - case _: - continue - - # ========== 分片结束事件 ========== + # ========== 结束事件 ========== case PartEndEvent( - event_kind=event_kind, index=index, part=part, next_part_kind=next_part_kind, ): match part: # 思考分片结束事件 - case ThinkingPart( - part_kind=part_kind, content=content - ): + case ThinkingPart(part_kind=part_kind, content=content): # 若下一分片种类为文本则将推理状态设置为非正在推理 if next_part_kind == Kind.TEXT: run.is_reasoning = False - - # 检索工具分片结束事件 - case ToolSearchCallPart( - tool_call_id=tool_call_id - ): - # 创建工具调用唯一标识与片段索引映射 - tool_call_ids[index] = tool_call_id + yield # 加载能力分片结束事件 case LoadCapabilityCallPart( @@ -309,9 +307,7 @@ class ConversationState(rx.State): run_id=current_run_id, ) # 调用工具分片结束事件 - case ToolCallPart( - part_kind=part_kind, tool_name=tool_name - ): + case ToolCallPart(part_kind=part_kind, tool_name=tool_name): yield Event( event_kind=EventKind(event_kind), part_index=index, @@ -323,114 +319,34 @@ class ConversationState(rx.State): # ========== 函数工具调用事件 ========== case FunctionToolCallEvent( - event_kind=event_kind, part=part, ): - yield Event( - event_kind=EventKind(event_kind), - part_index=self.tool_call_ids.get( - part.tool_call_id - ), - part_kind=PartKind(part.part_kind), - tool_name=part.tool_name, - run_id=current_run_id, - ) + match part: + # 工具检索分片函数工具调用事件 + case ToolSearchCallPart(tool_call_id=tool_call_id): + index = tool_call_ids[tool_call_id] + run.reasonings[index].content = "检索完成" - # ========== 函数工具回调事件 ========== + # ========== 函数工具结果事件 ========== case FunctionToolResultEvent( - event_kind=event_kind, content=content, part=part, ): - yield Event( - event_kind=EventKind(event_kind), - event_content=( - content if isinstance(content, str) else "" - ), # 暂仅处理文本类型，后续处理多模态和缓存 - part_index=self.tool_call_ids.get( - part.tool_call_id - ), - part_kind=PartKind(part.part_kind), - tool_name=part.tool_name, - run_id=current_run_id, + match part: + # 工具检索分片函数工具结果事件 + case ToolSearchCallPart(tool_call_id=tool_call_id): + index = tool_call_ids[tool_call_id] + run.reasonings[index].content += content or "" + yield + + # ========== 智能体运行结果事件 ========== + case AgentRunResultEvent(result=result): + # 保存新增消息 + await database_state.save_new_messages( + conversation_id=self.conversation_id, + run_id=run_id, + new_messages=result.new_messages(), ) - -yield Event( - event_kind=EventKind.RUN_END, - run_id=current_run_id, - run_new_messages=agent_run.new_messages(), -) - - - - - - - - - - - - # 运行，处理分片生命周期事件 - async for self.agent.run( - conversation_id=self.conversation_id, - user_prompt=user_prompt, - message_history=message_history, - ) as async_event: - if ( - not event - or not self.current_run_id - or self.current_run_id != event.run_id - ): - continue - - # ========== 运行开始 ========== - if event.event_kind == EventKind.RUN_START: - # 将事件运行唯一标识设置为当前运行唯一标识 - self.current_run_id = event.run_id - # 创建运行 - conversation.runs[self.current_run_id] = Run( - user_prompt=user_prompt, is_streaming=True - ) - yield # 通知前端渲染 - continue - - # 当前运行 - current_run = current_conversation.runs[self.current_run_id] - - # ========== 推理 ========== - if ( - event.event_kind == EventKind.PART_START - and event.part_kind == PartKind.THINKING - ): - current_run.thinking += event.event_content - yield # 通知前端渲染 - continue - - if part_index := event.part_index: - if part_index not in current_run.reasonings: - current_run.reasonings[part_index] += event.event_content - yield # 通知前端渲染 - continue - - # ========== 回答 ========== - if ( - event.part_kind == PartKind.TEXT - and event.event_content != EventKind.PART_END - ): - current_run.answer += event.event_content - yield # 通知前端渲染 - continue - - # ========== 运行结束 ========== - if event.event_kind == EventKind.RUN_END: - # 保存运行新增消息 - await database_state.save_new_message( - conversation_id=self.current_conversation_id, - run_id=self.current_run_id, - run_new_message=event.run_new_messages, - ) - # 设置当前运行流式输出状态非正在流式输出 - current_run.is_streaming = False - yield # 通知前端渲染 - continue + # 将运行状态设置为非正在运行 + run.is_running = False + yield diff --git a/智能体/application/state/database.py b/智能体/application/state/database.py index b894944..c22e85e 100644 --- a/智能体/application/state/database.py +++ b/智能体/application/state/database.py @@ -34,21 +34,21 @@ class DatabaseState(rx.State): ) return message_history - async def save_new_message( - self, conversation_id: str, run_id: str, new_message: List[ModelMessage] + async def save_new_messages( + self, conversation_id: str, run_id: str, new_messages: List[ModelMessage] ) -> None: """ 保存新增消息 :param conversation_id: 对话唯一标识 :param run_id: 运行唯一标识 - :param new_message: 新增消息 + :param new_messages: 新增消息 :return: None """ async with rx.asession() as session: record = MessageHistory.adapt( conversation_id=conversation_id, run_id=run_id, - new_message=new_message, + new_message=new_messages, ) session.add(record) await session.commit()