style(agent): 调整agent结构
This commit is contained in:
@@ -19,6 +19,8 @@ from pydantic import BaseModel
|
||||
from ulid import ULID
|
||||
import asyncio
|
||||
from kilostar.utils.access import Accessor, TokenData
|
||||
from kilostar.utils.check_user.role_check import RoleChecker
|
||||
from kilostar.core.postgres_database.model import UserAuthority
|
||||
|
||||
workflow_router = APIRouter(prefix="/api/v1/workflow", tags=["workflow"])
|
||||
|
||||
@@ -54,7 +56,8 @@ async def create_workflow(
|
||||
|
||||
|
||||
@workflow_router.get("/list")
|
||||
async def get_workflow_list(token_data: TokenData = Depends(Accessor.get_current_user)):
|
||||
async def get_workflow_list(
|
||||
token_data: TokenData = Depends(RoleChecker(allowed_roles=UserAuthority.USER))):
|
||||
postgres_database = ray_actor_hook("postgres_database").postgres_database
|
||||
workflows = await postgres_database.list_workflows.remote(
|
||||
user_id=token_data.user_id
|
||||
|
||||
@@ -14,14 +14,26 @@
|
||||
|
||||
|
||||
from kilostar.core.work.workflow.workflow import KiloStarWorkflow, WorkflowStep
|
||||
from kilostar.utils.agent_model import ResponseModel, DepsModel, InputModel
|
||||
from kilostar.utils.agent_model import ResponseModel, DepsModel, RequestModel
|
||||
from pydantic import Field
|
||||
from typing import Optional, List
|
||||
|
||||
|
||||
# 意识节点回复类
|
||||
class ConsciousnessNodeResponse(ResponseModel):
|
||||
"""Consciousness response model,是意识节点所有回复类型的父类"""
|
||||
pass
|
||||
|
||||
class ConsciousnessNodeDeps(DepsModel):
|
||||
"""ConsciousnessNodeDeps 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
original_command: str
|
||||
command: str
|
||||
available_skills: Optional[List[str]]
|
||||
|
||||
class ConsciousnessNodeInput(RequestModel):
|
||||
"""ConsciousnessNodeInput 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
pass
|
||||
|
||||
|
||||
@@ -47,23 +59,6 @@ class ForregulatoryNode(ConsciousnessNodeResponse):
|
||||
..., description="为监控节点提供的全工作流执行情况的技术性总结报告。"
|
||||
)
|
||||
|
||||
|
||||
class ConsciousnessNodeDeps(DepsModel):
|
||||
"""ConsciousnessNodeDeps 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
|
||||
original_command: str
|
||||
command: str
|
||||
available_skills: list[dict] | None = None
|
||||
|
||||
|
||||
class ConsciousnessNodeInput(InputModel):
|
||||
"""ConsciousnessNodeInput 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class ForWorkflowEngineInput(ConsciousnessNodeInput):
|
||||
"""ForWorkflowEngineInput 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 ForWorkflowEngineInput 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
@@ -14,26 +14,16 @@
|
||||
|
||||
import datetime
|
||||
import ray
|
||||
from typing import Union, overload
|
||||
from typing import Union
|
||||
from kilostar.adapter.model_adapter.agent_factory import AgentFactory
|
||||
from kilostar.core.global_state_machine.global_state_machine import GlobalStateMachine
|
||||
from kilostar.core.global_state_machine.model_provider import Provider
|
||||
from kilostar.core.individual.regulatory_node.template import (
|
||||
ForConsciousnessNode,
|
||||
ForUser,
|
||||
regulatoryNodeDeps,
|
||||
TerminationMessage,
|
||||
MessageRequest,
|
||||
RegulatoryNodeDeps,
|
||||
MessageResponse
|
||||
)
|
||||
from pydantic_ai import RunContext, Agent
|
||||
from kilostar.utils.ray_hook import ray_actor_hook
|
||||
|
||||
|
||||
class ClientMessage:
|
||||
def __init__(self, user_id: str, user_name: str, message: str):
|
||||
self.user_id = user_id
|
||||
self.user_name = user_name
|
||||
self.message = message
|
||||
self.platform = "client"
|
||||
|
||||
|
||||
@ray.remote
|
||||
@@ -62,7 +52,7 @@ class RegulatoryNode:
|
||||
global_state_machine: 全局状态机
|
||||
provider_title: 供应商名
|
||||
model_id: 模型id
|
||||
|
||||
tools_list: 工具列表
|
||||
Returns:
|
||||
无返回
|
||||
"""
|
||||
@@ -76,9 +66,8 @@ class RegulatoryNode:
|
||||
"3. 如果你收到的是 TerminationMessage（代表工作流已完成并生成了报告），请将报告内容转化为友好的面向用户的回复，使用 ForUser 格式。\n"
|
||||
"请保持冷静、专业，并严格遵循上述路由规则。"
|
||||
)
|
||||
output_type = Union[ForConsciousnessNode, ForUser]
|
||||
output_type = Union[MessageResponse]
|
||||
from kilostar.utils.get_tool import load_tools_from_list
|
||||
|
||||
provider: Provider = await global_state_machine.get_provider.remote(
|
||||
provider_title
|
||||
)
|
||||
@@ -90,13 +79,13 @@ class RegulatoryNode:
|
||||
model_id=model_id,
|
||||
output_type=output_type,
|
||||
system_prompt=system_prompt,
|
||||
deps_type=regulatoryNodeDeps,
|
||||
deps_type=RegulatoryNodeDeps,
|
||||
agent_name="regulatory_node",
|
||||
tools=callables,
|
||||
)
|
||||
|
||||
@self.agent.system_prompt
|
||||
async def dynamic_prompt(ctx: RunContext[regulatoryNodeDeps]):
|
||||
async def dynamic_prompt(ctx: RunContext[RegulatoryNodeDeps]):
|
||||
"""执行与 dynamic prompt 相关的核心业务流转操作。
|
||||
该方法封装了具体的算法策略或状态控制逻辑，确保操作能够在事务上下文中被原子且一致地执行。
|
||||
Args: ctx (RunContext[regulatoryNodeDeps]): 参与 dynamic prompt 逻辑运算或数据构建的上下文依赖对象。
|
||||
@@ -123,23 +112,7 @@ class RegulatoryNode:
|
||||
)
|
||||
return prompt
|
||||
|
||||
async def handle_chat_message(
|
||||
self, user_id: str, chat_id: str, message: str
|
||||
) -> str:
|
||||
payload = ClientMessage(
|
||||
user_id=user_id, user_name="", message=message
|
||||
)
|
||||
return await self._process(payload)
|
||||
|
||||
async def handle_client_message(
|
||||
self, user_id: str, user_name: str, message: str
|
||||
) -> str:
|
||||
payload = ClientMessage(
|
||||
user_id=user_id, user_name=user_name, message=message
|
||||
)
|
||||
return await self._process(payload)
|
||||
|
||||
async def working(self, payload: Union[ClientMessage, TerminationMessage]) -> str:
|
||||
async def working(self, payload: MessageRequest) -> str:
|
||||
"""working方法，是节点唯一的调用方法，对于_run函数的结果进行判断并实现最终回复
|
||||
Args:
|
||||
payload: 消息载荷，包含所有信息
|
||||
@@ -147,56 +120,27 @@ class RegulatoryNode:
|
||||
Returns:
|
||||
str,监控节点对于用户的回复
|
||||
"""
|
||||
return await self._process(payload)
|
||||
|
||||
async def _process(
|
||||
self, payload: Union[ClientMessage, TerminationMessage]
|
||||
) -> str:
|
||||
try:
|
||||
result = await self._run(payload)
|
||||
if isinstance(result, ForConsciousnessNode):
|
||||
self.logger.info("regulatoryNode: 任务需交意识节点处理")
|
||||
workflow_running_engine = ray_actor_hook(
|
||||
"workflow_running_engine"
|
||||
).workflow_running_engine
|
||||
await workflow_running_engine.put_event.remote(payload)
|
||||
return f"任务已创建，准备创建工作流。原因：{result.reasoning}"
|
||||
elif isinstance(result, ForUser):
|
||||
self.logger.info("regulatoryNode: 直接向用户返回简单回复。")
|
||||
return result.context
|
||||
else:
|
||||
self.logger.error(f"regulatoryNode: 未知响应类型: {type(result)}")
|
||||
return "抱歉，系统内部遇到未知错误，无法正确处理您的请求。"
|
||||
except Exception:
|
||||
self.logger.exception("regulatoryNode在处理请求时发生未捕获的严重错误")
|
||||
return "抱歉，监控节点处理请求时发生严重错误，请联系管理员。"
|
||||
|
||||
@overload
|
||||
async def _run(
|
||||
self, payload: ClientMessage
|
||||
) -> Union[ForConsciousnessNode, ForUser]: ...
|
||||
|
||||
@overload
|
||||
async def _run(self, payload: TerminationMessage) -> ForUser: ...
|
||||
await self._run(payload)
|
||||
return ""
|
||||
|
||||
async def _run(
|
||||
self, payload: Union[ClientMessage, TerminationMessage]
|
||||
) -> Union[ForConsciousnessNode, ForUser]:
|
||||
self, payload: MessageRequest
|
||||
) -> Union[MessageResponse, None]:
|
||||
platform = payload.platform
|
||||
user_name = payload.user_name
|
||||
message = payload.message
|
||||
time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
try:
|
||||
deps = regulatoryNodeDeps(
|
||||
platform=platform, user_name=user_name, time=time_str
|
||||
deps = RegulatoryNodeDeps(
|
||||
platform=platform,
|
||||
user_name=user_name,
|
||||
time=time_str
|
||||
)
|
||||
self.logger.debug("regulatoryNode 开始生成 (启用原生 Pydantic-AI 重试)")
|
||||
prompt_message = message
|
||||
if isinstance(payload, TerminationMessage):
|
||||
prompt_message = f"【工作流执行结束报告】\n请将以下技术报告转化为对用户的友好回复：\n{message}"
|
||||
self.agent.retries = 3
|
||||
result = await self.agent.run(prompt_message, deps=deps)
|
||||
return result.output
|
||||
except Exception as e:
|
||||
self.logger.exception(f"regulatoryNode 模型生成或解析最终失败: {str(e)}")
|
||||
return ForUser(context="系统当前负载过高或遇到复杂内部错误，请稍后再试。")
|
||||
agent_response = await self.agent.run(user_prompt=message,
|
||||
deps=deps,)
|
||||
response: MessageResponse = agent_response.output
|
||||
response.platform = platform
|
||||
response.platform_id = MessageRequest.platform_id
|
||||
return response
|
||||
except:
|
||||
pass
|
||||
@@ -11,51 +11,54 @@
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
from typing import Literal, Optional
|
||||
|
||||
from pydantic import Field
|
||||
from kilostar.utils.agent_model import ResponseModel, DepsModel
|
||||
from pydantic import BaseModel
|
||||
|
||||
from kilostar.utils.agent_model import ResponseModel, DepsModel, RequestModel
|
||||
|
||||
|
||||
class regulatoryNodeResponse(ResponseModel):
|
||||
"""regulatoryNodeResponse 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
class RegulatoryNodeResponse(ResponseModel):
|
||||
"""
|
||||
RegulatoryNodeResponse类
|
||||
一切regulatory_node回复的父类
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
class RegulatoryNodeRequest(RequestModel):
|
||||
"""
|
||||
RegulatoryNodeRequest类
|
||||
向regulatory请求的父类
|
||||
"""
|
||||
pass
|
||||
|
||||
class ForUser(regulatoryNodeResponse):
|
||||
"""ForUser 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 ForUser 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
context: str = Field(
|
||||
...,
|
||||
description="对用户的回复，应当使用和蔼的语气进行回复。用于直接解答简单问题或返回最终报告。",
|
||||
)
|
||||
|
||||
|
||||
class ForConsciousnessNode(regulatoryNodeResponse):
|
||||
"""ForConsciousnessNode 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
|
||||
reasoning: str = Field(..., description="选择将任务移交意识节点的简短原因。")
|
||||
|
||||
|
||||
class TerminationMessage(BaseModel):
|
||||
"""TerminationMessage 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 TerminationMessage 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
platform: str
|
||||
user_name: str
|
||||
message: str
|
||||
|
||||
|
||||
class regulatoryNodeDeps(DepsModel):
|
||||
"""regulatoryNodeDeps 核心组件类。
|
||||
这是一个系统执行节点类，作为多智能体架构中的独立处理单元。它能够接收工作流上下文，根据内置的大模型策略进行意图理解和自主决策，从而驱动特定阶段的任务闭环。"""
|
||||
|
||||
class RegulatoryNodeDeps(DepsModel):
|
||||
"""
|
||||
RegulatoryNodeDeps类
|
||||
regulatory_node的依赖模型
|
||||
"""
|
||||
platform: str
|
||||
user_name: str
|
||||
time: str
|
||||
retry_count: int = 0
|
||||
error_history: str = ""
|
||||
|
||||
class MessageRequest(RequestModel):
|
||||
"""
|
||||
MessageRequest类
|
||||
任何消息渠道向regulatory_node发送消息请求的模型
|
||||
"""
|
||||
platform: Literal["client"]
|
||||
user_name: str
|
||||
platform_id: Optional[str]
|
||||
message: str
|
||||
|
||||
class MessageResponse(RegulatoryNodeResponse):
|
||||
"""
|
||||
MessageResponse类
|
||||
regulatory_node回复的模型
|
||||
"""
|
||||
platform: Optional[Literal["client"]] = Field(description="系统自动填入的platform")
|
||||
platform_id: Optional[str] = Field(description="系统自动填入的platform_id")
|
||||
reply_message: str = Field(...,description="模型回复的消息")
|
||||
|
||||
@@ -17,21 +17,24 @@ from pydantic import BaseModel
|
||||
|
||||
|
||||
class ResponseModel(BaseModel):
|
||||
"""ResponseModel 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 ResponseModel 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
"""
|
||||
ResponseModel类
|
||||
继承自pydantic的BaseModel类，是一切回复模型的父类
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class DepsModel(BaseModel):
|
||||
"""DepsModel 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 DepsModel 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
"""
|
||||
DepsModel类
|
||||
继承自pydantic的BaseModel类，是agent运行时依赖模型的父类
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class InputModel(BaseModel):
|
||||
"""InputModel 核心组件类。
|
||||
这是一个领域数据模型或功能封装类，承载了 InputModel 相关的内聚属性定义与状态维护。它的存在隔离了局部的业务复杂性，并对外提供了类型安全的访问接口。"""
|
||||
|
||||
class RequestModel(BaseModel):
|
||||
"""
|
||||
RequestModel类
|
||||
继承自pydantic的BaseModel类，是一切请求模型的父类
|
||||
"""
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user