1

agent/application/pages/conversation.py

@@ -9,7 +9,7 @@ from application.states.models import (
     ConversationHistoryItem,
     MessageHistoryItem,
     MessageType,
-    WorkType,
+    WorkflowType,
 )
 
 
@@ -278,8 +278,9 @@ def use_guidance() -> rx.Component:
             rx.hstack(
                 rx.box(
                     "预定航班",
-                    on_click=lambda: ConversationState.init_work(
+                    # 点击事件：设置工作流为预定航班
-                        WorkType.BOOK_FLIGHT
+                    on_click=lambda: ConversationState.set_workflow(
+                        WorkflowType.BOOK_FLIGHT
                     ),
                     padding="10px 16px",
                     background_color="var(--prismui-background-color-6)",

agent/application/states/conversation.py

@@ -2,8 +2,8 @@
 """
 会话状态
 """
-from typing import AsyncGenerator
 from datetime import datetime
+from typing import AsyncGenerator
 
 from pydantic_ai.messages import (
     FunctionToolCallEvent,
@@ -21,19 +21,20 @@ from pydantic_ai.messages import (
 )
 from pydantic_ai.run import AgentRunResultEvent
 import reflex as rx
-from pydantic_ai._uuid import uuid7
+
 from application.states.database import DatabaseState
 from application.states.models import (
-    Conversation,
-    Message,
-    MessageType,
-    Work,
-    WorkType,
-    TaskStatus,
     AgentRunWarmUpEvent,
+    Conversation,
     ConversationHistoryItem,
-    usage_validate_python,
+    Message,
     MessageHistoryItem,
+    MessageType,
+    Workflow,
+    WorkflowType,
+    deps_dump_python,
+    deps_validate_python,
+    usage_validate_python,
 )
 
 
@@ -270,7 +271,7 @@ class ConversationState(rx.State):
     async def run(self) -> AsyncGenerator[None]:
         """
         运行
-        :return: AsyncGenerator[None]
+        :return: AsyncGenerator
         """
         # 当前会话
         conversation = self.conversations.get(self.actived_conversation_id)
@@ -306,33 +307,44 @@ class ConversationState(rx.State):
         # 初始化片段索引映射为消息实例唯一标识字典
         index_map_to_message_id: dict[int, str] = {}
 
-        # 对话
+        # 初始化工作流
-        if not conversation.work:
+        if conversation.workflow:
+            match conversation.workflow.type:
+                case WorkflowType.BOOK_FLIGHT:
+                    from application.workshop.book_flight import (
+                        run_stream_events,
+                    )
+
+                    # 运行并流式输出事件
+                    stream_events = run_stream_events(
+                        deps=deps_validate_python(dict(conversation.workflow.deps)),
+                        user_prompt=user_prompt,
+                        message_history=await db_state.get_message_history(
+                            conversation_id=self.actived_conversation_id
+                        ),
+                        usage=usage_validate_python(dict(conversation.usage)),
+                    )
+        else:
             from application.workshop.talk import (
                 run_stream_events,
             )
-        else:
+
-            # 匹配工作类型
+            # 运行并流式输出事件
-            match conversation.work.type:
+            stream_events = run_stream_events(
-                # 预定航班
+                user_prompt=user_prompt,
-                case WorkType.BOOK_FLIGHT:
+                message_history=await db_state.get_message_history(
-                    from application.workshop.book_flight import run_stream_events
+                    conversation_id=self.actived_conversation_id
+                ),
+                usage=usage_validate_python(dict(conversation.usage)),
+            )
 
         # 消息列表
         messages: list[Message] = []
-        # 运行并流式输出事件
-        stream_events = run_stream_events(
-            user_prompt=user_prompt,
-            message_history=await db_state.get_message_history(
-                conversation_id=self.actived_conversation_id
-            ),
-            usage=usage_validate_python(dict(conversation.usage)),
-        )
         # 获取运行流式输出事件
         async for event in stream_events:
-            # 将等待流式输出设置为否
+            # 若等待流式输出则将等待流式输出设置为否
-            conversation.awaiting_stream = False
+            if conversation.awaiting_stream:
-            message: Message | None = None
+                conversation.awaiting_stream = False
             match event:
                 # ========== 开始事件 ==========
                 case PartStartEvent(
@@ -451,22 +463,27 @@ class ConversationState(rx.State):
             )
 
     @rx.event
-    async def init_work(self, work_type: WorkType) -> None:
+    async def set_workflow(self, type: WorkflowType) -> None:
         """
-        初始化工作
+        设置工作流
         """
         # 当前会话
         conversation = self.conversations.get(self.actived_conversation_id)
         if not conversation:
             return
 
-        match work_type:
+        match type:
-            # 预定航班工作
+            case WorkflowType.BOOK_FLIGHT:
-            case WorkType.BOOK_FLIGHT:
+                from application.workshop.book_flight import deps
-                from application.workshop.book_flight import init_work
 
-        # 初始化预定航班任务
+                # 初始化预定航班工作流
-        conversation.work = (work := init_work())
+                workflow = Workflow(type=type, deps=deps_dump_python(deps))
+                # 设置用户提示词
+                self.set_user_prompt(
+                    f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班"
+                )
+
+        conversation.workflow = workflow
 
     @rx.event
     def toggle_message_history_item_shown(self, message_id: str) -> None:

agent/application/states/database.py

@@ -15,7 +15,7 @@ from application.states.models import (
     Conversation,
     Message,
     MessageType,
-    usage_dump_python,
+    usage_validate_json,
 )
 
 
@@ -72,8 +72,10 @@ class ConversationRecord(SQLModel, table=True):
     )
     user_id: str = Field(..., index=True, description="用户唯一标识")
     description: str = Field(default="新会话", description="会话描述")
-    usage: dict[str, Any] = Field(
+    usage: str = Field(default="", description="会话使用量")
-        default_factory=dict, sa_type=JSON, description="会话使用量"
+    workflow: str = Field(
+        default="",
+        description="会话工作流",
     )
     is_deleted: bool = Field(
         default=False,
@@ -85,6 +87,20 @@ class ConversationRecord(SQLModel, table=True):
     )
 
 
+class RunRecord(SQLModel, table=True):
+    """
+    运行记录
+    """
+
+    id: str = Field(
+        default_factory=lambda: str(uuid7()),
+        primary_key=True,
+        description="运行唯一标识",
+    )
+    conversation_id: str = Field(..., index=True, description="会话唯一标识")
+    new_messages: str = Field(default="", sa_type=JSON, description="新增消息")
+
+
 class MessageRecord(SQLModel, table=True):
     """
     消息记录
@@ -101,22 +117,6 @@ class MessageRecord(SQLModel, table=True):
     content: str = Field(default="", description="消息内容")
 
 
-class RunRecord(SQLModel, table=True):
-    """
-    运行记录
-    """
-
-    id: str = Field(
-        default_factory=lambda: str(uuid7()),
-        primary_key=True,
-        description="运行唯一标识",
-    )
-    conversation_id: str = Field(..., index=True, description="会话唯一标识")
-    new_messages: str = Field(
-        default_factory=list, sa_type=JSON, description="新增消息"
-    )
-
-
 class DatabaseState(rx.State):
     """
     数据库状态
@@ -220,9 +220,13 @@ class DatabaseState(rx.State):
                     Conversation(
                         id=conversation_record.id,
                         description=conversation_record.description,
-                        usage=conversation_record.usage,
+                        usage=usage_validate_json(
-                        messages={},
+                            conversation_record.usage
+                        ),  # 将 conversation_record.usage 由 JSON 字符串反序列化为 RunUsage
                         created_at=conversation_record.created_at,
+                        workflow=_validate_json(
+                            conversation_record.workflow
+                        ),  # 将 conversation_record.workflow 由 JSON 字符串反序列化为 Workflow,
                     ),
                 )
                 if not message_record:
@@ -371,7 +375,8 @@ class DatabaseState(rx.State):
                     conversation_id=conversation_id,
                     new_messages=ModelMessagesTypeAdapter.dump_json(
                         new_messages
-                    ).decode(),  # 将 messages 由 List[ModelMessage] 序列化为 JSON 字符串
+                    ).decode(),
+                    # 将 messages 由 List[ModelMessage] 序列化为 JSON 字符串
                 )
             )
             await session.commit()

