qzl/social-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend): 更新 agent 服务和配置层

Browse Source
This commit is contained in:
zl-q
2026-03-19 00:52:12 +08:00
parent c8e51d1329
commit 522b53c993
10 changed files with 189 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/schemas/agent/__init__.py
+2 -14
View File
@@ -3,17 +3,11 @@ from schemas.agent.forwarded_props import (
parse_forwarded_props_client_time,
)
from schemas.agent.runtime_models import (
AgentOutput,
ResultType,
RouterAgentOutput,
RouterUiDecision,
RunStatus,
ToolAgentOutput,
ToolStatus,
UiMode,
WorkerAgentOutput,
WorkerAgentOutputLite,
WorkerAgentOutputRich,
resolve_worker_output_model,
)
from schemas.agent.system_agent import AgentType, SystemAgentLLMConfig
from schemas.agent.ui_hints import (
@@ -26,23 +20,17 @@ from schemas.agent.ui_hints import (
__all__ = [
"AgentType",
"AgentOutput",
"ClientTimeContext",
"ResultType",
"RouterAgentOutput",
"RouterUiDecision",
"RunStatus",
"SystemAgentLLMConfig",
"ToolAgentOutput",
"ToolStatus",
"UiMode",
"UiHintAction",
"UiHintIntent",
"UiHintSection",
"UiHintStatus",
"UiHintsPayload",
"WorkerAgentOutputLite",
"WorkerAgentOutputRich",
"WorkerAgentOutput",
"resolve_worker_output_model",
"parse_forwarded_props_client_time",
]
backend/src/schemas/agent/runtime_models.py
+18 -330
View File
@@ -8,22 +8,6 @@ from pydantic import BaseModel, ConfigDict, Field
from schemas.agent.ui_hints import UiHintsPayload
class TaskType(str, Enum):
KNOWLEDGE = "knowledge"
RECOMMENDATION = "recommendation"
PLANNING = "planning"
SCHEDULING = "scheduling"
REMINDER_MANAGEMENT = "reminder_management"
TODO_MANAGEMENT = "todo_management"
COMMUNICATION_DRAFTING = "communication_drafting"
INFORMATION_ORGANIZATION = "information_organization"
STATUS_TRACKING = "status_tracking"
TRANSACTION_ASSIST = "transaction_assist"
ACTION_EXECUTION = "action_execution"
TROUBLESHOOTING = "troubleshooting"
UNKNOWN = "unknown"
class ResultType(str, Enum):
DIRECT_ANSWER = "direct_answer"
OPTIONS_WITH_RECOMMENDATION = "options_with_recommendation"
@@ -42,59 +26,6 @@ class ResultType(str, Enum):
UNKNOWN = "unknown"
class TaskTyping(BaseModel):
model_config = ConfigDict(extra="forbid")
primary: TaskType = Field(
...,
description=(
"Primary task category. Choose the single category that best "
"represents the core user intent."
),
examples=["planning"],
)
secondary: list[TaskType] = Field(
default_factory=list,
description=(
"Secondary task categories. Keep only strongly relevant supporting "
"categories, up to 3."
),
examples=[["scheduling", "action_execution"]],
)
class ResultTyping(BaseModel):
model_config = ConfigDict(extra="forbid")
primary: ResultType = Field(
...,
description=(
"Primary output type. It should match the execution mode and user "
"expectation; avoid unknown whenever possible."
),
examples=["action_plan"],
)
secondary: list[ResultType] = Field(
default_factory=list,
description=(
"Secondary output types. Use for compatible alternative response "
"shapes, up to 3."
),
examples=[["todo_list", "summary"]],
)
class ExecutionMode(str, Enum):
ONESTEP = "onestep"
TOOL_ASSISTED = "tool_assisted"
MULTISTEP = "multistep"
class UiMode(str, Enum):
NONE = "none"
RICH = "rich"
class RunStatus(str, Enum):
SUCCESS = "success"
PARTIAL_SUCCESS = "partial_success"
@@ -107,276 +38,33 @@ class ToolStatus(str, Enum):
PARTIAL = "partial"
class KeyEntity(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str = Field(
...,
description="Entity name, such as meeting/contact/location/project.",
)
type: str = Field(
...,
description="Entity type label, such as person/date/location/task.",
)
value: str | None = Field(
default=None,
description="Normalized entity value. Keep null if normalization is uncertain.",
)
class ConstraintItem(BaseModel):
model_config = ConfigDict(extra="forbid")
key: str = Field(
...,
description="Constraint key, such as deadline/budget/channel/privacy.",
)
value: str = Field(
...,
description="Constraint value in concise natural language or normalized form.",
)
required: bool = Field(
default=True,
description=(
"Whether this constraint is mandatory. True means execution cannot "
"proceed if violated."
),
)
class NormalizedTaskInput(BaseModel):
model_config = ConfigDict(extra="forbid")
user_text: str = Field(
...,
description="Normalized core user request text.",
examples=["Reschedule tomorrow's 9am standup to 3pm and notify attendees."],
)
multimodal_summary: list[str] = Field(
default_factory=list,
description="Key points extracted by router from images or attachments.",
examples=[["Screenshot shows a calendar conflict at 09:00."]],
)
class RouterUiDecision(BaseModel):
model_config = ConfigDict(extra="forbid")
ui_mode: UiMode = Field(
...,
description=(
"UI rendering mode decision for downstream worker schema selection. "
"Use 'none' when plain text response is sufficient; use 'rich' "
"when structured UI hints are beneficial."
),
examples=["none", "rich"],
)
ui_decision_reason: str = Field(
...,
description=(
"Brief reason for UI mode decision, focused on user intent and "
"information complexity."
),
examples=[
"User asked a simple factual question; plain text is sufficient.",
"User needs actionable options and status blocks; rich UI helps scanning.",
],
)
class RouterAgentOutput(BaseModel):
model_config = ConfigDict(extra="forbid")
normalized_task_input: NormalizedTaskInput = Field(
...,
description=(
"Normalized task input for routing. Preserve user intent faithfully "
"without adding or dropping critical semantics."
),
examples=[
{
"user_text": "Reschedule tomorrow's 9am standup to 3pm and notify attendees.",
"multimodal_summary": ["Calendar screenshot indicates 09:00 conflict."],
}
],
)
key_entities: list[KeyEntity] = Field(
default_factory=list,
description=(
"Key entities directly relevant to task execution. Return an empty "
"list when confidence is low."
),
examples=[
[
{"name": "standup", "type": "event", "value": "team-standup"},
{
"name": "tomorrow 9am",
"type": "datetime",
"value": "2026-03-14T09:00:00+08:00",
},
{
"name": "3pm",
"type": "datetime",
"value": "2026-03-14T15:00:00+08:00",
},
]
],
)
constraints: list[ConstraintItem] = Field(
default_factory=list,
description=(
"Execution constraints, including explicit constraints and "
"high-confidence inferred constraints."
),
examples=[
[
{"key": "must_notify_attendees", "value": "true", "required": True},
{"key": "timezone", "value": "Asia/Shanghai", "required": True},
]
],
)
task_typing: TaskTyping = Field(
...,
description=(
"Task typing result used by downstream agents for strategy and "
"capability boundaries."
),
examples=[{"primary": "scheduling", "secondary": ["communication_drafting"]}],
)
execution_mode: ExecutionMode = Field(
...,
description=(
"Recommended execution mode: onestep/tool_assisted/multistep. It "
"must be feasible under current context and capabilities."
),
examples=["tool_assisted"],
)
result_typing: ResultTyping = Field(
...,
description=(
"Expected result typing used to constrain downstream output "
"structure and expression."
),
examples=[
{
"primary": "execution_report",
"secondary": ["summary", "options_with_recommendation"],
}
],
)
ui: RouterUiDecision = Field(
...,
description=(
"Router decision on whether downstream worker should use rich UI "
"schema or lightweight text-only schema."
),
examples=[
{
"ui_mode": "rich",
"ui_decision_reason": "The request includes multiple actionable outcomes and benefits from structured blocks.",
}
],
)
class ErrorInfo(BaseModel):
model_config = ConfigDict(extra="forbid")
code: str = Field(
...,
description="Stable error code for programmatic handling and analytics.",
)
message: str = Field(
...,
description="Human-readable error message for user or upstream agent.",
)
retryable: bool = Field(
default=False,
description="Whether retrying can likely resolve this error.",
)
details: dict[str, Any] | None = Field(
default=None,
description="Diagnostic details. Must not contain sensitive data or secrets.",
)
code: str = Field(..., description="Stable error code for programmatic handling.")
message: str = Field(..., description="Human-readable error message.")
retryable: bool = Field(default=False)
details: dict[str, Any] | None = Field(default=None)
class ToolAgentOutput(BaseModel):
model_config = ConfigDict(extra="forbid")
tool_name: str = Field(..., description="Invoked tool name.")
tool_call_id: str = Field(
..., description="Tool call identifier for this invocation."
)
tool_call_args: dict[str, Any] | None = Field(
default=None,
description="Snapshot of tool call arguments for traceability and debugging.",
)
status: ToolStatus = Field(..., description="Tool execution status.")
result: str = Field(
...,
description=(
"Compact machine-oriented tool result. Keep it short but include "
"action-critical facts (ids/status/counts) for downstream agent steps."
),
)
error: ErrorInfo | None = Field(
default=None, description="Tool execution error details."
)
tool_name: str
tool_call_id: str
tool_call_args: dict[str, Any] | None = None
status: ToolStatus
result: str
error: ErrorInfo | None = None
class WorkerAgentOutputLite(BaseModel):
class AgentOutput(BaseModel):
model_config = ConfigDict(extra="forbid")
status: RunStatus = Field(
default=RunStatus.SUCCESS,
description="Worker execution status: success/partial_success/failed.",
examples=["success"],
)
answer: str = Field(
...,
description=(
"Primary user-facing response text. Lead with conclusion, then "
"include only necessary details."
),
examples=[
"Done. I moved the standup to 3:00 PM tomorrow and prepared attendee notifications."
],
)
key_points: list[str] = Field(
default_factory=list,
description="Key point summary, recommended 0-5 items, one sentence each.",
examples=[["Original slot conflicted at 09:00.", "New slot set to 15:00."]],
)
result_type: ResultType = Field(
default=ResultType.UNKNOWN,
description="Structured result type of this response. Avoid unknown whenever possible.",
examples=["execution_report"],
)
suggested_actions: list[str] = Field(
default_factory=list,
description="Suggested next actions, 0-3 items, actionable and relevant.",
examples=[["Review attendee RSVP status after notifications are sent."]],
)
error: ErrorInfo | None = Field(
default=None,
description="Error information for failed or partially failed runs; null on success.",
)
class WorkerAgentOutputRich(WorkerAgentOutputLite):
ui_hints: UiHintsPayload | None = Field(
default=None,
description=(
"Optional expressive UI semantic annotations. Focus on information "
"and interaction intent, not concrete visual styling instructions."
),
)
WorkerAgentOutput = WorkerAgentOutputLite | WorkerAgentOutputRich
def resolve_worker_output_model(ui_mode: UiMode) -> type[WorkerAgentOutputLite]:
if ui_mode == UiMode.RICH:
return WorkerAgentOutputRich
return WorkerAgentOutputLite
status: RunStatus = Field(default=RunStatus.SUCCESS)
answer: str
key_points: list[str] = Field(default_factory=list)
result_type: ResultType = Field(default=ResultType.UNKNOWN)
suggested_actions: list[str] = Field(default_factory=list)
error: ErrorInfo | None = None
ui_hints: UiHintsPayload | None = None
backend/src/schemas/agent/system_agent.py
+38 -2
View File
@@ -2,15 +2,51 @@ from __future__ import annotations
from enum import Enum
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from core.agentscope.tools.tool_config import ToolGroup
class AgentType(str, Enum):
ROUTER = "router"
WORKER = "worker"
MEMORY = "memory"
class ContextBuildStrategy(str, Enum):
DAY = "day"
NUMBER = "number"
class ContextMessagesConfig(BaseModel):
mode: ContextBuildStrategy = ContextBuildStrategy.NUMBER
count: int = Field(default=20, ge=1, le=200)
class SystemAgentLLMConfig(BaseModel):
temperature: float | None = Field(default=None, ge=0.0, le=2.0)
max_tokens: int | None = Field(default=None, ge=1)
timeout_seconds: float | None = Field(default=30.0, gt=0.0, le=300.0)
context_messages: ContextMessagesConfig = Field(
default_factory=ContextMessagesConfig
)
enabled_tool_groups: list[ToolGroup] = Field(default_factory=list, max_length=8)
@field_validator("enabled_tool_groups", mode="before")
@classmethod
def _normalize_enabled_tool_groups(cls, value: object) -> list[ToolGroup]:
if value is None:
return []
if not isinstance(value, list):
raise ValueError("enabled_tool_groups must be a list")
normalized: list[ToolGroup] = []
for item in value:
if isinstance(item, ToolGroup):
group = item
else:
raw_group = str(item or "").strip().lower()
if not raw_group:
continue
group = ToolGroup(raw_group)
if group not in normalized:
normalized.append(group)
return normalized
backend/src/schemas/messages/chat_message.py
+2 -3
View File
@@ -6,7 +6,7 @@ from typing import Any, ClassVar
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
from schemas.agent.runtime_models import RouterAgentOutput, WorkerAgentOutputRich
from schemas.agent.runtime_models import AgentOutput
from ..agent import AgentType, ToolAgentOutput
@@ -24,9 +24,8 @@ class AgentChatMessageMetadata(BaseModel):
run_id: str
agent_type: AgentType | None = None
user_message_attachments: list[UserMessageAttachment] | None = None
router_agent_output: RouterAgentOutput | None = None
tool_agent_output: ToolAgentOutput | None = None
worker_agent_output: WorkerAgentOutputRich | None = None
agent_output: AgentOutput | None = None
class AgentChatMessage(BaseModel):