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

1

This commit is contained in:
liubiren 2026-07-06 08:24:10 +08:00
parent cc8efb7143
commit 1789f8a0a3
3 changed files with 73 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
智能体/application/models.py
View File

@ -74,7 +74,7 @@ class Kind(StrEnum):
THINKING = "thinking" THINKING = "thinking"
TOOL_SEARCH = "tool-search" TOOL_SEARCH = "tool-search"
CAPABILITY_LOAD = "capability-load" LOAD_CAPABILITY = "load-capability"
TOOL_CALL = "tool-call" TOOL_CALL = "tool-call"
TEXT = "text" TEXT = "text"
TOOL_RETURN = "tool-return" TOOL_RETURN = "tool-return"

216
智能体/application/state/conversation.py
View File

@ -21,6 +21,7 @@ from pydantic_ai.messages import (
ToolCallPart, ToolCallPart,
ToolSearchCallPart, ToolSearchCallPart,
) )
from pydantic_ai.run import AgentRunResultEvent
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
import reflex as rx import reflex as rx
@ -193,8 +194,8 @@ class ConversationState(rx.State):
conversation_id=self.conversation_id conversation_id=self.conversation_id
) )
# 初始化工具调用唯一标识与片段索引映射字典 # 初始化工具调用唯一标识和片段索引映射字典
tool_call_ids : Dict[int, str] = {} tool_call_ids: Dict[str, int] = {}
# 创建运行 # 创建运行
conversation.runs[str(uuid7())] = Run(user_prompt=user_prompt) conversation.runs[str(uuid7())] = Run(user_prompt=user_prompt)
@ -202,7 +203,7 @@ class ConversationState(rx.State):
run_id, run = next(reversed(conversation.runs.items())) run_id, run = next(reversed(conversation.runs.items()))
# 将运行状态设置为正在运行 # 将运行状态设置为正在运行
run.is_running = True run.is_running = True
yield # 通知前端渲染运行状态 yield # 通知前端渲染
async with self.agent.run_stream_events( async with self.agent.run_stream_events(
conversation_id=self.conversation_id, conversation_id=self.conversation_id,
@ -211,7 +212,7 @@ class ConversationState(rx.State):
) as events: ) as events:
async for event in events: async for event in events:
match event: match event:
# ========== 分片开始事件 ========== # ========== 开始事件 ==========
case PartStartEvent( case PartStartEvent(
index=index, index=index,
part=part, part=part,
@ -219,82 +220,79 @@ class ConversationState(rx.State):
): ):
match part: match part:
# 思考分片开始事件 # 思考分片开始事件
case ThinkingPart( case ThinkingPart(content=content):
content=content
):
# 若上一分片种类为空则将推理状态设置为正在推理 # 若上一分片种类为空则将推理状态设置为正在推理
if not previous_part_kind: if not previous_part_kind:
run.is_reasoning = True run.is_reasoning = True
# 初始化推理 run.reasonings[index] = Reasoning(
run.reasonings[index] = Reasoning(kind=Kind.THINKING, content=content) kind=Kind.THINKING, content=content
yield # 通知前端渲染推理 )
yield
case ToolSearchCallPart( # 工具检索分片开始事件
tool_call_id=tool_call_id case ToolSearchCallPart(tool_call_id=tool_call_id):
):
# 创建工具调用唯一标识与片段索引映射 # 创建工具调用唯一标识与片段索引映射
tool_call_ids[index] = tool_call_id tool_call_ids[tool_call_id] = index
# 初始化推理 run.reasonings[index] = Reasoning(
run.reasonings[index] = Reasoning(kind=Kind.TOOL_SEARCH, content="正在检索") kind=Kind.TOOL_SEARCH, content="正在检索"
yield # 通知前端渲染推理 )
yield
# 能力加载分片开始事件
case LoadCapabilityCallPart(tool_call_id=tool_call_id):
tool_call_ids[tool_call_id] = index
run.reasonings[index] = Reasoning(
kind=Kind.LOAD_CAPABILITY, content="正在加载能力"
)
yield
# 工具调用分片开始事件
case ToolCallPart(tool_call_id=tool_call_id):
tool_call_ids[tool_call_id] = index
run.reasonings[index] = Reasoning(
kind=Kind.TOOL_CALL, content="正在调用工具"
)
yield
# 文本分片开始事件 # 文本分片开始事件
case TextPart( case TextPart(content=content):
content=content
):
# 初始化回复正文
run.assistant_content = content run.assistant_content = content
yield # 通知前端渲染回复正文 yield
case _: # ========== 增量事件 ==========
continue 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,
): ):
run.reasonings[index].content += content_delta or "" run.reasonings[index].content += content_delta or ""
yield # 通知前端渲染推理 yield
# 文本分片增量事件 # 文本分片增量事件
case TextPartDelta( case TextPartDelta(
content_delta=content_delta, content_delta=content_delta,
): ):
run.assistant_content += content_delta run.assistant_content += content_delta
yield # 通知前端渲染回复正文 yield
case _: # ========== 结束事件 ==========
continue
# ========== 分片结束事件 ==========
case PartEndEvent( case PartEndEvent(
event_kind=event_kind,
index=index, index=index,
part=part, part=part,
next_part_kind=next_part_kind, next_part_kind=next_part_kind,
): ):
match part: match part:
# 思考分片结束事件 # 思考分片结束事件
case ThinkingPart( case ThinkingPart(part_kind=part_kind, content=content):
part_kind=part_kind, content=content
):
# 若下一分片种类为文本则将推理状态设置为非正在推理 # 若下一分片种类为文本则将推理状态设置为非正在推理
if next_part_kind == Kind.TEXT: if next_part_kind == Kind.TEXT:
run.is_reasoning = False run.is_reasoning = False
yield
# 检索工具分片结束事件
case ToolSearchCallPart(
tool_call_id=tool_call_id
):
# 创建工具调用唯一标识与片段索引映射
tool_call_ids[index] = tool_call_id
# 加载能力分片结束事件 # 加载能力分片结束事件
case LoadCapabilityCallPart( case LoadCapabilityCallPart(
@ -309,9 +307,7 @@ class ConversationState(rx.State):
run_id=current_run_id, run_id=current_run_id,
) )
# 调用工具分片结束事件 # 调用工具分片结束事件
case ToolCallPart( case ToolCallPart(part_kind=part_kind, tool_name=tool_name):
part_kind=part_kind, tool_name=tool_name
):
yield Event( yield Event(
event_kind=EventKind(event_kind), event_kind=EventKind(event_kind),
part_index=index, part_index=index,
@ -323,114 +319,34 @@ class ConversationState(rx.State):
# ========== 函数工具调用事件 ========== # ========== 函数工具调用事件 ==========
case FunctionToolCallEvent( case FunctionToolCallEvent(
event_kind=event_kind,
part=part, part=part,
): ):
yield Event( match part:
event_kind=EventKind(event_kind), # 工具检索分片函数工具调用事件
part_index=self.tool_call_ids.get( case ToolSearchCallPart(tool_call_id=tool_call_id):
part.tool_call_id index = tool_call_ids[tool_call_id]
), run.reasonings[index].content = "检索完成"
part_kind=PartKind(part.part_kind),
tool_name=part.tool_name,
run_id=current_run_id,
)
# ========== 函数工具回调事件 ========== # ========== 函数工具结果事件 ==========
case FunctionToolResultEvent( case FunctionToolResultEvent(
event_kind=event_kind,
content=content, content=content,
part=part, part=part,
): ):
yield Event( match part:
event_kind=EventKind(event_kind), # 工具检索分片函数工具结果事件
event_content=( case ToolSearchCallPart(tool_call_id=tool_call_id):
content if isinstance(content, str) else "" index = tool_call_ids[tool_call_id]
), # 暂仅处理文本类型，后续处理多模态和缓存 run.reasonings[index].content += content or ""
part_index=self.tool_call_ids.get( yield
part.tool_call_id
),
part_kind=PartKind(part.part_kind),
tool_name=part.tool_name,
run_id=current_run_id,
)
yield Event( # ========== 智能体运行结果事件 ==========
event_kind=EventKind.RUN_END, case AgentRunResultEvent(result=result):
run_id=current_run_id, # 保存新增消息
run_new_messages=agent_run.new_messages(), await database_state.save_new_messages(
)
# 运行，处理分片生命周期事件
async for self.agent.run(
conversation_id=self.conversation_id, conversation_id=self.conversation_id,
user_prompt=user_prompt, run_id=run_id,
message_history=message_history, new_messages=result.new_messages(),
) as async_event:
if (
not event
or not self.current_run_id
or self.current_run_id != event.run_id
):
continue
# ========== 运行开始 ==========
if event.event_kind == EventKind.RUN_START:
# 将事件运行唯一标识设置为当前运行唯一标识
self.current_run_id = event.run_id
# 创建运行
conversation.runs[self.current_run_id] = Run(
user_prompt=user_prompt, is_streaming=True
) )
yield # 通知前端渲染 # 将运行状态设置为非正在运行
continue run.is_running = False
yield
# 当前运行
current_run = current_conversation.runs[self.current_run_id]
# ========== 推理 ==========
if (
event.event_kind == EventKind.PART_START
and event.part_kind == PartKind.THINKING
):
current_run.thinking += event.event_content
yield # 通知前端渲染
continue
if part_index := event.part_index:
if part_index not in current_run.reasonings:
current_run.reasonings[part_index] += event.event_content
yield # 通知前端渲染
continue
# ========== 回答 ==========
if (
event.part_kind == PartKind.TEXT
and event.event_content != EventKind.PART_END
):
current_run.answer += event.event_content
yield # 通知前端渲染
continue
# ========== 运行结束 ==========
if event.event_kind == EventKind.RUN_END:
# 保存运行新增消息
await database_state.save_new_message(
conversation_id=self.current_conversation_id,
run_id=self.current_run_id,
run_new_message=event.run_new_messages,
)
# 设置当前运行流式输出状态非正在流式输出
current_run.is_streaming = False
yield # 通知前端渲染
continue

8
智能体/application/state/database.py
View File

@ -34,21 +34,21 @@ class DatabaseState(rx.State):
) )
return message_history return message_history
async def save_new_message( async def save_new_messages(
self, conversation_id: str, run_id: str, new_message: List[ModelMessage] self, conversation_id: str, run_id: str, new_messages: List[ModelMessage]
) -> None: ) -> None:
""" """
保存新增消息 保存新增消息
:param conversation_id: 对话唯一标识 :param conversation_id: 对话唯一标识
:param run_id: 运行唯一标识 :param run_id: 运行唯一标识
:param new_message: 新增消息 :param new_messages: 新增消息
:return: None :return: None
""" """
async with rx.asession() as session: async with rx.asession() as session:
record = MessageHistory.adapt( record = MessageHistory.adapt(
conversation_id=conversation_id, conversation_id=conversation_id,
run_id=run_id, run_id=run_id,
new_message=new_message, new_message=new_messages,
) )
session.add(record) session.add(record)
await session.commit() await session.commit()