基于reflex的智能体初步实现

.vscode/settings.json

@@ -1,3 +1,4 @@
 {
-    "python.languageServer": "None"
+  "python.languageServer": "None",
+  "python.analysis.extraPaths": ["${workspaceFolder}"]
 }

utils/agent.py

@@ -4,53 +4,52 @@
 """
 
 # 列举导入模块
+from asyncio import Queue, QueueEmpty, Task, Task, create_task, sleep
 from pathlib import Path
 import time
-from typing import AsyncGenerator, List, Optional, Dict, Literal
+from typing import AsyncGenerator, Dict, List, Literal, Optional, Union
 from uuid import uuid4
-from pydantic_ai.messages import ModelMessage
+
+from pydantic import Field
+from pydantic import BaseModel
 from pydantic_ai import Agent as PydanticAIAgent, ModelMessage
 from pydantic_ai.capabilities import AgentCapability
-from asyncio import Task, create_task, sleep, Queue, QueueEmpty, Task
+from pydantic_ai.run import AgentRunResultEvent
-from pydantic import Field
-
-from pydantic import BaseModel
-
-
 from pydantic_ai.messages import (
-    ModelMessagesTypeAdapter,
+    ModelMessage,
-    ModelResponse,
     FinalResultEvent,
-    ModelMessageEvent,
+    LoadCapabilityCallPart,
-    PartStartEvent,
+    AgentStreamEvent,
+    ModelMessagesTypeAdapter,
+    NativeToolCallPart,
+    NativeToolReturnPart,
+    NativeToolSearchCallPart,
     PartDeltaEvent,
     PartEndEvent,
+    PartStartEvent,
     TextPart,
-    ToolSearchCallPart,
-    NativeToolSearchCallPart,
-    LoadCapabilityCallPart,
-    ThinkingPart,
-    ToolCallPart,
-    NativeToolCallPart,
     TextPartDelta,
+    ThinkingPart,
     ThinkingPartDelta,
-    ModelResponseStreamEvent,
+    ToolCallPartDelta,
+    ToolCallPart,
     ToolReturnPart,
-    AgentRunResultEvent,
+    ToolSearchCallPart,
-    NativeToolReturnPart,
 )
 from pydantic_ai.models.openai import OpenAIChatModel
 from pydantic_ai.output import OutputSpec
 from pydantic_ai.providers.openai import OpenAIProvider
 
-from .sqlite import SQLite
+from sys import path
+from pathlib import Path
+
+path.append(Path(__file__).parent.as_posix())
+from sqlite import SQLite
 
 
-class AgentMemory(SQLite):
+class Memory(SQLite):
     """
-    智能体记忆体，支持：
+    记忆体
-    create：新增对话消息
-    read：查询会话历史消息
     """
 
     def __init__(self):
@@ -58,7 +57,7 @@ class AgentMemory(SQLite):
         初始化
         """
         # 构建数据库路径
-        super().__init__(database=Path(__file__).parent.resolve() / "agent_memory.db")
+        super().__init__(database=Path(__file__).parent.resolve() / "memory.db")
 
         try:
             with self:
@@ -80,14 +79,12 @@ class AgentMemory(SQLite):
         except Exception as exception:
             raise RuntimeError(f"初始化数据库发生异常：{str(exception)}") from exception
 
-    def create_new_messages(
+    def save(self, session_id: str, new_messages: List[ModelMessage]) -> bool:
-        self, session_id: str, new_messages: List[ModelMessage]
-    ) -> bool:
         """
-        新增新对话消息
+        将本次对话消息保存至数据库
         :param session_id: 会话唯一标识
         :param new_messages: 新对话消息
-        :return: 新增是否成功
+        :return: 保存是否成功
         """
         try:
             with self:
@@ -107,7 +104,7 @@ class AgentMemory(SQLite):
 
     def get_message_history(self, session_id: str) -> List[ModelMessage]:
         """
-        获取指定会话的消息历史
+        获取消息历史
         :param session_id: 会话唯一标识
         :return: 消息历史
         """
