diff --git a/agent/alembic/versions/3a74f94a7ad4_.py b/agent/alembic/versions/3a74f94a7ad4_.py new file mode 100644 index 0000000..413e5a8 --- /dev/null +++ b/agent/alembic/versions/3a74f94a7ad4_.py @@ -0,0 +1,36 @@ +"""empty message + +Revision ID: 3a74f94a7ad4 +Revises: d52f1ee0f9c4 +Create Date: 2026-09-03 12:30:13.538072 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel + +# revision identifiers, used by Alembic. +revision: str = '3a74f94a7ad4' +down_revision: Union[str, Sequence[str], None] = 'd52f1ee0f9c4' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('conversationrecord', schema=None) as batch_op: + batch_op.add_column(sa.Column('workflow', sqlmodel.sql.sqltypes.AutoString(), server_default=sa.text("('')"), nullable=False)) + + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('conversationrecord', schema=None) as batch_op: + batch_op.drop_column('workflow') + + # ### end Alembic commands ### diff --git a/agent/application/pages/conversation.py b/agent/application/pages/conversation.py index 6f02cdb..e87a1c8 100644 --- a/agent/application/pages/conversation.py +++ b/agent/application/pages/conversation.py @@ -418,9 +418,9 @@ def thinking(item: MessageHistoryItem) -> rx.Component: ) -def result_output(item: MessageHistoryItem) -> rx.Component: +def output(item: MessageHistoryItem) -> rx.Component: """ - 结果输出 + 输出 :return: Component """ return rx.markdown( @@ -550,7 +550,7 @@ def message_history_item(item: MessageHistoryItem) -> rx.Component: item.type, (MessageType.USER_PROMPT, user_prompt(item)), (MessageType.THINKING, thinking(item)), - (MessageType.RESULT_OUTPUT, result_output(item)), + (MessageType.OUTPUT, output(item)), ), padding="0 16px", width="100%", diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index e51be98..897da2b 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -3,12 +3,13 @@ 会话状态 """ from datetime import datetime -from typing import AsyncGenerator +from typing import AsyncGenerator, cast from pydantic_ai.messages import ( FunctionToolCallEvent, FunctionToolResultEvent, LoadCapabilityCallPart, + ToolReturnPart, PartDeltaEvent, PartEndEvent, PartStartEvent, @@ -24,7 +25,7 @@ import reflex as rx from application.states.database import DatabaseState from application.states.models import ( - AgentRunWarmUpEvent, + AgentRunEvent, Conversation, ConversationHistoryItem, Message, @@ -32,9 +33,9 @@ from application.states.models import ( MessageType, Workflow, WorkflowType, - deps_dump_python, - deps_validate_python, - usage_validate_python, + workflow_dump_json, + workflow_validate_json, + usage_validate_json, ) @@ -306,6 +307,8 @@ class ConversationState(rx.State): # 初始化片段索引映射为消息实例唯一标识字典 index_map_to_message_id: dict[int, str] = {} + # 初始化工具调用片段工具名称映射为消息实例唯一标识字典 + tool_name_map_to_message_id: dict[str, str] = {} # 初始化工作流 if conversation.workflow: @@ -317,12 +320,12 @@ class ConversationState(rx.State): # 运行并流式输出事件 stream_events = run_stream_events( - deps=deps_validate_python(dict(conversation.workflow.deps)), + deps=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)), + usage=conversation.usage, ) else: from application.workshop.talk import ( @@ -335,7 +338,7 @@ class ConversationState(rx.State): message_history=await db_state.get_message_history( conversation_id=self.actived_conversation_id ), - usage=usage_validate_python(dict(conversation.usage)), + usage=conversation.usage, ) # 消息列表 @@ -352,7 +355,7 @@ class ConversationState(rx.State): part=part, ): match part: - # 思考分片开始事件 + # 思考分片 case ThinkingPart(content=content): # 构建消息实例 message = Message( @@ -367,10 +370,24 @@ class ConversationState(rx.State): index_map_to_message_id[index] = message.id yield # 通知前端更新渲染 - # 文本分片开始事件 + # 工具调用分片 + case ToolCallPart(tool_name=tool_name): + # 构建消息实例 + message = Message( + type=MessageType.TOOL_CALL, + title=f"正在{tool_name}", + is_running=True, + ) + # 添加至消息字典 + conversation.messages[message.id] = message + # 将消息实例唯一标识与工具调用片段工具名称映射 + tool_name_map_to_message_id[tool_name] = message.id + yield # 通知前端更新渲染 + + # 文本分片 case TextPart(content=content): # 构建消息实例 - message = Message(type=MessageType.TEXT, content=content) + message = Message(type=MessageType.OUTPUT, content=content) # 添加至消息字典 conversation.messages[message.id] = message # 将消息实例唯一标识与片段索引映射 @@ -380,7 +397,7 @@ class ConversationState(rx.State): # ========== 增量事件 ========== case PartDeltaEvent(index=index, delta=delta): match delta: - # 思考分片增量事件 + # 思考分片增量 case ThinkingPartDelta( content_delta=content_delta, ): @@ -391,7 +408,7 @@ class ConversationState(rx.State): ].content += content_delta yield # 通知前端更新渲染 - # 文本分片增量事件 + # 文本分片增量 case TextPartDelta( content_delta=content_delta, ): @@ -407,17 +424,18 @@ class ConversationState(rx.State): part=part, ): match part: - # 思考分片结束事件 + # 思考分片 case ThinkingPart(content=content): # 获取消息实例 message = conversation.messages[ index_map_to_message_id[index] ] message.is_running = False - message.title = "思考完成" + message.title = "思考已完成" yield # 通知前端更新渲染 messages.append(message) + # 文本分片 case TextPart(content=content): # 获取消息实例 message = conversation.messages[ @@ -425,11 +443,27 @@ class ConversationState(rx.State): ] messages.append(message) - # ========== 智能体运行预热事件 ========== - case AgentRunWarmUpEvent(content=content): + # ========== 工具调用结果事件 ========== + case FunctionToolResultEvent(part=part): + match part: + # 工具返回分片 + case ToolReturnPart(tool_name=tool_name, content=content): + # 获取消息实例 + message = conversation.messages[ + tool_name_map_to_message_id[tool_name] + ] + message.title = f"{tool_name}已完成" + message.content = cast( + str, content + ) # 约定工具返回分片内容必为字符串 + yield # 通知前端更新渲染 + messages.append(message) + + # ========== 智能体运行事件 ========== + case AgentRunEvent(content=content): # 构建消息实例 message = Message( - type=MessageType.TEXT, + type=MessageType.OUTPUT, content=content, ) # 添加至消息字典 @@ -439,9 +473,11 @@ class ConversationState(rx.State): # ========== 智能体运行结果事件 ========== case AgentRunResultEvent(result=result): - # 更新使用量 + # 更新会话记录 await db_state.update_conversation_record( - conversation.id, usage=result.usage + conversation.id, + usage=result.usage, + workflow=conversation.workflow, ) # 创建运行记录 await db_state.create_run_record( @@ -449,18 +485,18 @@ class ConversationState(rx.State): new_messages=result.new_messages(), ) - # 将正在运行设置为否 - conversation.is_running = False - # 将等待流式输出设置为否 - conversation.awaiting_stream = False - self.conversations[self.actived_conversation_id] = conversation - yield + # 将正在运行设置为否 + conversation.is_running = False + # 将等待流式输出设置为否 + conversation.awaiting_stream = False + self.conversations[self.actived_conversation_id] = conversation + yield - # 批量创建消息记录 - await db_state.create_message_records( - conversation_id=conversation.id, - messages=messages, - ) + # 批量创建消息记录 + await db_state.create_message_records( + conversation_id=conversation.id, + messages=messages, + ) @rx.event async def set_workflow(self, type: WorkflowType) -> None: @@ -477,7 +513,7 @@ class ConversationState(rx.State): from application.workshop.book_flight import deps # 初始化预定航班工作流 - workflow = Workflow(type=type, deps=deps_dump_python(deps)) + workflow = Workflow(type=type, deps=deps) # 设置用户提示词 self.set_user_prompt( f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班" diff --git a/agent/application/states/database.py b/agent/application/states/database.py index e0b6382..44baf05 100644 --- a/agent/application/states/database.py +++ b/agent/application/states/database.py @@ -4,8 +4,7 @@ """ from datetime import datetime, timedelta from random import choices -from typing import Any - +from dataclasses import asdict from pydantic_ai import ModelMessage, ModelMessagesTypeAdapter, RunUsage from pydantic_ai._uuid import uuid7 import reflex as rx @@ -15,7 +14,11 @@ from application.states.models import ( Conversation, Message, MessageType, + Workflow, + usage_dump_json, usage_validate_json, + workflow_dump_json, + workflow_validate_json, ) @@ -72,19 +75,19 @@ class ConversationRecord(SQLModel, table=True): ) user_id: str = Field(..., index=True, description="用户唯一标识") description: str = Field(default="新会话", description="会话描述") - usage: str = Field(default="", description="会话使用量") - workflow: str = Field( - default="", - description="会话工作流", + created_at: datetime = Field( + default_factory=datetime.now, description="会话创建时间" ) is_deleted: bool = Field( default=False, index=True, description="会话已删除：True 表示已删除，False 表示未删除", ) - created_at: datetime = Field( - default_factory=datetime.now, description="会话创建时间" - ) + usage: str = Field(default="", description="会话使用量") + workflow: str = Field( + default="", + description="会话工作流", + ) # 原则上嵌套列表或字典若不涉及查询则以字符串储存 class RunRecord(SQLModel, table=True): @@ -224,7 +227,7 @@ class DatabaseState(rx.State): conversation_record.usage ), # 将 conversation_record.usage 由 JSON 字符串反序列化为 RunUsage created_at=conversation_record.created_at, - workflow=_validate_json( + workflow=workflow_validate_json( conversation_record.workflow ), # 将 conversation_record.workflow 由 JSON 字符串反序列化为 Workflow, ), @@ -257,7 +260,9 @@ class DatabaseState(rx.State): record.id: Conversation( id=record.id, description=record.description, - usage=record.usage, + usage=usage_validate_json( + record.usage + ), # 将 record.usage 由 JSON 字符串反序列化为 RunUsage messages={}, # 新会话默认消息字典为空 created_at=record.created_at, ) @@ -269,6 +274,7 @@ class DatabaseState(rx.State): description: str | None = None, is_deleted: bool | None = None, usage: RunUsage | None = None, + workflow: Workflow | None = None, ) -> None: """ 更新会话记录 @@ -276,6 +282,7 @@ class DatabaseState(rx.State): :param description: 会话描述 :param is_deleted: 会话已删除 :param usage: 使用量 + :param workflow: 工作流 :return: None """ async with rx.asession() as session: @@ -287,7 +294,13 @@ class DatabaseState(rx.State): if isinstance(is_deleted, bool): record.is_deleted = is_deleted if isinstance(usage, RunUsage): - record.usage = usage_dump_python(usage) + record.usage = usage_dump_json( + RunUsage(**asdict(usage)) + ) # 将 RunUsage 序列化为 JSON 字符串 + if isinstance(workflow, Workflow): + record.workflow = workflow_dump_json( + Workflow(**asdict(workflow)) + ) # 将 Workflow 序列化为 JSON 字符串 await session.commit() async def create_message_record( @@ -375,7 +388,7 @@ class DatabaseState(rx.State): conversation_id=conversation_id, new_messages=ModelMessagesTypeAdapter.dump_json( new_messages - ).decode(), + ).decode("utf-8"), # 将 messages 由 List[ModelMessage] 序列化为 JSON 字符串 ) ) diff --git a/agent/application/states/models.py b/agent/application/states/models.py index 77d0463..1afeb88 100644 --- a/agent/application/states/models.py +++ b/agent/application/states/models.py @@ -9,17 +9,24 @@ from typing import Any from pydantic import BaseModel, Field, TypeAdapter from pydantic_ai import RunUsage from pydantic_ai._uuid import uuid7 -from pydantic_ai.usage import UsageLimits -class AgentRunWarmUpEvent(BaseModel): + +class NoResult(BaseModel): """ - 智能体运行预热事件类 + 无结果类 + """ + + ... + +class AgentRunEvent(BaseModel): + """ + 智能体运行事件类 """ content: str = Field( default="", - description="预热内容", + description="智能体运行内容", ) @@ -30,9 +37,8 @@ class MessageType(StrEnum): USER_PROMPT = "user_prompt" THINKING = "thinking" - WORK = "work" - RESULT_OUTPUT = "result_output" - TEXT = "text" + TOOL_CALL = "tool_call" + OUTPUT = "output" class Message(BaseModel): @@ -66,7 +72,7 @@ class Workflow(BaseModel): """ type: WorkflowType = Field(..., description="工作类型") - deps: dict[str, Any] = Field(default_factory=dict, description="工作依赖项") + deps: Any = Field(default=None, description="工作依赖项") class Conversation(BaseModel): @@ -76,10 +82,11 @@ class Conversation(BaseModel): id: str = Field(..., description="会话唯一标识") description: str = Field(..., description="会话描述") - user_prompt: str = Field(default="", description="用户提示词") + created_at: datetime = Field(..., description="会话创建日期时间") usage: RunUsage = Field(..., description="会话使用量") messages: dict[str, Message] = Field(default_factory=dict, description="消息字典") - created_at: datetime = Field(..., description="会话创建日期时间") + workflow: Workflow | None = Field(default=None, description="工作流") + user_prompt: str = Field(default="", description="用户提示词") is_running: bool = Field( default=False, description="正在运行，True 表示正在运行， False 表示运行结束" ) @@ -87,7 +94,6 @@ class Conversation(BaseModel): default=False, description="等待流式输出，True 表示等待流式输出，False 表示已开始流式输出或已完成", ) - workflow: Workflow | None = Field(default=None, description="工作流") class ConversationHistoryItem(BaseModel): @@ -106,26 +112,26 @@ class MessageHistoryItem(Message): """ -# 依赖项适配器 -DepsAdapter = TypeAdapter(Any) +# 工作流适配器 +WorkflowAdapter = TypeAdapter(Any) -def deps_validate_python(deps: dict[str, Any]) -> Any: +def workflow_validate_json(workflow: str) -> Any: """ - 将依赖项反序列化 + 将工作流反序列化 """ - if not deps: + if not workflow: return None - return DepsAdapter.validate_python(deps) + return WorkflowAdapter.validate_json(workflow) -def deps_dump_python(deps: Any) -> dict[str, Any]: +def workflow_dump_json(workflow: Any) -> str: """ - 将依赖项序列化 - :param deps: 依赖项 - :return: python 字典 + 将工作流序列化 + :param workflow: 工作流 + :return: 字符串 """ - return DepsAdapter.dump_python(deps) + return WorkflowAdapter.dump_json(workflow).decode("utf-8") # 使用量（RunUsage）适配器 @@ -141,10 +147,10 @@ def usage_validate_json(usage: str) -> RunUsage: return UsageAdapter.validate_json(usage) -def usage_dump_json(usage: RunUsage) -> dict[str, Any]: +def usage_dump_json(usage: RunUsage) -> str: """ 将使用量序列化 :param usage: 使用量 - :return: python 字典 + :return: 字符串 """ - return UsageAdapter.dump_python(usage) + return UsageAdapter.dump_json(usage).decode("utf-8") diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index 90f0e84..85a2a9c 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -3,8 +3,9 @@ 预定航班工作流（范式） """ import datetime -from typing import AsyncGenerator, cast - +from re import I +from typing import AsyncGenerator, cast, Any +from dataclasses import replace from pydantic import BaseModel, Field, field_validator from pydantic_ai import ( Agent, @@ -26,10 +27,10 @@ from pydantic_ai.messages import ( ) from pydantic_ai.run import AgentRunResult, AgentRunResultEvent -from application.states.models import AgentRunWarmUpEvent +from application.states.models import AgentRunEvent, NoResult from application.workshop.models import ( DEEPSEEK_V4_FLASH_MODEL, - DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING, + DEEPSEEK_V4_FLASH_MODEL_SETTINGS, ) @@ -57,14 +58,6 @@ class Flight(BaseModel): return airport_code -class NoResult(BaseModel): - """ - 无结果类 - """ - - ... - - class Deps(BaseModel): """ 依赖类 @@ -76,16 +69,13 @@ class Deps(BaseModel): searched_flights: list[Flight] | None = Field( default=None, description="查询到的航班" ) - extracted_flight_number: str | None = Field( - default=None, description="提取到的航班号" - ) booked_flight: Flight | None = Field(default=None, description="预定到的航班") # 主智能体 agent = Agent[Deps, Flight | NoResult | DeferredToolRequests]( model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING, + model_settings=DEEPSEEK_V4_FLASH_MODEL_SETTINGS, deps_type=Deps, output_type=Flight | NoResult | DeferredToolRequests, system_prompt=( @@ -99,7 +89,7 @@ agent = Agent[Deps, Flight | NoResult | DeferredToolRequests]( # 提取所有航班信息智能体 extraction_flights_agent = Agent[Deps, list[Flight]]( model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING, + model_settings=DEEPSEEK_V4_FLASH_MODEL_SETTINGS, deps_type=Deps, output_type=list[Flight], system_prompt=( @@ -113,7 +103,7 @@ extraction_flights_agent = Agent[Deps, list[Flight]]( # 提取航班号智能体 extraction_flight_number_agent = Agent[Deps, str | None]( model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING, + model_settings=DEEPSEEK_V4_FLASH_MODEL_SETTINGS, deps_type=Deps, output_type=str | None, system_prompt=( @@ -195,7 +185,7 @@ async def search_flights(ctx: RunContext[Deps]) -> list[Flight]: raise ModelRetry(f"提取到的所有航班信息不可能为空") # 查询到的航班 - ctx.deps.searched_flights = sorted( + searched_flights = sorted( [ flight for flight in extracted_flights @@ -205,10 +195,14 @@ async def search_flights(ctx: RunContext[Deps]) -> list[Flight]: ], key=lambda flight: flight.airfare, ) - return ctx.deps.searched_flights + if not searched_flights: # 模拟业务规则约束：查询到的航班不能为空 + raise ModelRetry("查询到的航班不能为空") + + ctx.deps.searched_flights = searched_flights + return searched_flights -@agent.tool(name="预定航班") +@agent.tool(name="预定航班", requires_approval=True) async def book_flight(ctx: RunContext[Deps]) -> Flight | NoResult: """ 预定航班 @@ -220,25 +214,24 @@ async def book_flight(ctx: RunContext[Deps]) -> Flight | NoResult: raise ApprovalRequired() # 提取到的航班号 - ctx.deps.extracted_flight_number = ( + extracted_flight_number = ( await extraction_flight_number_agent.run( - f"用户提示词：\n{ctx.prompt}\n查询到的航班：\n{(flight_numbers:= [flight.number for flight in searched_flights])}", + f"用户提示词：\n{cast(str, ctx.prompt).upper()}\n查询到的航班：\n{(flight_numbers:= [flight.number for flight in searched_flights])}", deps=ctx.deps, usage=ctx.usage, usage_limits=UsageLimits(request_limit=10), ) ).output - if not ctx.deps.extracted_flight_number: + if not extracted_flight_number: return NoResult() - if ctx.deps.extracted_flight_number not in flight_numbers: - raise ModelRetry( - f"提取到的航班号 {ctx.deps.extracted_flight_number} 不在查询到的航班中" - ) + + if extracted_flight_number not in flight_numbers: + raise ModelRetry(f"提取到的航班号 {extracted_flight_number} 不在查询到的航班中") # 预定到的航班 ctx.deps.booked_flight = next( flight for flight in searched_flights - if flight.number == ctx.deps.extracted_flight_number + if flight.number == extracted_flight_number ) return ctx.deps.booked_flight @@ -278,7 +271,7 @@ deps = Deps( async def run_stream_events( - deps: Deps, + deps: Any, user_prompt: str, message_history: list[ModelMessage], usage: RunUsage, @@ -286,15 +279,15 @@ async def run_stream_events( deferred_tool_results: DeferredToolResults | None = None, ) -> AsyncGenerator[ AgentStreamEvent - | AgentRunResultEvent[str | DeferredToolRequests] - | AgentRunWarmUpEvent + | AgentRunResultEvent[Flight | NoResult | DeferredToolRequests] + | AgentRunEvent | None, ]: """ 运行并流式输出事件 """ - # 构建智能体运行预热事件 - yield AgentRunWarmUpEvent( + # 构建智能体运行事件 + yield AgentRunEvent( content=f"正在预定 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班" ) tool_names = set() @@ -351,9 +344,8 @@ async def run_stream_events( case AgentRunResult(output=output): match output: case Flight(): - yield AgentRunResultEvent( - result=AgentRunResult(output="预定成功") - ) + yield AgentRunEvent(content="预定成功") + case DeferredToolRequests(approvals=approvals): yield AgentRunResultEvent( result=AgentRunResult( @@ -362,3 +354,4 @@ async def run_stream_events( ) ) ) + yield event diff --git a/agent/application/workshop/models.py b/agent/application/workshop/models.py index 2de7b9a..373f4b8 100644 --- a/agent/application/workshop/models.py +++ b/agent/application/workshop/models.py @@ -14,6 +14,6 @@ DEEPSEEK_V4_FLASH_MODEL = OpenAIChatModel( ), ) -DEEPSEEK_V4_FLASH_MODEL_DISABLED_THINKING = ModelSettings( +DEEPSEEK_V4_FLASH_MODEL_SETTINGS = ModelSettings( temperature=0, extra_body={"thinking": {"type": "disabled"}} # 温度控制 ) # 不启用思考模式 diff --git a/agent/database.db b/agent/database.db index 74f44ce..5f320ec 100644 Binary files a/agent/database.db and b/agent/database.db differ