liubiren
/
Python
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

1

This commit is contained in:
liubiren 2026-08-19 19:51:59 +08:00
parent 988a8dc557
commit d72cdf815d
4 changed files with 171 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

188
agent/application/states/conversation.py
View File

@ -29,6 +29,7 @@ from application.states.models import (
MessageType, MessageType,
Work, Work,
WorkType, WorkType,
TaskStatus,
ConversationHistoryItem, ConversationHistoryItem,
usage_validate_python, usage_validate_python,
MessageHistoryItem, MessageHistoryItem,
@ -303,12 +304,10 @@ class ConversationState(rx.State):
# 初始化片段索引映射为消息实例唯一标识字典 # 初始化片段索引映射为消息实例唯一标识字典
index_map_to_message_id: dict[int, str] = {} index_map_to_message_id: dict[int, str] = {}
# 初始化工具调用唯一标识集合
tool_call_ids: set[str] = set()
# 非结构化对话 # 对话
if not conversation.work: if not conversation.work:
from application.workshop.unstructured_dialogue import ( from application.workshop.talk import (
run_stream_events, run_stream_events,
) )
else: else:
@ -318,7 +317,6 @@ class ConversationState(rx.State):
case WorkType.BOOK_FLIGHT: case WorkType.BOOK_FLIGHT:
from application.workshop.book_flight import run_stream_events from application.workshop.book_flight import run_stream_events
# 消息列表 # 消息列表
messages: list[Message] = [] messages: list[Message] = []
# 运行并流式输出事件 # 运行并流式输出事件
@ -329,108 +327,104 @@ class ConversationState(rx.State):
), ),
usage=usage_validate_python(dict(conversation.usage)), usage=usage_validate_python(dict(conversation.usage)),
) )
try: # 获取运行流式输出事件
# 获取运行流式输出事件 async for event in stream_events:
async for event in stream_events: # 将等待流式输出设置为否
# 将等待流式输出设置为否 conversation.awaiting_stream = False
conversation.awaiting_stream = False message: Message | None = None
message: Message | None = None match event:
match event: # ========== 开始事件 ==========
# ========== 开始事件 ========== case PartStartEvent(
case PartStartEvent( index=index,
index=index, part=part,
part=part, ):
): match part:
match part: # 思考分片开始事件
# 思考分片开始事件 case ThinkingPart(content=content):
case ThinkingPart(content=content): # 构建消息实例
# 构建消息实例 message = Message(
message = Message( type=MessageType.THINKING,
type=MessageType.THINKING, title="正在思考",
title="正在思考", content=content,
content=content, is_running=True,
is_running=True, )
) # 添加至消息字典
# 添加至消息字典 conversation.messages[message.id] = message
conversation.messages[message.id] = message # 将消息实例唯一标识与片段索引映射
# 将消息实例唯一标识与片段索引映射 index_map_to_message_id[index] = message.id
index_map_to_message_id[index] = message.id yield # 通知前端更新渲染
yield # 通知前端更新渲染
# 文本分片开始事件 # 文本分片开始事件
case TextPart(content=content): case TextPart(content=content):
# 构建消息实例 # 构建消息实例
message = Message(type=MessageType.RESULT_OUTPUT) message = Message(type=MessageType.RESULT_OUTPUT)
message.content = content message.content = content
# 添加至消息字典 # 添加至消息字典
conversation.messages[message.id] = message conversation.messages[message.id] = message
# 将消息实例唯一标识与片段索引映射 # 将消息实例唯一标识与片段索引映射
index_map_to_message_id[index] = message.id index_map_to_message_id[index] = message.id
yield # 通知前端更新渲染 yield # 通知前端更新渲染
# ========== 增量事件 ========== # ========== 增量事件 ==========
case PartDeltaEvent(index=index, delta=delta): case PartDeltaEvent(index=index, delta=delta):
match delta: match delta:
# 思考分片增量事件 # 思考分片增量事件
case ThinkingPartDelta( case ThinkingPartDelta(
content_delta=content_delta, content_delta=content_delta,
): ):
# 若内容增量不为空则增量更新 # 若内容增量不为空则增量更新
if content_delta: if content_delta:
conversation.messages[
index_map_to_message_id[index]
].content += content_delta
yield # 通知前端更新渲染
# 文本分片增量事件
case TextPartDelta(
content_delta=content_delta,
):
# 增量更新
conversation.messages[ conversation.messages[
index_map_to_message_id[index] index_map_to_message_id[index]
].content += content_delta ].content += content_delta
yield # 通知前端更新渲染 yield # 通知前端更新渲染
# ========== 结束事件 ========== # 文本分片增量事件
case PartEndEvent( case TextPartDelta(
index=index, content_delta=content_delta,
part=part, ):
): # 增量更新
match part: conversation.messages[
# 思考分片结束事件 index_map_to_message_id[index]
case ThinkingPart(content=content): ].content += content_delta
# 获取消息实例 yield # 通知前端更新渲染
message = conversation.messages[
index_map_to_message_id[index]
]
message.is_running = False
message.title = "思考完成"
yield # 通知前端更新渲染
messages.append(message)
case TextPart(content=content): # ========== 结束事件 ==========
# 获取消息实例 case PartEndEvent(
message = conversation.messages[ index=index,
index_map_to_message_id[index] part=part,
] ):
messages.append(message) 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 TextPart(content=content):
case AgentRunResultEvent(result=result): # 获取消息实例
# 更新使用量 message = conversation.messages[
await db_state.update_conversation_record( index_map_to_message_id[index]
conversation.id, usage=result.usage ]
) messages.append(message)
# 创建运行记录
await db_state.create_run_record( # ========== 智能体运行结果事件 ==========
conversation_id=conversation.id, case AgentRunResultEvent(result=result):
new_messages=result.new_messages(), # 更新使用量
) 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 conversation.is_running = False
# 将等待流式输出设置为否 # 将等待流式输出设置为否