@@ -142,13 +139,13 @@ class Buffer(BaseModel):
     part_type: Literal["thinking", "text"] = Field(..., description="片段类型")
     content_delta: str = Field(default="", description="片段内容增量")
     task: Optional[Task] = Field(default=None, description="延迟刷新异步协程任务")
+    model_config = {"arbitrary_types_allowed": True}
 
 
-class EventsDebouncer:
+class Debouncer:
     """
-    模型消息事件防抖器
+    防抖器
-    使用 Pydantic-AI 中 run_stream_events 方法时需就其返回的模型消息事件处理，延迟刷新片段类型为思考或文本的片段增量事件以实现防抖
+    用于处理 run_stream_events 返回的模型消息事件
-    在 Pydantic-AI 中模型流式输出包含若干类型片段，例如思考片段、文本片段等。每类型片段包含若干模型消息事件，例如片段开始事件、片段增量事件、片段结束事件等
     """
 
     def __init__(self, delay: float = 0.2):
@@ -164,10 +161,11 @@ class EventsDebouncer:
         self.pending_flush_queue = Queue()
 
     async def handle_event(
-        self, event: ModelMessageEvent
+        self, event: Union[AgentStreamEvent, AgentRunResultEvent]
-    ) -> AsyncGenerator[ModelMessageEvent]:
+    ) -> AsyncGenerator[Union[AgentStreamEvent, AgentRunResultEvent]]:
         """
         处理模型消息事件
+        在 Pydantic-AI 中模型流式输出包含若干类型片段，例如思考片段、文本片段等。每类型片段包含若干模型消息事件，例如片段开始事件、片段增量事件、片段结束事件等
         :param event: 模型消息事件
         :yield: AsyncGenerator
         """
@@ -294,20 +292,22 @@ class Agent:
         instructions: str,
         output_type: OutputSpec = str,
         capabilities: Optional[List[AgentCapability]] = None,
+        retries: int = 1,
     ):
         """
         初始化智能体
         :param session_id: 会话唯一标识
         :param instructions: 指令
-        :param skills: 智能体技能列表，默认为不使用技能
+        :param capabilities: 技能列表，默认为不使用技能
         :param output_type: 输出类型
+        :param retries: 重试次数，默认为1次
         :return: 智能体实例
         """
         # 会话唯一标识
         self.session_id = session_id
 
         # 实例智能体的记忆体
-        self.agent_memory = AgentMemory()
+        self.memory = Memory()
 
         # 实例智能体
         self.agent = PydanticAIAgent(
@@ -321,7 +321,7 @@ class Agent:
             instructions=instructions,
             capabilities=capabilities,
             output_type=output_type,
-            retries=1,
+            retries=retries,
         )
 
         self.agent.to_web()