agent/application/states/models.py

@@ -12,28 +12,15 @@ from pydantic_ai._uuid import uuid7
 from pydantic_ai.usage import UsageLimits
 
 
-class TaskStatus(StrEnum):
+class AgentRunWarmUpEvent(BaseModel):
     """
-    任务状态枚举
+    智能体运行预热事件类
     """
 
-    RUNNING = "running"
+    content: str = Field(
-    DONE = "done"
+        default="",
-    PENDING_APPROVAL = "pending_approval"
+        description="预热内容",
-    ERROR = "error"
+    )
-
-
-class Task(BaseModel):
-    """
-    任务类
-    """
-
-    id: str = Field(default_factory=lambda: str(uuid7()), description="任务唯一标识")
-    tool_name: str = Field(..., description="工具名称")
-    tool_call_id: str = Field(..., description="工具调用唯一标识")
-    status: TaskStatus = Field(default=TaskStatus.RUNNING, description="任务状态")
-    title: str = Field(default="", description="任务标题")
-    content: str = Field(default="", description="任务内容")
 
 
 class MessageType(StrEnum):
@@ -57,7 +44,6 @@ class Message(BaseModel):
     type: MessageType = Field(..., description="消息类型")
     title: str = Field(default="", description="消息标题")
     content: str = Field(default="", description="消息内容")