3
agent/application/states/models.py
View File

@ -19,6 +19,7 @@ class TaskStatus(StrEnum):
RUNNING = "running" RUNNING = "running"
DONE = "done" DONE = "done"
PENDING_APPROVAL = "pending_approval"
ERROR = "error" ERROR = "error"
@ -57,7 +58,7 @@ class Message(BaseModel):
content: str = Field(default="", description="消息内容") content: str = Field(default="", description="消息内容")
tasks: list[Task] = Field(default_factory=list, description="任务列表") tasks: list[Task] = Field(default_factory=list, description="任务列表")
is_running: bool = Field( is_running: bool = Field(
default=False, description="正在运行，True 表示正在运行，False 表示运行完成" default=False, description="正在运行，True 表示正在运行， False 表示运行结束"
) )
is_shown: bool = Field( is_shown: bool = Field(
default=False, description="展示组件，True 表示展示，False 表示隐藏" default=False, description="展示组件，True 表示展示，False 表示隐藏"

284
agent/application/workshop/book_flight.py
View File

@ -21,7 +21,7 @@ from pydantic_ai import (
) )
from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider 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 enum import StrEnum
from pydantic_ai._uuid import uuid7 from pydantic_ai._uuid import uuid7
from pydantic_ai.messages import ( from pydantic_ai.messages import (
@ -49,6 +49,7 @@ class TaskStatus(StrEnum):
RUNNING = "running" RUNNING = "running"
DONE = "done" DONE = "done"
ERROR = "error" ERROR = "error"
PENDING_APPROVAL = "pending_approval"
class Task(BaseModel): class Task(BaseModel):
@ -145,11 +146,11 @@ class Deps(BaseModel):
date: datetime.date = Field(..., description="日期") date: datetime.date = Field(..., description="日期")
origin_airport_code: str = Field(..., description="出发机场代码") origin_airport_code: str = Field(..., description="出发机场代码")
destination_airport_code: str = Field(..., description="到达机场代码") destination_airport_code: str = Field(..., description="到达机场代码")
matched_flights: list[Flight] | None = Field( searched_flights: list[Flight] | None = Field(
default=None, description="匹配到的航班" default=None, description="查询到的航班"
) )
selected_flight: Flight | NoSelectedFlight | None = Field( recommended_flight: Flight | NoSelectedFlight | None = Field(
default=None, description="选择到的航班" default=None, description="推荐航班"
) )
@ -228,9 +229,9 @@ source_material = """
@agent.tool @agent.tool
async def match_flights(ctx: RunContext[Deps]) -> list[Flight]: async def search_flights(ctx: RunContext[Deps]) -> list[Flight]:
""" """
匹配满足用户需求的航班 查询航班
**如何提高提取准确率** **如何提高提取准确率**
1. 系统提示词规则约束 1. 系统提示词规则约束
2. 降低模型温度 2. 降低模型温度
@ -238,7 +239,7 @@ async def match_flights(ctx: RunContext[Deps]) -> list[Flight]:
4. 业务规则约束（例如，输出不可为空列表） 4. 业务规则约束（例如，输出不可为空列表）
**设计工具需先设计业务流程** **设计工具需先设计业务流程**
本示例按照 匹配满足用户需求的航班 -> 选择航班 -> 预定航班 本示例按照 查询航班 -> 推荐航班 -> 预定航班
""" """
extraction_agent_result = await extraction_agent.run( extraction_agent_result = await extraction_agent.run(
f"来源资料：\n{source_material}", f"来源资料：\n{source_material}",
@ -254,35 +255,35 @@ async def match_flights(ctx: RunContext[Deps]) -> list[Flight]:
raise ModelRetry( raise ModelRetry(
f"提取到的所有航班信息数量应为 8 ，不是 {extracted_flight_counts}" f"提取到的所有航班信息数量应为 8 ，不是 {extracted_flight_counts}"
) )
# 匹配用户需求的航班 # 查询到的航班
ctx.deps.matched_flights = [ ctx.deps.searched_flights = [
flight flight
for flight in extracted_flights for flight in extracted_flights
if flight.date == ctx.deps.date if flight.date == ctx.deps.date
and flight.origin_airport_code == ctx.deps.origin_airport_code and flight.origin_airport_code == ctx.deps.origin_airport_code
and flight.destination_airport_code == ctx.deps.destination_airport_code and flight.destination_airport_code == ctx.deps.destination_airport_code
] ]
return ctx.deps.matched_flights return ctx.deps.searched_flights
@agent.tool @agent.tool
async def select_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight: async def recommend_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight:
""" """
选择航班 推荐航班
**如何提高提取准确率** **如何提高提取准确率**
1. 系统提示词规则约束 1. 系统提示词规则约束
2. 降低模型温度 2. 降低模型温度
3. 输出类型约定和输出模型校验 3. 输出类型约定和输出模型校验
4. 业务规则约束（例如，输出不可为空列表） 4. 业务规则约束（例如，输出不可为空列表）
""" """
if (matched_flights := ctx.deps.matched_flights) is None: if (searched_flights := ctx.deps.searched_flights) is None:
raise ModelRetry("必须先使用 match_flights 匹配满足用户需求的航班") raise ModelRetry("必须先使用 search_flights 查询航班")
ctx.deps.selected_flight = ( ctx.deps.recommended_flight = (
min(matched_flights, key=lambda flight: flight.airfare) min(searched_flights, key=lambda flight: flight.airfare)
if matched_flights if searched_flights
else NoSelectedFlight() else NoSelectedFlight()
) )
return ctx.deps.selected_flight return ctx.deps.recommended_flight
@agent.tool @agent.tool
@ -290,19 +291,15 @@ async def book_flight(ctx: RunContext[Deps]) -> Flight | NoSelectedFlight:
""" """
预定航班 预定航班
""" """
if (selected_flight := ctx.deps.selected_flight) is None: if (recommended_flight := ctx.deps.recommended_flight) is None:
raise ModelRetry("必须先使用 select_flight 选择航班") raise ModelRetry("必须先使用 recommend_flight 推荐航班")
if isinstance(selected_flight, NoSelectedFlight): if isinstance(recommended_flight, NoSelectedFlight):
return selected_flight return recommended_flight
else: else:
if not ctx.tool_call_approved: if not ctx.tool_call_approved:
raise ApprovalRequired( raise ApprovalRequired()
metadata={ return recommended_flight
"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
@agent.output_validator @agent.output_validator
@ -339,10 +336,10 @@ async def run_stream_events(
usage: RunUsage, usage: RunUsage,
deferred_tool_results: DeferredToolResults | None = None, deferred_tool_results: DeferredToolResults | None = None,
) -> AsyncGenerator: ) -> AsyncGenerator:
# 构建工作输出消息 # 构建消息
message = Message( message = Message(
type=MessageType.WORK_OUTPUT, type=MessageType.WORK_OUTPUT,
title="正在预定航班", title="预定航班",
is_running=True, is_running=True,
) )
yield message yield message
@ -354,6 +351,7 @@ async def run_stream_events(
deferred_tool_results=deferred_tool_results, deferred_tool_results=deferred_tool_results,
) as events: ) as events:
async for event in events: async for event in events:
print(event)
match event: match event:
case PartStartEvent( case PartStartEvent(
part=part, part=part,
@ -363,7 +361,7 @@ async def run_stream_events(
tool_name=tool_name, tool_call_id=tool_call_id tool_name=tool_name, tool_call_id=tool_call_id
): ):
match tool_name: match tool_name:
case "match_flights": case "search_flights":
if tool_name not in message.tasks: if tool_name not in message.tasks:
message.tasks[tool_name] = Task( message.tasks[tool_name] = Task(
tool_name=tool_name, tool_name=tool_name,
@ -372,15 +370,23 @@ async def run_stream_events(
) )
yield message yield message
case "select_flight": case "recommend_flight":
if tool_name not in message.tasks: if tool_name not in message.tasks:
message.tasks[tool_name] = Task( message.tasks[tool_name] = Task(
tool_name=tool_name, tool_name=tool_name,
tool_call_id=tool_call_id, tool_call_id=tool_call_id,
title="正在选择航班", title="正在推荐航班",
) )
yield message 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): case FunctionToolResultEvent(part):
match part: match part:
@ -390,27 +396,52 @@ async def run_stream_events(
content=content, content=content,
): ):
match tool_name: match tool_name:
case "match_flights": case "search_flights":
matched_flights = cast(list[Flight], content)
task = message.tasks[tool_name] 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( 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}" 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 matched_flight in matched_flights for searched_flight in searched_flights
] ]
) )
task.status = TaskStatus.DONE task.status = TaskStatus.DONE
yield message yield message
case "select_flight": case "recommend_flight":
selected_flight = cast(Flight, content)
task = message.tasks[tool_name] task = message.tasks[tool_name]
task.title = f"已选择航班" recommended_flight = cast(
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}" 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 task.status = TaskStatus.DONE
yield message 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(): async def main():
# 实例化依赖 # 实例化依赖
@ -418,7 +449,6 @@ async def main():
date=datetime.date(2025, 1, 10), date=datetime.date(2025, 1, 10),
origin_airport_code="SFO", origin_airport_code="SFO",
destination_airport_code="ANC", destination_airport_code="ANC",
selected_flight=None,
) )
user_prompt = f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班" user_prompt = f"帮我查询并预定在 {deps.date.strftime('%Y-%m-%d')} 从 {deps.origin_airport_code} 到 {deps.destination_airport_code} 的航班"
message_history: list[ModelMessage] = [] message_history: list[ModelMessage] = []
@ -430,172 +460,12 @@ async def main():
) )
# 消费异步生成器 # 消费异步生成器
async for event in events: async for event in events:
print(event) print()
if isinstance(event, AgentRunResultEvent): if isinstance(event, AgentRunResultEvent):
# 保存本次运行新增消息 # 保存本次运行新增消息
message_history.extend(event.result.new_messages()) message_history.extend(event.result.new_messages())
if isinstance(event, Message):
print(event)
asyncio.run(main()) 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,
)
"""

2
agent/application/workshop/unstructured_dialogue.py → agent/application/workshop/talk.py
View File

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
非结构化对话 对话
""" """
from typing import AsyncGenerator from typing import AsyncGenerator