@@ -337,62 +337,61 @@ class Agent:
         """
         """定义：一次会话（session）包含若干论对话（turn），每一轮对话由用户输入消息（message）和智能体输出消息组成"""
         # 获取指定会话的消息历史
-        message_history = self.agent_memory.get_message_history(
+        message_history = self.memory.get_message_history(session_id=self.session_id)
-            session_id=self.session_id
-        )
 
-        # 实例模型消息事件防抖器
+        # 实例防抖器
-        event_debouncer = EventsDebouncer(delay=delay)
+        debouncer = Debouncer(delay=delay)
 
         async with self.agent.run_stream_events(
             user_prompt=user_prompt,
             message_history=message_history,
         ) as events:
             async for event in events:
-                if isinstance(event, AgentRunResultEvent):
-                    evnet : AgentRunResultEvent = event
-                    new_messages = evnet.result
-                
                 # 处理模型消息事件
-                async for event_ in event_debouncer.handle_event(event):
+                async for event in debouncer.handle_event(event):
-                    match event_:  # event_ 为处理后模型消息事件
+                    match event:
-                        case PartStartEvent():
+                        # 片段开始事件
-                            part = event_.part
+                        case PartStartEvent(part=part):
                             match part:
                                 case ThinkingPart(content=content):
                                     yield f"00:{content}"
                                 case TextPart(content=content):
                                     yield f"01:{content}"
                                 case (
-                                    NativeToolSearchCallPart()
+                                    NativeToolSearchCallPart(tool_name=tool_name)
-                                    | NativeToolCallPart()
+                                    | NativeToolCallPart(tool_name=tool_name)
-                                    | ToolSearchCallPart()
+                                    | ToolSearchCallPart(tool_name=tool_name)
-                                    | ToolCallPart()
+                                    | ToolCallPart(tool_name=tool_name)
-                                    | LoadCapabilityCallPart()
+                                    | LoadCapabilityCallPart(tool_name=tool_name)
-                                ) as tool:
+                                ):
-                                    yield f"02:使用技能：{tool.tool_name}"
+                                    yield f"02:正在使用技能：{tool_name}"
-                                case (
+                                case NativeToolReturnPart(
-                                    NativeToolReturnPart() | ToolReturnPart()
+                                    content=content
-                                ) as tool_return:
+                                ) | ToolReturnPart(content=content):
-                                    yield f"03:技能返回：{tool_return.content}"
+                                    yield f"04:技能返回：{content}"
                                 case _:
-                                    yield f"99:未知片段类型{type(part)}"
+                                    yield f"06:未知片段类型{type(part).__name__}"
-                        case PartDeltaEvent():
+                        # 片段增量事件
-                            delta = event_.delta
+                        case PartDeltaEvent(delta=delta):
                             match delta:
                                 case ThinkingPartDelta(content_delta=content_delta):
                                     yield f"00:{content_delta}"
                                 case TextPartDelta(content_delta=content_delta):
                                     yield f"01:{content_delta}"
+                                case ToolCallPartDelta(args_delta=args_delta):
+                                    yield f"03:技能参数：{args_delta}"
                                 case _:
-                                    yield f"99:未知片段类型{type(delta)}"
+                                    yield f"06:未知片段类型{type(delta).__name__}"
-                        # 片段结束事件无需处理
+                        # 片段结束事件、最终结果事件无需处理
-                        case PartEndEvent() | AgentRunResultEvent():
+                        case PartEndEvent():
-                            pass
+                            continue
+                        case FinalResultEvent():
+                            continue
+                        case AgentRunResultEvent(result=result):
+                            self.memory.save(
+                                session_id=self.session_id,
+                                new_messages=result.new_messages(),
+                            )
+                            yield "05:FinalResultEvent"
                         case _:
-                            yield f"99:未知事件类型{type(event_)}"
+                            yield f"06:未知事件类型{type(event).__name__}"
-
-        self.agent_memory.create_new_messages(
-            session_id=self.session_id,
-            new_messages=new_messages,
-        )

产品需求文档AI生成/application/application.py

@@ -4,7 +4,7 @@
 """
 
 import reflex
-from .pages.index import index
+from .page.index import index
 
 app = reflex.App()
 app.add_page(index)

产品需求文档AI生成/application/backend/state.py

@@ -3,22 +3,24 @@
 应用状态管理模块
 """
 
-from typing import Any, AsyncGenerator, Dict, List, Literal
-from uuid import uuid4
 from enum import StrEnum
+from uuid import uuid4
+from typing import Any, AsyncGenerator, Dict, List
+
 from pydantic import BaseModel, Field
 import reflex
-from pathlib import Path
-from sys import path
 
-path.append((Path(__file__).resolve().parent.parent.parent.as_posix()))
+from sys import path
+from pathlib import Path
+
+path.append(Path(__file__).parent.parent.parent.parent.as_posix())
 from utils.agent import Agent
 
 # 所有会话绑定的智能体
-agents: Dict[str, Any] = {}
+agents: Dict[str, Agent] = {}
 
 
-def retrieve_agent(state) -> Agent:
+def get_current_session_agent(state) -> Agent:
     """
     获取当前会话绑定的智能体
     :return: 当前会话绑定的智能体
@@ -26,58 +28,61 @@ def retrieve_agent(state) -> Agent:
     current_session_name = state.current_session_name
     if current_session_name not in agents:
         agents[current_session_name] = Agent(
-            session_id=state.sessions[current_session_name].id,
+            session_id=uuid4().hex,
             instructions="You are a friendly chatbot",
         )
     return agents[current_session_name]
 
 
