1

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:

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 logfire_api.variables import ValueDoesNotEqual
-from pydantic_ai import Agent, ModelMessage, ModelRetry, RunContext, UsageLimits, RunUsage, DeferredToolRequests
+from pydantic import BaseModel, Field, field_validator
-from pydantic_ai.run import AgentRunResultEvent
+from pydantic_ai import (
-
+    Agent,
-from application.states.models import (
+    ApprovalRequired,
-    WorkFlow,
+    DeferredToolRequests,
-    WorkFlowType,
+    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="航班日期")
+    date: datetime.date = Field(..., description="日期")
-    origin: str = Field(..., description="航班出发机场")
+    origin_airport_code: str = Field(..., description="出发机场代码")
-    destination: str = Field(..., description="航班到达机场")
+    destination_airport_code: str = Field(..., description="到达机场代码")
-    source_material: 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="航班日期")
+    date: datetime.date = Field(..., description="日期")
-    origin: str = Field(..., description="航班出发机场")
+    origin_airport_code: str = Field(..., description="出发机场代码")
-    destination: str = Field(..., description="航班到达机场")
+    destination_airport_code: str = Field(..., description="到达机场代码")
-    airfare: int = 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="座位行")
+system_prompt = """你的任务是根据用户提供的日期、出发机场和到达机场选择航班。
-    column: Literal["A", "B", "C", "D", "E", "F"] = Field(description="座位列")
+1. 使用 extract_flights 从来源资料提取所有航班信息；
+2. 使用 select_flight 从所有航班信息中选择匹配航班；
+"""
 
-
+# 主智能体
-class SeatPreferenceNoExtracted(BaseModel):
+agent = Agent[Deps, Flight | FlightNoFound](
-    """
-    未提取到座位偏好类
-    """
-
-
-# 航班查询智能体
-flight_inquiry_agent = Agent[Dependences, Flight | FlightNoFound | DeferredToolRequests](
     model=DEEPSEEK_V4_FLASH_MODEL,
-    model_settings=MODEL_SETTINGS_DISABLED_THINKING,
+    model_settings=MODEL_SETTINGS,
-    deps_type=Dependences,
+    deps_type=Deps,
-    output_type=Flight | FlightNoFound | DeferredToolRequests,
+    output_type=Flight | FlightNoFound,
-    system_prompt="你的任务是根据给定的日期、出发机场和到达机场帮助用户找到最便宜的航班。",
+    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(
+    result = await extraction_agent.run(
-        ctx.deps.source_material, usage=ctx.usage
+        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
+    ctx: RunContext[Deps], output: Flight | FlightNoFound
-) -> Flight | FlightNoFound | DeferredToolRequests:
+) -> 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
+    async with agent.run_stream_events(
-    while True:
+        user_prompt=user_prompt,
-        if not task:
+        deps=deps,
-            return
+        message_history=message_history,
+        usage=usage,
+        deferred_tool_results=deferred_tool_results,
+    ) as events:
+        async for event in events:
+            yield event
 
-        match task.node:
+
-            # 航班查询
+async def main():
-            case "flight_search":
+    deps = Deps(
-                async with flight_search_agent.run_stream_events(
+        source_material=source_material,
-                    user_prompt=user_prompt,
+        date=datetime.date(2025, 1, 10),
-                    deps=task.deps,
+        origin="SFO",
-                    message_history=message_history,
+        destination="ANC",
-                    usage=usage,
+    )
-                ) as events:
+    # message_history 传空列表，不要传 None
-                    async for event in events:
+    msg_history: list[ModelMessage] = []
-                        if not isinstance(event, AgentRunResultEvent):
+    prompt = f"Find me a flight from {deps.origin} to {deps.destination} on {deps.date}"
-                            yield event
+
-                        else:
+    gen = run_stream_events(
-                            result = event.result
+        user_prompt=prompt,
-                            # 更新任务使用量
+        deps=deps,
-                            task.usage = usage_to_dict(result.usage)
+        message_history=msg_history,
-                            if isinstance(result.output, FlightDetail):
+        usage=RunUsage(),
-                                content = "\n---\n已查询到航班，请回复 buy 购票 或 search 重新查询\n"
+    )
-                                # 更新任务节点为提取座位偏好
+    # 消费异步生成器
-                                task.node = "seat_preference_extraction"
+    async for ev in gen:
-                            else:
+        print(ev)
-                                content = (
+
-                                    "\n---\n未查询到满足您需求的航班，流程结束！\n"
+
-                                )
+asyncio.run(main())
-                                # 更新任务为空
+
-                                task = None
+"""
-                            # 返回任务节点结果事件
+if not isinstance(event, AgentRunResultEvent):
-                            yield TaskNodeResultEvent(
+                    yield event
-                                task=task,
+                else:
-                                content=content,
+                    result = event.result
-                            )
+                    # 更新任务使用量
-                            yield event
+                    task.usage = usage_to_dict(result.usage)
-                            return
+                    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)),
@@ -329,3 +393,24 @@ async def run_stream_events(
 
                             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,
+)
+"""

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(