-    tasks: list[Task] = Field(default_factory=list, description="任务列表")
     is_running: bool = Field(
         default=False, description="正在运行，True 表示正在运行， False 表示运行结束"
     )
@@ -66,18 +52,7 @@ class Message(BaseModel):
     )
 
 
-class AgentRunWarmUpEvent(BaseModel):
+class WorkflowType(StrEnum):
-    """
-    智能体运行预热事件类
-    """
-
-    content: str = Field(
-        default="",
-        description="预热内容",
-    )
-
-
-class WorkType(StrEnum):
     """
     工作类型枚举
     """
@@ -85,18 +60,13 @@ class WorkType(StrEnum):
     BOOK_FLIGHT = "预定航班"
 
 
-class Work(BaseModel):
+class Workflow(BaseModel):
     """
     工作流类
     """
 
-    type: WorkType = Field(..., description="工作类型")
+    type: WorkflowType = Field(..., description="工作类型")
-    deps: Any = Field(default=None, description="工作依赖项")
+    deps: dict[str, Any] = Field(default_factory=dict, description="工作依赖项")
-    usage: RunUsage = Field(default=RunUsage(), description="工作使用量")
-    usage_limits: UsageLimits | None = Field(
-        default=None,
-        description="工作使用量限制",
-    )
 
 
 class Conversation(BaseModel):
@@ -106,10 +76,9 @@ class Conversation(BaseModel):
 
     id: str = Field(..., description="会话唯一标识")
     description: str = Field(..., description="会话描述")
-    work: Work | None = Field(default=None, description="工作")
     user_prompt: str = Field(default="", description="用户提示词")
-    usage: dict[str, Any] = Field(..., description="会话使用量")
+    usage: RunUsage = Field(..., description="会话使用量")
-    messages: dict[str, Message] = Field(..., description="消息字典")
+    messages: dict[str, Message] = Field(default_factory=dict, description="消息字典")
     created_at: datetime = Field(..., description="会话创建日期时间")
     is_running: bool = Field(
         default=False, description="正在运行，True 表示正在运行， False 表示运行结束"
@@ -118,6 +87,7 @@ class Conversation(BaseModel):
         default=False,
         description="等待流式输出，True 表示等待流式输出，False 表示已开始流式输出或已完成",
     )
+    workflow: Workflow | None = Field(default=None, description="工作流")
 
 
 class ConversationHistoryItem(BaseModel):
@@ -136,27 +106,45 @@ class MessageHistoryItem(Message):
     """
 
 
-# Usage 适配器
+# 依赖项适配器
+DepsAdapter = TypeAdapter(Any)
+
+
+def deps_validate_python(deps: dict[str, Any]) -> Any:
+    """
+    将依赖项反序列化
+    """
+    if not deps:
+        return None
+    return DepsAdapter.validate_python(deps)
+
+
+def deps_dump_python(deps: Any) -> dict[str, Any]:
+    """
+    将依赖项序列化
+    :param deps: 依赖项
+    :return: python 字典
+    """
+    return DepsAdapter.dump_python(deps)
+
+
+# 使用量（RunUsage）适配器
 UsageAdapter = TypeAdapter(RunUsage)
 
 
-def usage_validate_python(usage: dict[str, Any]) -> RunUsage:
+def usage_validate_json(usage: str) -> RunUsage:
     """
-    将 Usage 反序列化
+    将使用量反序列化
     """
     if not usage:
         return RunUsage()
-    return UsageAdapter.validate_python(usage)
+    return UsageAdapter.validate_json(usage)
 
 
-def usage_dump_python(usage: RunUsage) -> dict[str, Any]:
+def usage_dump_json(usage: RunUsage) -> dict[str, Any]:
     """