-# 消息块类型
+class MessageType(StrEnum):
-class MessageBlockType(StrEnum):
+    """消息类型类"""
 
-    content = "content"
+    THINKING = "thinking"
-    thinking = "thinking"
+    TEXT = "text"
-    tool_call = "tool_call"
+    CALL = "call"
-    tool_result = "tool_result"
+    TOOL_ARGS = "tool_args"
-    error = "error"
+    TOOL_RETURN = "tool_return"
+    RESULT = "result"
+    ERROR = "error"
 
 
-# 消息块类型前缀映射
+# 消息类型前缀映射
-MESSAGE_BLOCK_TYPE_PREFIX_MAP = {f"{i:02d}:": m for i, m in enumerate(MessageBlockType)}
+MESSAGE_TYPE_PREFIX_MAP = {f"{i:02d}:": mt for i, mt in enumerate(MessageType)}
 
 
-class MessageBlock(BaseModel):
+# 会话、对话、消息关系：一次会话包含若干轮对话，每轮对话包含若干条消息
-    """消息块数据模型，包含类型和内容"""
+class Message(BaseModel):
+    """消息类"""
 
-    type: MessageBlockType = Field(..., description="类型")
+    id: str = Field(default_factory=lambda: uuid4().hex, description="消息唯一标识")
+    type_: MessageType = Field(..., description="类型")
     content: str = Field(default="", description="内容")
 
 
 class Turn(BaseModel):
-    """对话数据模型，包含用户输入消息和智能体输出消息"""
+    """对话类"""
 
-    input: str = Field(..., description="用户输入的消息")
+    id: str = Field(default_factory=lambda: uuid4().hex, description="对话唯一标识")
-    output: List[MessageBlock] = Field(
+    input_: str = Field(..., description="输入消息")
-        default_factory=list, description="智能体输出的消息"
+    output: List[Message] = Field(default_factory=list, description="输出消息")
-    )
 
 
 class Session(BaseModel):
-    """会话数据模型，包含会话唯一标识和会话对话列表"""
+    """会话类"""
 
-    id: str = Field(default_factory=lambda: uuid4().hex, description="会话唯一标识")
+    is_processing: bool = Field(
-    is_processing: bool = Field(default=False, description="会话是否正在处理中")
+        default=False,
-    turns: List[Turn] = Field(default_factory=list, description="会话对话列表")
+        description="会话状态：True 表示正在处理中，False 表示未处理或处理完成",
+    )
+    turns: List[Turn] = Field(default_factory=list, description="对话列表")
 
 
-# Reflex.State 统一管理应用数据与功能状态，作为前后端交互枢纽，借助响应式特性实现页面自动更新
 class State(reflex.State):
-    """应用状态"""
+    """统一管理应用数据与功能状态，作为前后端交互枢纽，借助响应式特性实现页面自动更新"""
 
     # 当前会话名称
-    current_session_name: str = "NewSession"
+    current_session_name: str = "新会话"
-
+    # 会话列表（会话名称作为会话对象的唯一标识，不允许重复）
-    # 所有会话
     sessions: Dict[str, Session] = {current_session_name: Session()}
 
     # 新建会话模态窗是否打开
@@ -86,8 +91,8 @@ class State(reflex.State):
     @reflex.var
     def get_session_names(self) -> List[str]:
         """
-        获取所有会话名称
+        获取会话名称列表
-        :return: 所有会话名称
+        :return: 会话名称列表
         """
         return list(self.sessions)
 
@@ -104,7 +109,7 @@ class State(reflex.State):
     def get_current_session_status(self) -> bool:
         """
         获取当前会话状态
-        :return: 当前会话状态，其中 True 表示正在处理中，False 表示处理完成
+        :return: 当前会话状态
         """
         if self.current_session_name not in self.sessions:
             return False
@@ -114,7 +119,7 @@ class State(reflex.State):
     def get_current_session_turns(self) -> List[Turn]:
         """
         获取当前会话对话列表
-        :return: 当前会话对话列表
+        :return: 对话列表
         """
         if self.current_session_name not in self.sessions:
             return []
