From d72cdf815df339d260ef8511100ef5d15c492257 Mon Sep 17 00:00:00 2001 From: liubiren Date: Wed, 19 Aug 2026 19:51:59 +0800 Subject: [PATCH] 1 --- agent/application/states/conversation.py | 188 ++++++------ agent/application/states/models.py | 3 +- agent/application/workshop/book_flight.py | 284 +++++------------- .../{unstructured_dialogue.py => talk.py} | 2 +- 4 files changed, 171 insertions(+), 306 deletions(-) rename agent/application/workshop/{unstructured_dialogue.py => talk.py} (98%) diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index 5abbf26..40142c1 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -29,6 +29,7 @@ from application.states.models import ( MessageType, Work, WorkType, + TaskStatus, ConversationHistoryItem, usage_validate_python, MessageHistoryItem, @@ -303,12 +304,10 @@ class ConversationState(rx.State): # 初始化片段索引映射为消息实例唯一标识字典 index_map_to_message_id: dict[int, str] = {} - # 初始化工具调用唯一标识集合 - tool_call_ids: set[str] = set() - # 非结构化对话 + # 对话 if not conversation.work: - from application.workshop.unstructured_dialogue import ( + from application.workshop.talk import ( run_stream_events, ) else: @@ -318,7 +317,6 @@ class ConversationState(rx.State): case WorkType.BOOK_FLIGHT: from application.workshop.book_flight import run_stream_events - # 消息列表 messages: list[Message] = [] # 运行并流式输出事件 @@ -329,108 +327,104 @@ class ConversationState(rx.State): ), usage=usage_validate_python(dict(conversation.usage)), ) - try: - # 获取运行流式输出事件 - async for event in stream_events: - # 将等待流式输出设置为否 - conversation.awaiting_stream = False - message: Message | None = None - match event: - # ========== 开始事件 ========== - case PartStartEvent( - index=index, - part=part, - ): - match part: - # 思考分片开始事件 - case ThinkingPart(content=content): - # 构建消息实例 - message = Message( - type=MessageType.THINKING, - title="正在思考", - content=content, - is_running=True, - ) - # 添加至消息字典 - conversation.messages[message.id] = message - # 将消息实例唯一标识与片段索引映射 - index_map_to_message_id[index] = message.id - yield # 通知前端更新渲染 + # 获取运行流式输出事件 + async for event in stream_events: + # 将等待流式输出设置为否 + conversation.awaiting_stream = False + message: Message | None = None + match event: + # ========== 开始事件 ========== + case PartStartEvent( + index=index, + part=part, + ): + match part: + # 思考分片开始事件 + case ThinkingPart(content=content): + # 构建消息实例 + message = Message( + type=MessageType.THINKING, + title="正在思考", + content=content, + is_running=True, + ) + # 添加至消息字典 + conversation.messages[message.id] = message + # 将消息实例唯一标识与片段索引映射 + index_map_to_message_id[index] = message.id + yield # 通知前端更新渲染 - # 文本分片开始事件 - case TextPart(content=content): - # 构建消息实例 - message = Message(type=MessageType.RESULT_OUTPUT) - message.content = content - # 添加至消息字典 - conversation.messages[message.id] = message - # 将消息实例唯一标识与片段索引映射 - index_map_to_message_id[index] = message.id - yield # 通知前端更新渲染 + # 文本分片开始事件 + case TextPart(content=content): + # 构建消息实例 + message = Message(type=MessageType.RESULT_OUTPUT) + message.content = content + # 添加至消息字典 + conversation.messages[message.id] = message + # 将消息实例唯一标识与片段索引映射 + index_map_to_message_id[index] = message.id + yield # 通知前端更新渲染 - # ========== 增量事件 ========== - case PartDeltaEvent(index=index, delta=delta): - match delta: - # 思考分片增量事件 - case ThinkingPartDelta( - content_delta=content_delta, - ): - # 若内容增量不为空则增量更新 - if content_delta: - conversation.messages[ - index_map_to_message_id[index] - ].content += content_delta - yield # 通知前端更新渲染 - - # 文本分片增量事件 - case TextPartDelta( - content_delta=content_delta, - ): - # 增量更新 + # ========== 增量事件 ========== + case PartDeltaEvent(index=index, delta=delta): + match delta: + # 思考分片增量事件 + case ThinkingPartDelta( + content_delta=content_delta, + ): + # 若内容增量不为空则增量更新 + if content_delta: conversation.messages[ index_map_to_message_id[index] ].content += content_delta yield # 通知前端更新渲染 - # ========== 结束事件 ========== - case PartEndEvent( - index=index, - part=part, - ): - match part: - # 思考分片结束事件 - case ThinkingPart(content=content): - # 获取消息实例 - message = conversation.messages[ - index_map_to_message_id[index] - ] - message.is_running = False - message.title = "思考完成" - yield # 通知前端更新渲染 - messages.append(message) + # 文本分片增量事件 + case TextPartDelta( + content_delta=content_delta, + ): + # 增量更新 + conversation.messages[ + index_map_to_message_id[index] + ].content += content_delta + yield # 通知前端更新渲染 - case TextPart(content=content): - # 获取消息实例 - message = conversation.messages[ - index_map_to_message_id[index] - ] - messages.append(message) + # ========== 结束事件 ========== + case PartEndEvent( + index=index, + part=part, + ): + match part: + # 思考分片结束事件 + case ThinkingPart(content=content): + # 获取消息实例 + message = conversation.messages[ + index_map_to_message_id[index] + ] + message.is_running = False + message.title = "思考完成" + yield # 通知前端更新渲染 + messages.append(message) - # ========== 智能体运行结果事件 ========== - case AgentRunResultEvent(result=result): - # 更新使用量 - await db_state.update_conversation_record( - conversation.id, usage=result.usage - ) - # 创建运行记录 - await db_state.create_run_record( - conversation_id=conversation.id, - new_messages=result.new_messages(), - ) + case TextPart(content=content): + # 获取消息实例 + message = conversation.messages[ + index_map_to_message_id[index] + ] + messages.append(message) + + # ========== 智能体运行结果事件 ========== + case AgentRunResultEvent(result=result): + # 更新使用量 + await db_state.update_conversation_record( + conversation.id, usage=result.usage + ) + # 创建运行记录 + await db_state.create_run_record( + conversation_id=conversation.id, + new_messages=result.new_messages(), + ) - except Exception as e: - ... - finally: # 将正在运行设置为否 conversation.is_running = False # 将等待流式输出设置为否 diff --git a/agent/application/states/models.py b/agent/application/states/models.py index 807433c..4d4b179 100644 --- a/agent/application/states/models.py +++ b/agent/application/states/models.py @@ -19,6 +19,7 @@ class TaskStatus(StrEnum): RUNNING = "running" DONE = "done" + PENDING_APPROVAL = "pending_approval" ERROR = "error" @@ -57,7 +58,7 @@ class Message(BaseModel): content: str = Field(default="", description="消息内容") tasks: list[Task] = Field(default_factory=list, description="任务列表") is_running: bool = Field( - default=False, description="正在运行，True 表示正在运行，False 表示运行完成" + default=False, description="正在运行，True 表示正在运行， False 表示运行结束" ) is_shown: bool = Field( default=False, description="展示组件，True 表示展示，False 表示隐藏" diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index 8d45c47..cabd794 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -21,7 +21,7 @@ from pydantic_ai import ( ) from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.providers.openai import OpenAIProvider -from pydantic_ai.run import AgentRunResultEvent +from pydantic_ai.run import AgentRunResultEvent, AgentRunResult from enum import StrEnum from pydantic_ai._uuid import uuid7 from pydantic_ai.messages import ( @@ -49,6 +49,7 @@ class TaskStatus(StrEnum): RUNNING = "running" DONE = "done" ERROR = "error" + PENDING_APPROVAL = "pending_approval" class Task(BaseModel): @@ -145,11 +146,11 @@ class Deps(BaseModel): date: datetime.date = Field(..., description="日期") origin_airport_code: str = Field(..., description="出发机场代码") destination_airport_code: str = Field(..., description="到达机场代码") - matched_flights: list[Flight] | None = Field( - default=None, description="匹配到的航班" + searched_flights: list[Flight] | None = Field( + default=None, description="查询到的航班" ) - selected_flight: Flight | NoSelectedFlight | None = Field( - default=None, description="选择到的航班" + recommended_flight: Flight | NoSelectedFlight | None = Field( + default=None, description="推荐航班" ) @@ -228,9 +229,9 @@ source_material = """ @agent.tool -async def match_flights(ctx: RunContext[Deps]) -> list[Flight]: +async def search_flights(ctx: RunContext[Deps]) -> list[Flight]: """ - 匹配满足用户需求的航班 + 查询航班 **如何提高提取准确率** 1. 系统提示词规则约束 2. 降低模型温度 @@ -238,7 +239,7 @@ async def match_flights(ctx: RunContext[Deps]) -> list[Flight]: 4. 业务规则约束（例如，输出不可为空列表） **设计工具需先设计业务流程** - 本示例按照 匹配满足用户需求的航班 -> 选择航班 -> 预定航班 + 本示例按照 查询航班 -> 推荐航班 -> 预定航班 """ extraction_agent_result = await extraction_agent.run( f"来源资料：\n{source_material}", @@ -254,35 +255,35 @@ async def match_flights(ctx: RunContext[Deps]) -> list[Flight]: raise ModelRetry( f"提取到的所有航班信息数量应为 8 ，不是 {extracted_flight_counts}" ) - # 匹配用户需求的航班 - ctx.deps.matched_flights = [ + # 查询到的航班 + ctx.deps.searched_flights = [ flight for flight in extracted_flights if flight.date == ctx.deps.date and flight.origin_airport_code == ctx.deps.origin_airport_code and flight.destination_airport_code == ctx.deps.destination_airport_code ] - return ctx.deps.matched_flights + return ctx.deps.searched_flights @agent.tool -async def select_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight: +async def recommend_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight: """ - 选择航班 + 推荐航班 **如何提高提取准确率** 1. 系统提示词规则约束 2. 降低模型温度 3. 输出类型约定和输出模型校验 4. 业务规则约束（例如，输出不可为空列表） """ - if (matched_flights := ctx.deps.matched_flights) is None: - raise ModelRetry("必须先使用 match_flights 匹配满足用户需求的航班") - ctx.deps.selected_flight = ( - min(matched_flights, key=lambda flight: flight.airfare) - if matched_flights + if (searched_flights := ctx.deps.searched_flights) is None: + raise ModelRetry("必须先使用 search_flights 查询航班") + ctx.deps.recommended_flight = ( + min(searched_flights, key=lambda flight: flight.airfare) + if searched_flights else NoSelectedFlight() ) - return ctx.deps.selected_flight + return ctx.deps.recommended_flight @agent.tool @@ -290,19 +291,15 @@ async def book_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight: """ 预定航班 """ - if (selected_flight := ctx.deps.selected_flight) is None: - raise ModelRetry("必须先使用 select_flight 选择航班") + if (recommended_flight := ctx.deps.recommended_flight) is None: + raise ModelRetry("必须先使用 recommend_flight 推荐航班") - if isinstance(selected_flight, NoSelectedFlight): - return selected_flight + if isinstance(recommended_flight, NoSelectedFlight): + return recommended_flight else: if not ctx.tool_call_approved: - raise ApprovalRequired( - metadata={ - "reason": f"需要用户确认是否预定 {selected_flight.number} 在 {selected_flight.date.strftime('%Y-%m-%d')} 从 {selected_flight.origin_airport_code} 到 {selected_flight.destination_airport_code} 的航班" - } - ) - return selected_flight + raise ApprovalRequired() + return recommended_flight @agent.output_validator @@ -339,10 +336,10 @@ async def run_stream_events( usage: RunUsage, deferred_tool_results: DeferredToolResults | None = None, ) -> AsyncGenerator: - # 构建工作输出消息 + # 构建消息 message = Message( type=MessageType.WORK_OUTPUT, - title="正在预定航班", + title="预定航班", is_running=True, ) yield message @@ -354,6 +351,7 @@ async def run_stream_events( deferred_tool_results=deferred_tool_results, ) as events: async for event in events: + print(event) match event: case PartStartEvent( part=part, @@ -363,7 +361,7 @@ async def run_stream_events( tool_name=tool_name, tool_call_id=tool_call_id ): match tool_name: - case "match_flights": + case "search_flights": if tool_name not in message.tasks: message.tasks[tool_name] = Task( tool_name=tool_name, @@ -372,15 +370,23 @@ async def run_stream_events( ) yield message - case "select_flight": + case "recommend_flight": if tool_name not in message.tasks: message.tasks[tool_name] = Task( tool_name=tool_name, tool_call_id=tool_call_id, - title="正在选择航班", + title="正在推荐航班", ) yield message + case "book_flight": + if tool_name not in message.tasks: + message.tasks[tool_name] = Task( + tool_name=tool_name, + tool_call_id=tool_call_id, + title="正在预定航班", + ) + yield message case FunctionToolResultEvent(part): match part: @@ -390,27 +396,52 @@ async def run_stream_events( content=content, ): match tool_name: - case "match_flights": - matched_flights = cast(list[Flight], content) + case "search_flights": task = message.tasks[tool_name] - task.title = f"已查询到 {len(matched_flights)} 班航班" + searched_flights = cast(list[Flight], content) + task.title = ( + f"已查询到 {len(searched_flights)} 班航班" + ) task.content = "\n".join( [ - f"{matched_flight.number} - {matched_flight.date.strftime('%Y-%m-%d')} - {matched_flight.origin_airport_code} ~ {matched_flight.destination_airport_code} ${matched_flight.airfare}" - for matched_flight in matched_flights + f"{searched_flight.number} - {searched_flight.date.strftime('%Y-%m-%d')} 从 {searched_flight.origin_airport_code} 飞往 {searched_flight.destination_airport_code} - {searched_flight.airfare}" + for searched_flight in searched_flights ] ) task.status = TaskStatus.DONE yield message - case "select_flight": - selected_flight = cast(Flight, content) + case "recommend_flight": task = message.tasks[tool_name] - task.title = f"已选择航班" - task.content = f"{selected_flight.number} - {selected_flight.date.strftime('%Y-%m-%d')} - {selected_flight.origin_airport_code} ~ {selected_flight.destination_airport_code} ${selected_flight.airfare}" + recommended_flight = cast( + Flight | NoSelectedFlight, content + ) + if isinstance(recommended_flight, Flight): + task.title = "为您推荐" + task.content = f"{recommended_flight.number} - {recommended_flight.date.strftime('%Y-%m-%d')} 从 {recommended_flight.origin_airport_code} 飞往 {recommended_flight.destination_airport_code} - {recommended_flight.airfare}" + else: + task.title = "为您推荐" + task.status = TaskStatus.DONE yield message + case AgentRunResultEvent( + result=result, + ): + match result: + case AgentRunResult(output=output): + match output: + case DeferredToolRequests(approvals=approvals): + for approval in approvals: + match approval: + case ToolCallPart(tool_name=tool_name): + task = message.tasks[tool_name] + task.content = "请确认是否预定 " + task.status = ( + TaskStatus.PENDING_APPROVAL + ) + yield message + async def main(): # 实例化依赖 @@ -418,7 +449,6 @@ async def main(): date=datetime.date(2025, 1, 10), origin_airport_code="SFO", destination_airport_code="ANC", - selected_flight=None, ) user_prompt = f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班" message_history: list[ModelMessage] = [] @@ -430,172 +460,12 @@ async def main(): ) # 消费异步生成器 async for event in events: - print(event) + print() if isinstance(event, AgentRunResultEvent): # 保存本次运行新增消息 message_history.extend(event.result.new_messages()) + if isinstance(event, Message): + print(event) asyncio.run(main()) - -""" -@selection_agent.tool -async def select_flight(ctx: RunContext[Deps]) -> Flight | FlightNoFound: - - result = await selection_agent.run(ctx.deps.source_material, usage=ctx.usage) - if isinstance(result.output, Flight): - if not ctx.tool_call_approved: - raise ApprovalRequired(metadata={"reason": "protected"}) - return result.output - -if not isinstance(event, AgentRunResultEvent): - yield event - else: - result = event.result - # 更新任务使用量 - task.usage = usage_to_dict(result.usage) - if isinstance(result.output, FlightDetail): - content = "\n---\n已查询到航班，请回复 buy 购票 或 search 重新查询\n" - # 更新任务节点为提取座位偏好 - task.node = "seat_preference_extraction" - else: - content = ( - "\n---\n未查询到满足您需求的航班，流程结束！\n" - ) - # 更新任务为空 - task = None - # 返回任务节点结果事件 - yield TaskNodeResultEvent( - task=task, - content=content, - ) - yield event - return - # 提取座位偏好 - case "seat_preference_extraction": - if user_prompt == "buy": - # 返回任务节点结果事件 - yield TaskNodeResultEvent( - task=task, - content="请和我说下您的座位偏好：\nA、F 座位是靠窗位；1 排、14 排、20 排腿部空间更大、更舒展，你更想要靠窗座位，宽敞大空间座位", - ) - return - - elif user_prompt == "search": - # 更新任务节点为航班查询 - task.node = "flight_search" - - else: -async with seat_preference_extraction_agent.run_stream_events( - user_prompt=user_prompt, - message_history=message_history, - usage=usage_to_object(dict(task.usage)), - usage_limits=usage_limits_to_object(dict(task.usage_limits)), - ) as events: - async for event in events: - if not isinstance(event, AgentRunResultEvent): - yield event - else: - result = event.result - # 更新任务使用量 - task.usage = usage_to_dict(result.usage) - # 更新任务为空 - task = None - # 返回任务节点结果事件 - yield TaskNodeResultEvent( - task=task, - content="已帮您定好座位，流程结束！", - ) - yield event - return - - - # 工具调用分片开始事件 - case ToolCallPart( - tool_name=tool_name, tool_call_id=tool_call_id - ): - # 构建消息实例 - message = Message( - type=MessageType.TOOL_CALL, - title=tool_name, - content=",", - is_running=True, - ) - tool_call_ids.add(tool_call_id) - # 添加至消息字典 - dialog.thoughts[index] = Thought( - type="tool_call", - content="正在生成调用参数", - ) - - # ========== 函数工具调用事件 ========== - case FunctionToolCallEvent(tool_call_id=tool_call_id, part=part): - # 获取分片索引 - index = tool_call_ids[tool_call_id] - match dialog.thoughts[index].type: - # 工具检索 - case "tool_search": - dialog.thoughts[index].content = ( - f"正在检索 {part.args_as_json_str()}" - ) - - # 能力加载 - case "capability_load": - dialog.thoughts[index].content = ( - f"正在加载能力 {part.tool_name}" - ) - - # 工具调用 - case "tool_call": - dialog.thoughts[index].content = ( - f"正在调用工具 {part.tool_name}" - ) - - case _: - continue - - - # ========== 函数工具结果事件 ========== - case FunctionToolResultEvent( - tool_call_id=tool_call_id, - content=content, - ): - index = tool_call_ids[tool_call_id] - match dialog.thoughts[index].type: - # 工具检索 - case "tool_search": - dialog.thoughts[index].content = ( - content if isinstance(content, str) else "" - ) # 暂仅考虑文本内容 - - # 能力加载 - case "capability_load": - dialog.thoughts[index].content = f"已加载 {content}" - - # 工具调用 - case "tool_call": - dialog.thoughts[index].content = f"已调用 {content}" - - case _: - continue - - - def init_work_flow() -> WorkFlow: - return WorkFlow( - type=WorkFlowType.BOOK_FLIGHT, - deps=Dependences( - source_material=source_material, - date=datetime.date(2025, 1, 10), - origin="SFO", - destination="ANC", - ), - ) -from application.states.models import ( - WorkFlow, - WorkFlowType, -) -from application.workshop.models import ( - DEEPSEEK_V4_FLASH_MODEL, - MODEL_SETTINGS_DISABLED_THINKING, -) -""" diff --git a/agent/application/workshop/unstructured_dialogue.py b/agent/application/workshop/talk.py similarity index 98% rename from agent/application/workshop/unstructured_dialogue.py rename to agent/application/workshop/talk.py index e3aea77..0352ee1 100644 --- a/agent/application/workshop/unstructured_dialogue.py +++ b/agent/application/workshop/talk.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- """ -非结构化对话 +对话 """ from typing import AsyncGenerator