diff --git a/agent/application/states/conversation.py b/agent/application/states/conversation.py index a9cdd0e..aed9306 100644 --- a/agent/application/states/conversation.py +++ b/agent/application/states/conversation.py @@ -279,10 +279,10 @@ class ConversationState(rx.State): conversation.is_running = True # 将等待流式输出设置为是 conversation.awaiting_stream = True + yield # 通知前端更新渲染 # 获取数据库状态 db_state = await self.get_db_state() - # 先创建消息记录再添加消息实例 conversation.messages.update( await db_state.create_message_record( @@ -350,6 +350,7 @@ class ConversationState(rx.State): conversation.messages[message.id] = message # 将消息实例唯一标识与片段索引映射 index_map_to_message_id[index] = message.id + yield # 通知前端更新渲染 # 文本分片开始事件 case TextPart(content=content): @@ -360,6 +361,7 @@ class ConversationState(rx.State): conversation.messages[message.id] = message # 将消息实例唯一标识与片段索引映射 index_map_to_message_id[index] = message.id + yield # 通知前端更新渲染 # ========== 增量事件 ========== case PartDeltaEvent(index=index, delta=delta): @@ -373,6 +375,7 @@ class ConversationState(rx.State): conversation.messages[ index_map_to_message_id[index] ].content += content_delta + yield # 通知前端更新渲染 # 文本分片增量事件 case TextPartDelta( @@ -382,6 +385,7 @@ class ConversationState(rx.State): conversation.messages[ index_map_to_message_id[index] ].content += content_delta + yield # 通知前端更新渲染 # ========== 结束事件 ========== case PartEndEvent( @@ -397,6 +401,7 @@ class ConversationState(rx.State): ] message.is_running = False message.title = "思考完成" + yield # 通知前端更新渲染 messages.append(message) case TextPart(content=content): @@ -404,7 +409,6 @@ class ConversationState(rx.State): message = conversation.messages[ index_map_to_message_id[index] ] - message.content = content messages.append(message) # ========== 智能体运行结果事件 ========== @@ -419,10 +423,6 @@ class ConversationState(rx.State): new_messages=result.new_messages(), ) - # 强制更新会话并推送前端 - self.conversations[self.actived_conversation_id] = conversation - yield - except Exception as e: ... finally: diff --git a/agent/application/workshop/book_flight.py b/agent/application/workshop/book_flight.py index d862633..7c432b4 100644 --- a/agent/application/workshop/book_flight.py +++ b/agent/application/workshop/book_flight.py @@ -2,29 +2,52 @@ """ 预定航班（范式） """ +import asyncio import datetime -from typing import AsyncGenerator, Literal +from typing import AsyncGenerator -from pydantic import BaseModel, Field -from pydantic_ai import Agent, ModelMessage, ModelRetry, RunContext, UsageLimits, RunUsage, DeferredToolRequests -from pydantic_ai.run import AgentRunResultEvent - -from application.states.models import ( - WorkFlow, - WorkFlowType, +from logfire_api.variables import ValueDoesNotEqual +from pydantic import BaseModel, Field, field_validator +from pydantic_ai import ( + Agent, + ApprovalRequired, + DeferredToolRequests, + DeferredToolResults, + ModelMessage, + ModelRetry, + ModelSettings, + RunContext, + RunUsage, + UsageLimits, ) -from application.workshop.models import DEEPSEEK_V4_FLASH_MODEL, MODEL_SETTINGS_DISABLED_THINKING +from pydantic_ai.models.openai import OpenAIChatModel +from pydantic_ai.providers.openai import OpenAIProvider +from sqlalchemy.sql.dml import ReturningDelete -class Dependences(BaseModel): +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 Deps(BaseModel): """ 依赖类 """ - date: datetime.date = Field(..., description="航班日期") - origin: str = Field(..., description="航班出发机场") - destination: str = Field(..., description="航班到达机场") - source_material: str = Field(..., description="航班来源资料") + date: datetime.date = Field(..., description="日期") + origin_airport_code: str = Field(..., description="出发机场代码") + destination_airport_code: str = Field(..., description="到达机场代码") + source_material: str = Field(..., description="来源资料") + class Flight(BaseModel): """ @@ -32,10 +55,22 @@ class Flight(BaseModel): """ number: str = Field(..., description="航班号") - date: datetime.date = Field(..., description="航班日期") - origin: str = Field(..., description="航班出发机场") - destination: str = Field(..., description="航班到达机场") - airfare: int = Field(..., description="航班机票价格") + date: datetime.date = Field(..., description="日期") + origin_airport_code: str = Field(..., description="出发机场代码") + destination_airport_code: str = Field(..., description="到达机场代码") + airfare: int = Field(..., description="机票价格") + + @field_validator("origin_airport_code", "destination_airport_code") + @classmethod + def validate_airport_code(cls, airport_code: str) -> str: + """校验机场代码""" + if ( + not airport_code.isalpha() + or not airport_code.isupper() + or len(airport_code) != 3 + ): + raise ValueError(f"机场代码须为英文、大写、3位字符串") + return airport_code class FlightNoFound(BaseModel): @@ -43,85 +78,104 @@ class FlightNoFound(BaseModel): 未查询到航班类 """ -class SeatPreference(BaseModel): - """ - 座位偏好类 - """ - row: int = Field(ge=1, le=30, description="座位行") - column: Literal["A", "B", "C", "D", "E", "F"] = Field(description="座位列") +system_prompt = """你的任务是根据用户提供的日期、出发机场和到达机场选择航班。 +1. 使用 extract_flights 从来源资料提取所有航班信息； +2. 使用 select_flight 从所有航班信息中选择匹配航班； +""" - -class SeatPreferenceNoExtracted(BaseModel): - """ - 未提取到座位偏好类 - """ - - -# 航班查询智能体 -flight_inquiry_agent = Agent[Dependences, Flight | FlightNoFound | DeferredToolRequests]( +# 主智能体 +agent = Agent[Deps, Flight | FlightNoFound]( model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=MODEL_SETTINGS_DISABLED_THINKING, - deps_type=Dependences, - output_type=Flight | FlightNoFound | DeferredToolRequests, - system_prompt="你的任务是根据给定的日期、出发机场和到达机场帮助用户找到最便宜的航班。", + model_settings=MODEL_SETTINGS, + deps_type=Deps, + output_type=Flight | FlightNoFound, + system_prompt=system_prompt, ) # 航班信息提取智能体 -flights_extraction_agent = Agent( +extraction_agent = Agent( model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=MODEL_SETTINGS_DISABLED_THINKING, + model_settings=MODEL_SETTINGS, output_type=list[Flight], - system_prompt="你的任务是根据给定的航班来源资料提取出所有航班信息，包括航班号、航班日期、航班出发机场、航班到达机场和航班机票价格。禁止编造。") + system_prompt=( + "你的任务是根据给定的来源资料提取出所有航班信息，包括航班号、日期、出发机场代码、到达机场代码和机票价格。", + "其中，出发机场代码和到达机场代码须为英文、大写、3位字符串，例如 San Francisco International Airport (SFO) 的机场代码为 SFO。", + "若某航班信息不全则跳过该航班，若没有任何航班信息则返回空列表。", + "禁止编造。", + ), +) # 通过提示词约束输出 +# 航班选择智能体 +selection_agent = Agent( + model=DEEPSEEK_V4_FLASH_MODEL, + model_settings=MODEL_SETTINGS, + output_type=Flight | FlightNoFound, + system_prompt="你的任务是从所有航班信息中选择匹配用户提供的日期、出发机场和到达机场的航班：若有多个航班则选择最便宜的航班，选择完成需用户确认是否预定；若没有则返回 FlightNoFound。", +) -@flight_inquiry_agent.tool -async def extract_flight_details(ctx: RunContext[Dependences]) -> list[Flight]: +""" +如何提高提取准确率 +1. 系统提示词规则约束 +2. 降低模型温度 +3. 输出类型约定和输出模型校验 +4. 业务规则约束（例如，输出不可为空列表） +""" + + +@agent.tool +async def extract_flights(ctx: RunContext[Deps]) -> list[Flight]: """ 提取所有航班信息 """ - result = await flights_extraction_agent.run( - ctx.deps.source_material, usage=ctx.usage + result = await extraction_agent.run( + ctx.deps.source_material, + usage_limits=UsageLimits(request_limit=3), + usage=ctx.usage, ) + if len(output := result.output) != 8: + raise ModelRetry("提取到的所有航班信息应为 8 条") + return output + + +@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 -@flight_inquiry_agent.output_validator +@agent.output_validator async def validate_output( - ctx: RunContext[Dependences], output: Flight | FlightNoFound | DeferredToolRequests -) -> Flight | FlightNoFound | DeferredToolRequests: + ctx: RunContext[Deps], output: Flight | FlightNoFound +) -> Flight | FlightNoFound: """ - 输出校验：航班信息 + 输出校验 """ # 不校验未查询到航班 if isinstance(output, FlightNoFound): return output - # 不校验延迟工具请求 - if isinstance(output, DeferredToolRequests): - return output - errors = [] if output.date != ctx.deps.date: errors.append(f"航班日期应为 {ctx.deps.date}, 不是 {output.date}") if output.origin != ctx.deps.origin: errors.append(f"航班出发机场应为 {ctx.deps.origin}, 不是 {output.origin}") if output.destination != ctx.deps.destination: - errors.append(f"航班到达机场应为 {ctx.deps.destination}, 不是 {output.destination}") + errors.append( + f"航班到达机场应为 {ctx.deps.destination}, 不是 {output.destination}" + ) if errors: raise ModelRetry("\n".join(errors)) return output -# 座位偏好提取智能体 -seat_preference_extraction_agent = Agent[object, SeatPreference | SeatPreferenceNoExtracted]( - model=DEEPSEEK_V4_FLASH_MODEL, - model_settings=MODEL_SETTINGS_DISABLED_THINKING, - output_type=SeatPreference | SeatPreferenceNoExtracted, - system_prompt="你的任务是根据用户回答提取座位偏好。座位规则说明：A 座、F 座为靠窗座位；第 1 排是前排座位，腿部空间更大；14 排、20 排同样拥有加宽腿部空间", -) - source_material = """ 1. Flight SFO-AK123 - Price: $350 @@ -166,62 +220,72 @@ source_material = """ """ -def init_work_flow() -> WorkFlow: - return WorkFlow( - type=WorkFlowType.BOOK_FLIGHT, - deps=Deps_( - flight_info=flight_info, - date=datetime.date(2025, 1, 10), - origin="SFO", - destination="ANC", - ), - usage_limits={}, - ) - - async def run_stream_events( - usage: RunUsage, user_prompt: str, + deps: Deps, message_history: list[ModelMessage], + usage: RunUsage, + deferred_tool_results: DeferredToolResults | None = None, ) -> AsyncGenerator: - result = None - while True: - if not task: - return + async with agent.run_stream_events( + user_prompt=user_prompt, + deps=deps, + message_history=message_history, + usage=usage, + deferred_tool_results=deferred_tool_results, + ) as events: + async for event in events: + yield event - match task.node: - # 航班查询 - case "flight_search": - async with flight_search_agent.run_stream_events( - user_prompt=user_prompt, - deps=task.deps, - message_history=message_history, - usage=usage, - ) 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) - 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 + +async def main(): + deps = Deps( + source_material=source_material, + date=datetime.date(2025, 1, 10), + origin="SFO", + destination="ANC", + ) + # message_history 传空列表，不要传 None + msg_history: list[ModelMessage] = [] + prompt = f"Find me a flight from {deps.origin} to {deps.destination} on {deps.date}" + + gen = run_stream_events( + user_prompt=prompt, + deps=deps, + message_history=msg_history, + usage=RunUsage(), + ) + # 消费异步生成器 + async for ev in gen: + print(ev) + + +asyncio.run(main()) + +""" +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": @@ -237,7 +301,7 @@ async def run_stream_events( task.node = "flight_search" else: - async with seat_preference_extraction_agent.run_stream_events( +async with seat_preference_extraction_agent.run_stream_events( user_prompt=user_prompt, message_history=message_history, usage=usage_to_object(dict(task.usage)), @@ -328,4 +392,25 @@ async def run_stream_events( dialog.thoughts[index].content = f"已调用 {content}" case _: - continue \ No newline at end of file + 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/models.py b/agent/application/workshop/models.py index a22a536..9f018c1 100644 --- a/agent/application/workshop/models.py +++ b/agent/application/workshop/models.py @@ -3,7 +3,7 @@ 智能体相关模块 """ from pydantic_ai import ModelSettings -from pydantic_ai.models.openai import OpenAIChatModel, OpenAIChatModelSettings +from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.providers.openai import OpenAIProvider DEEPSEEK_V4_FLASH_MODEL = OpenAIChatModel( diff --git a/agent/database.db b/agent/database.db index 061df73..74f44ce 100644 Binary files a/agent/database.db and b/agent/database.db differ