@@ -123,19 +128,19 @@ class State(reflex.State):
     @reflex.event
     def create_session(self, form_data: Dict[str, Any]) -> None:
         """
-        创建会话
+        新建会话
-        :param form_data: 创建会话表单数据
+        :param form_data: 新建会话表单数据
         :return: None
         """
         session_name = form_data["session_name"].strip()
 
-        # 若创建会话名称为空则默认使用"NewSession"作为会话名称
+        # 若新建会话名称为空则默认使用"新会话"作为会话名称
         if not session_name:
-            session_name = "NewSession"
+            session_name = "新会话"
 
         original_session_name = session_name
         counter = 1
-        # 若会话名称重复则在会话名称后面添加序号至不重复
+        # 若会话名称重复则在会话名称后面添加标号至不重复
         while session_name in self.sessions:
             session_name = f"{original_session_name}({counter})"
             counter += 1
@@ -155,14 +160,13 @@ class State(reflex.State):
         """
         if session_name not in self.sessions:
             return
-
         del self.sessions[session_name]
 
-        # 若删除会话后所有会话为空则默认创建空白会话
+        # 若会话列表为空则新建会话（区别于 create_session 方法，此处为后台新建会话）
         if not self.sessions:
-            self.sessions["NewSession"] = Session()
+            self.sessions["新会话"] = Session()
 
-        # 删除会话后，若当前会话名称不存在则默认使用第一个会话名称
+        # 若当前会话名称不存在则默认使用第一个会话名称
         if self.current_session_name not in self.sessions:
             self.current_session_name = next(iter(self.sessions))
 
@@ -178,63 +182,52 @@ class State(reflex.State):
     @reflex.event
     async def adapt_input(self, form_data: dict[str, Any]) -> AsyncGenerator:
         """
-        适配用户输入
+        适配输入
-        :param form_data: 对话表单数据
+        :param form_data: 输入栏组件的表单数据
         :return: AsyncGenerator
         """
-        input = form_data["input"].strip()
+        input_ = form_data["input"].strip()
-        if not input:
+        if not input_:
             return
 
-        async for value in self.process_input(input=input):
-            yield value
-
-    async def process_input(self, input: str) -> AsyncGenerator:
-        """
-        处理用户输入
-        :param input: 用户输入
-        :return: AsyncGenerator
-        """
         # 当前会话
         current_session = self.sessions[self.current_session_name]
         # 当前会话正在处理
         current_session.is_processing = True
-        # 将用户输入添加到当前会话对话列表
+        # 将输入添加到当前会话对话列表
         current_session.turns.append(
             Turn(
-                input=input,
+                input_=input_,
             )
         )
-        yield  # 通知前端更新状态（显示用户输入）
+        yield  # 通知前端渲染输入消息
 
         # 当前对话
         current_turn = current_session.turns[-1]
 
         # 获取当前会话绑定的智能体
-        agent = retrieve_agent(self)
+        agent = get_current_session_agent(self)
-        async for chunk in agent.output_message_streamed(user_prompt=input):
+        async for event in agent.stream_messages_events(user_prompt=input_):
-            # 跳过空分块
+            # 跳过空事件
-            if not chunk:
+            if not event:
-                yield
                 continue
 
-            # 匹配消息块类型
+            # 匹配消息类型
-            prefix_matched = next(
+            prefix = next(
-                (t for t in MESSAGE_BLOCK_TYPE_PREFIX_MAP if chunk.startswith(t)), None
+                (t for t in MESSAGE_TYPE_PREFIX_MAP if event.startswith(t)), None
             )
-            # 跳过未匹配分块
+            # 跳过未匹配事件
-            if not prefix_matched:
+            if not prefix:
-                yield
                 continue
 
-            # 消息块类型
+            # 消息类型
-            type = MESSAGE_BLOCK_TYPE_PREFIX_MAP[prefix_matched]
+            type_ = MESSAGE_TYPE_PREFIX_MAP[prefix]
-            # 若当前对话输出为空或当前消息块类型和上一个消息块类型不一致则创建消息块
+            # 若当前对话输出为空或当前消息类型和上一个消息类型不一致则创建消息
-            if not current_turn.output or current_turn.output[-1].type != type:
+            if not current_turn.output or current_turn.output[-1].type_ != type_:
-                current_turn.output.append(MessageBlock(type=type))
+                current_turn.output.append(Message(type_=type_))
-            current_turn.output[-1].content += chunk.removeprefix(prefix_matched)
+            current_turn.output[-1].content += event.removeprefix(prefix)
 
-            yield  # 通知前端更新状态（打字机效果显示输出）
+            yield  # 通知前端渲染输出消息
 
         # 当前会话处理完成
         current_session.is_processing = False

产品需求文档AI生成/application/frontend/navigation_bar.py

@@ -5,7 +5,7 @@
 
 import reflex
 
-from ..state import State
+from ..backend.state import State
 
 
 def create_session_modal(trigger) -> reflex.Component:

产品需求文档AI生成/application/frontend/session.py

@@ -5,12 +5,12 @@
 import reflex
 from reflex.constants.colors import ColorType
 
-from ..state import State, MessageBlockType, MessageBlock, Turn
+from ..backend.state import State, MessageType, Message, Turn
 
 
 def input_bubble(message: str, color: ColorType) -> reflex.Component:
     """
