From 7df4e51bbbc8d3a59f92c4f50ca618854c6c83fb Mon Sep 17 00:00:00 2001 From: liubiren Date: Tue, 1 Sep 2026 20:25:42 +0800 Subject: [PATCH] 1 --- agent/application/pages/conversation.py | 7 +- agent/application/states/conversation.py | 91 +++++++----- agent/application/states/database.py | 49 +++--- agent/application/states/models.py | 94 ++++++------ agent/application/workshop/book_flight.py | 172 +++++----------------- agent/application/workshop/models.py | 4 +- 6 files changed, 166 insertions(+), 251 deletions(-) diff --git a/agent/application/pages/conversation.py b/agent/application/pages/conversation.py index 790f83d..6f02cdb 100644 --- a/agent/application/pages/conversation.py +++ b/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)", diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index e27c980..e51be98 100644 --- a/agent/application/states/conversation.py +++ b/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: - # 预定航班 - case WorkType.BOOK_FLIGHT: - from application.workshop.book_flight import run_stream_events + + # 运行并流式输出事件 + 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)), + ) # 消息列表 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 - message: Message | None = None + # 若等待流式输出则将等待流式输出设置为否 + if conversation.awaiting_stream: + 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: - # 预定航班工作 - case WorkType.BOOK_FLIGHT: - from application.workshop.book_flight import init_work + match type: + case WorkflowType.BOOK_FLIGHT: + from application.workshop.book_flight import deps - # 初始化预定航班任务 - 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: diff --git a/agent/application/states/database.py b/agent/application/states/database.py index 8d4246c..e0b6382 100644 --- a/agent/application/states/database.py +++ b/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( - default_factory=dict, sa_type=JSON, description="会话使用量" + usage: str = Field(default="", 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, - messages={}, + usage=usage_validate_json( + 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() diff --git a/agent/application/states/models.py b/agent/application/states/models.py index 5818eab..77d0463 100644 --- a/agent/application/states/models.py +++ b/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" - DONE = "done" - PENDING_APPROVAL = "pending_approval" - 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="任务内容") + 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): - """ - 智能体运行预热事件类 - """ - - content: str = Field( - default="", - description="预热内容", - ) - - -class WorkType(StrEnum): +class WorkflowType(StrEnum): """ 工作类型枚举 """ @@ -85,18 +60,13 @@ class WorkType(StrEnum): BOOK_FLIGHT = "预定航班" -class Work(BaseModel): +class Workflow(BaseModel): """ 工作流类 """ - type: WorkType = Field(..., description="工作类型") - deps: Any = Field(default=None, description="工作依赖项") - usage: RunUsage = Field(default=RunUsage(), description="工作使用量") - usage_limits: UsageLimits | None = Field( - default=None, - description="工作使用量限制", - ) + type: WorkflowType = Field(..., description="工作类型") + deps: dict[str, Any] = Field(default_factory=dict, 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="会话使用量") - messages: dict[str, Message] = Field(..., description="消息字典") + usage: RunUsage = 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) diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index e8fe701..90f0e84 100644 --- a/agent/application/workshop/book_flight.py +++ b/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 - -class TaskStatus(StrEnum): - """ - 任务状态枚举 - """ - - 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", - ), +from application.states.models import AgentRunWarmUpEvent +from application.workshop.models import ( + DEEPSEEK_V4_FLASH_MODEL, + DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING, ) -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 - | NoResult - | DeferredToolRequests, - None, + | 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: - output = approvals - yield event - - -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()) + yield AgentRunResultEvent( + result=AgentRunResult( + output=DeferredToolRequests( + approvals=approvals + ) + ) + ) diff --git a/agent/application/workshop/models.py b/agent/application/workshop/models.py index 9f018c1..2de7b9a 100644 --- a/agent/application/workshop/models.py +++ b/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"}} # 温度控制 +) # 不启用思考模式