-    将 Usage 序列化
+    将使用量序列化
     :param usage: 使用量
     :return: python 字典
     """
     return UsageAdapter.dump_python(usage)
-
-
-# UsageLimits 适配器
-UsageLimitsAdapter = TypeAdapter(UsageLimits)

agent/application/workshop/book_flight.py

@@ -2,11 +2,10 @@
 """
 预定航班工作流（范式）
 """
-import asyncio
 import datetime
 from typing import AsyncGenerator, cast
 
-from pydantic import BaseModel, Field, field_validator, TypeAdapter
+from pydantic import BaseModel, Field, field_validator
 from pydantic_ai import (
     Agent,
     ApprovalRequired,
@@ -14,112 +13,25 @@ from pydantic_ai import (
     DeferredToolResults,
     ModelMessage,
     ModelRetry,
-    ModelSettings,
     RunContext,
     RunUsage,
     UsageLimits,
 )
-from pydantic_ai.models.openai import OpenAIChatModel
-from pydantic_ai.providers.openai import OpenAIProvider
-from pydantic_ai.run import AgentRunResultEvent, AgentRunResult
-from enum import StrEnum
-from pydantic_ai._uuid import uuid7
 from pydantic_ai.messages import (
     AgentStreamEvent,
-    FunctionToolCallEvent,
     FunctionToolResultEvent,
-    LoadCapabilityCallPart,
-    PartDeltaEvent,
-    PartEndEvent,
     PartStartEvent,
-    TextPart,
-    TextPartDelta,
-    ThinkingPart,
-    ThinkingPartDelta,
     ToolCallPart,
-    ToolSearchCallPart,
     ToolReturnPart,
 )
+from pydantic_ai.run import AgentRunResult, AgentRunResultEvent
 
-
+from application.states.models import AgentRunWarmUpEvent
-class TaskStatus(StrEnum):
+from application.workshop.models import (
-    """
+    DEEPSEEK_V4_FLASH_MODEL,
-    任务状态枚举
+    DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING,
-    """
-
-    RUNNING = "running"
-    DONE = "done"
-    ERROR = "error"
-    PENDING_APPROVAL = "pending_approval"
-
-
-class Task(BaseModel):
-    """
-    任务类
-    """
-
-    tool_name: str = Field(..., description="工具名称")
-    tool_call_id: str = Field(..., description="工具调用唯一标识")
-    status: TaskStatus = Field(default=TaskStatus.RUNNING, description="任务状态")
-    title: str = Field(default="", description="任务标题")
-    content: str = Field(default="", description="任务内容")
-
-
-class MessageType(StrEnum):
-    """
-    消息类型枚举
-    """
-
-    USER_PROMPT = "user_prompt"
-    THINKING = "thinking"
-    WORK_OUTPUT = "work_output"
-    RESULT_OUTPUT = "result_output"
-    TEXT = "text"
-
-
-class AgentRunWarmUpEvent(BaseModel):
-    """
-    智能体运行预热事件类
-    """
-
-    content: str = Field(
-        default="",
-        description="预热内容",
-    )
-
-
-class Message(BaseModel):
-    """
-    消息类
-    """
-
-    id: str = Field(default_factory=lambda: str(uuid7()), description="消息唯一标识")
-    type: MessageType = Field(..., description="消息类型")
-    title: str = Field(default="", description="消息标题")
-    content: str = Field(default="", description="消息内容")
-    tasks: dict[str, Task] = Field(
-        default_factory=dict, description="任务字典，键为工具调用唯一标识"
-    )
-    is_running: bool = Field(
-        default=False, description="正在运行，True 表示正在运行，False 表示运行完成"
-    )
-    is_shown: bool = Field(
-        default=False, description="展示组件，True 表示展示，False 表示隐藏"
-    )
-
-
-DEEPSEEK_V4_FLASH_MODEL = OpenAIChatModel(
-    model_name="deepseek-v4-flash",
-    provider=OpenAIProvider(
-        base_url="https://tokenhub.tencentmaas.com/v1",
-        api_key="sk-D9Y1mCe8VlvNqLuSC4mAjqEwxJ2nW4C0h8a7EPn8kg9RLsHq",
-    ),
 )
 
-MODEL_SETTINGS = ModelSettings(
-    temperature=0, extra_body={"thinking": {"type": "disabled"}}  # 温度控制
-)  # 禁用思考模式
-
 
 class Flight(BaseModel):
     """
@@ -173,7 +85,7 @@ class Deps(BaseModel):
 # 主智能体
 agent = Agent[Deps, Flight | NoResult | DeferredToolRequests](
     model=DEEPSEEK_V4_FLASH_MODEL,
-    model_settings=MODEL_SETTINGS,
+    model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING,
     deps_type=Deps,
     output_type=Flight | NoResult | DeferredToolRequests,
     system_prompt=(
@@ -187,7 +99,7 @@ agent = Agent[Deps, Flight | NoResult | DeferredToolRequests](
 # 提取所有航班信息智能体
 extraction_flights_agent = Agent[Deps, list[Flight]](
     model=DEEPSEEK_V4_FLASH_MODEL,
-    model_settings=MODEL_SETTINGS,
+    model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING,
     deps_type=Deps,
     output_type=list[Flight],
     system_prompt=(
@@ -201,7 +113,7 @@ extraction_flights_agent = Agent[Deps, list[Flight]](
 # 提取航班号智能体
 extraction_flight_number_agent = Agent[Deps, str | None](
     model=DEEPSEEK_V4_FLASH_MODEL,
-    model_settings=MODEL_SETTINGS,
+    model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING,
     deps_type=Deps,
     output_type=str | None,
     system_prompt=(
@@ -358,21 +270,29 @@ async def validate_output(
     return output
 
 
+deps = Deps(
+    date=datetime.date(2025, 1, 10),
+    origin_airport_code="SFO",
+    destination_airport_code="ANC",
+)
+
+
 async def run_stream_events(
     deps: Deps,
     user_prompt: str,
     message_history: list[ModelMessage],
     usage: RunUsage,
+    usage_limits: UsageLimits | None = None,
     deferred_tool_results: DeferredToolResults | None = None,
 ) -> AsyncGenerator[
     AgentStreamEvent
-    | AgentRunResultEvent
+    | AgentRunResultEvent[str | DeferredToolRequests]
     | AgentRunWarmUpEvent
-    | Flight
+    | None,
-    | NoResult
-    | DeferredToolRequests,
-    None,
 ]:
+    """
+    运行并流式输出事件
+    """
     # 构建智能体运行预热事件
     yield AgentRunWarmUpEvent(
         content=f"正在预定 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班"
@@ -383,10 +303,10 @@ async def run_stream_events(
         deps=deps,
         message_history=message_history,
         usage=usage,
+        usage_limits=usage_limits,
         deferred_tool_results=deferred_tool_results,
     ) as events:
         async for event in events:
-            print(event)
             match event:
                 case PartStartEvent(
                     part=part,
@@ -416,7 +336,9 @@ async def run_stream_events(
 
                                 case "预定航班":
                                     if isinstance(content, NoResult):
-                                        event.part.content = "未查询到符合条件的航班"
+                                        event.part.content = (
+                                            "未提取到航班号，请检查后重试"
+                                        )
 
                                     if isinstance(content, Flight):
                                         event.part.content = f"已预定航班：\n{content.number} {content.airfare} 于 {content.date.strftime('%Y-%m-%d')} 从 {content.origin_airport_code} 到 {content.destination_airport_code}\n"
@@ -428,35 +350,15 @@ async def run_stream_events(
                     match result:
                         case AgentRunResult(output=output):
                             match output:
+                                case Flight():
+                                    yield AgentRunResultEvent(
+                                        result=AgentRunResult(output="预定成功")
+                                    )
                                 case DeferredToolRequests(approvals=approvals):
-                                    for approval in approvals:
+                                    yield AgentRunResultEvent(
-                                        output = approvals
+                                        result=AgentRunResult(
-                                        yield event
+                                            output=DeferredToolRequests(
-
+                                                approvals=approvals
-
+                                            )
-async def main():
+                                        )
-    # 实例化依赖
+                                    )
-    deps = Deps(
-        date=datetime.date(2025, 1, 10),
-        origin_airport_code="SFO",
-        destination_airport_code="ANC",
-    )
-    user_prompt = f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班"
-    message_history: list[ModelMessage] = []
-    events = run_stream_events(
-        deps=deps,
-        user_prompt=user_prompt,
-        message_history=message_history,
-        usage=RunUsage(),
-    )
-    # 消费异步生成器
-    async for event in events:
-        print()
-        if isinstance(event, AgentRunResultEvent):
-            # 保存本次运行新增消息
-            message_history.extend(event.result.new_messages())
-        if isinstance(event, Message):
-            print(event)
-
-
-asyncio.run(main())

agent/application/workshop/models.py

@@ -14,4 +14,6 @@ DEEPSEEK_V4_FLASH_MODEL = OpenAIChatModel(
     ),
 )
 
-MODEL_SETTINGS_DISABLED_THINKING = ModelSettings(extra_body={"thinking": {"type": "disabled"}})  # 禁用思考模式
+DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING = ModelSettings(
+    temperature=0, extra_body={"thinking": {"type": "disabled"}}  # 温度控制
+)  # 不启用思考模式