-    输入气泡组件
+    输入消息气泡组件
     :param message: 消息
     :param color: 颜色
     :return: Component
@@ -25,26 +25,26 @@ def input_bubble(message: str, color: ColorType) -> reflex.Component:
     )
 
 
-def output_bubble(message_block: MessageBlock) -> reflex.Component:
+def output_bubble(message: Message) -> reflex.Component:
     """
-    输出气泡组件
+    输出消息气泡组件
-    :param message_block: 消息块
+    :param message: 消息
     :return: 气泡组件
     """
     color = reflex.cond(
-        message_block.type == MessageBlockType.content,
+        message.type_ == MessageType.TEXT,
         "accent",
         reflex.cond(
-            message_block.type == MessageBlockType.thinking,
+            message.type_ == MessageType.THINKING,
             "iris",
             reflex.cond(
-                message_block.type == MessageBlockType.tool_call,
+                message.type_ == MessageType.CALL,
                 "orange",
                 reflex.cond(
-                    message_block.type == MessageBlockType.tool_result,
+                    message.type_ == MessageType.TOOL_RETURN,
                     "teal",
                     reflex.cond(
-                        message_block.type == MessageBlockType.error,
+                        message.type_ == MessageType.ERROR,
                         "red",
                         "mauve",  # 兜底
                     ),
@@ -53,7 +53,7 @@ def output_bubble(message_block: MessageBlock) -> reflex.Component:
         ),
     )
     return reflex.markdown(
-        message_block.content,
+        message.content,
         color=reflex.color(color=color, shade=12),
         background_color=reflex.color(color=color, shade=4),
         display="inline-block",
@@ -61,18 +61,19 @@ def output_bubble(message_block: MessageBlock) -> reflex.Component:
         padding_block="0.5em",
         border_radius="8px",
         margin_bottom="4px",
+        key=message.id,
     )
 
 
 def turn(turn: Turn) -> reflex.Component:
     """
-    对话组件
+    一轮对话组件
     :param turn: 对话
     :return: Component
     """
     return reflex.box(
         reflex.box(
-            input_bubble(message=turn.input, color="mauve"),
+            input_bubble(message=turn.input_, color="mauve"),
             text_align="right",
             margin_bottom="8px",
         ),
@@ -83,6 +84,7 @@ def turn(turn: Turn) -> reflex.Component:
         ),
         max_width="50em",
         margin_inline="auto",
+        key=turn.id,
     )
 
 

产品需求文档AI生成/application/page/index.py

@@ -5,8 +5,8 @@
 
 import reflex
 
-from ..components.navigation_bar import navigation_bar
+from ..frontend.navigation_bar import navigation_bar
-from ..components.session import session_area, input_bar
+from ..frontend.session import session_area, input_bar
 
 
 def index() -> reflex.Component: