908705107/ai-agent-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: improve agent collaboration observability

Browse Source
This commit is contained in:
cls_宁波本机
2026-06-15 11:33:08 +08:00
parent 7b7c82309c
commit d5f1a4510f
13 changed files with 593 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-fastapi/ai_platform/nodes/builtin/intent_node.py
+4 -8
View File
@@ -116,17 +116,12 @@ class IntentNode(BaseNode):
error='请至少配置一个意图',
)
if not model_id:
return NodeResult(
success=False,
error='请选择用于意图识别的模型',
)
llm_service = LLMService(context.db_session)
model_id = await llm_service.resolve_chat_model_id(model_id)
# 检查模型是否支持 Function Calling
supports_fc = self._check_model_supports_function_call(model_id)
llm_service = LLMService(context.db_session)
if supports_fc:
# 使用 Function Calling 方式（更准确）
intent_name, confidence, total_tokens = await self._execute_with_function_calling_async(
@@ -169,6 +164,7 @@ class IntentNode(BaseNode):
'intent': intent_name,
'confidence': confidence,
'matched_branch': next_node_id,
'model_id': model_id,
},
)
backend-fastapi/ai_platform/nodes/builtin/llm_node.py
+17
View File
@@ -163,6 +163,7 @@ class LLMNode(BaseNode):
# 调用 LLM（异步）
llm_service = LLMService(context.db_session)
model_id = await llm_service.resolve_chat_model_id(model_id)
output_var = self.config.get('output_variable', 'llm_response')
# 根据输出模式选择执行方式
@@ -172,6 +173,11 @@ class LLMNode(BaseNode):
llm_service, model_id, messages, temperature, max_tokens,
output_schema, output_var, start_time
)
result.metadata = {
**(result.metadata or {}),
'model_id': str(model_id),
'output_variable': output_var,
}
return result
else:
# 普通文本输出模式（异步）
@@ -194,9 +200,11 @@ class LLMNode(BaseNode):
tokens_used=response.total_tokens,
elapsed_time=elapsed_time,
metadata={
'model_id': str(model_id),
'model': response.model,
'prompt_tokens': response.prompt_tokens,
'completion_tokens': response.completion_tokens,
'output_variable': output_var,
},
)
@@ -314,6 +322,11 @@ class LLMNode(BaseNode):
llm_service, model_id, messages, temperature, max_tokens,
output_schema, output_var, start_time
)
result.metadata = {
**(result.metadata or {}),
'model_id': str(model_id),
'output_variable': output_var,
}
return result
else:
# 普通文本输出模式
@@ -350,6 +363,10 @@ class LLMNode(BaseNode):
},
tokens_used=total_tokens,
elapsed_time=elapsed_time,
metadata={
'model_id': str(model_id),
'output_variable': output_var,
},
)
except Exception as e:
backend-fastapi/ai_platform/schemas/agent_schema.py
+1 -1
View File
@@ -251,7 +251,7 @@ class AgentMessageResponse(BaseModel):
content: str = ""
attachments: List[AttachmentResponse] = Field(default_factory=list)
status: str = "completed"
reasoning_steps: List[ReasoningStep] = Field(default_factory=list)
reasoning_steps: List[Dict[str, Any]] = Field(default_factory=list)
tool_calls: List[ToolCallRecord] = Field(default_factory=list)
prompt_tokens: int = 0
completion_tokens: int = 0
backend-fastapi/ai_platform/services/agent_service.py
+161 -3
View File
@@ -3,6 +3,7 @@
"""
import logging
import time
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy import select
@@ -45,7 +46,93 @@ class AgentService:
for key, value in self._agent_collaboration_metadata(agent).items():
collaboration.setdefault(key, value)
enriched["collaboration"] = collaboration
communication = dict(enriched.get("communication") or {})
actor = dict(communication.get("actor") or {})
for key in (
"agent_id",
"agent_code",
"agent_name",
"collaboration_role",
"collaboration_mode",
"model_id",
):
if collaboration.get(key):
actor.setdefault(key, collaboration[key])
communication.setdefault("event", enriched.get("type") or "")
communication.setdefault("status", "failed" if enriched.get("type") == "error" else "completed")
communication.setdefault("channel", "agent_chat")
communication.setdefault("actor", actor)
communication.setdefault(
"message",
enriched.get("content")
or enriched.get("message")
or enriched.get("accumulated_content")
or "",
)
enriched["communication"] = communication
return enriched
@staticmethod
def _event_to_reasoning_step(event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
event_type = event.get('type')
if event_type not in {
'action',
'annotation_reply',
'error',
'knowledge_retrieval',
'loop_complete',
'loop_iteration_complete',
'loop_iteration_error',
'loop_iteration_start',
'node_complete',
'node_start',
'observation',
'parallel_complete',
'parallel_start',
'thought',
'tool_call',
'tool_result',
'waiting_input',
}:
return None
collaboration = event.get('collaboration') or {}
communication = event.get('communication') or {}
config = event.get('waiting_config') or event.get('config') or {}
content = (
event.get('node_label')
or event.get('content')
or communication.get('message')
or config.get('title')
or config.get('question')
or event.get('node_type')
or event_type
)
step = {
'type': event_type,
'content': content,
'tool': event.get('tool') or '',
'params': event.get('params') or config,
'node_id': event.get('node_id') or '',
'node_type': event.get('node_type') or '',
'branch_id': event.get('branch_id') or collaboration.get('branch_id') or '',
'branch_label': event.get('branch_label') or collaboration.get('branch_label') or '',
'agent_code': collaboration.get('agent_code') or event.get('agent_code') or '',
'agent_name': collaboration.get('agent_name') or event.get('agent_name') or '',
'model': collaboration.get('model') or event.get('model') or '',
'model_id': collaboration.get('model_id') or event.get('model_id') or '',
'subflow_name': collaboration.get('subflow_name') or event.get('subflow_name') or '',
'from_subflow': bool(collaboration.get('from_subflow') or event.get('from_subflow')),
'collaboration_role': collaboration.get('collaboration_role') or '',
'collaboration_mode': collaboration.get('collaboration_mode') or event.get('collaboration_mode') or '',
'communication': communication,
'output': event.get('outputs') or event.get('output'),
'status': 'running' if event_type in {'node_start', 'parallel_start', 'waiting_input', 'loop_iteration_start'} else 'completed',
'timestamp': datetime.now().isoformat(),
}
return {key: value for key, value in step.items() if value not in (None, '', {}, [])}
async def chat(
self,
@@ -209,16 +296,31 @@ class AgentService:
)
self._db.add(assistant_msg)
await self._db.flush()
reasoning_steps: List[Dict[str, Any]] = []
yield {
yield self._attach_agent_collaboration({
'type': 'start',
'conversation_id': str(conversation.id),
'message_id': str(assistant_msg.id),
}
}, agent)
try:
# 构建系统提示词
system_prompt = self._build_system_prompt_simple(agent)
model_id = await self._resolve_agent_model_id(agent)
if not model_id:
raise ValueError('未找到可用的 chat 模型，请先在模型配置中启用一个模型')
event = self._attach_agent_collaboration({
'type': 'thought',
'content': '智能体开始分析用户需求并调用默认 chat 模型',
'model_id': model_id,
}, agent)
step = self._event_to_reasoning_step(event)
if step:
reasoning_steps.append(step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
yield event
# 标注直接回复检查（高相似度时跳过 LLM，直接返回标注答案）
annotation_reply = await self._check_annotation_direct_reply(agent, user_message)
@@ -257,11 +359,16 @@ class AgentService:
knowledge_context = await self._retrieve_knowledge_context(agent, user_message)
if knowledge_context:
system_prompt = self._inject_knowledge_context(system_prompt, knowledge_context)
yield {
event = {
'type': 'knowledge_retrieval',
'content': knowledge_context['summary'],
'result_count': knowledge_context['result_count'],
}
step = self._event_to_reasoning_step(event)
if step:
reasoning_steps.append(step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
yield event
# 获取对话历史
history = await self._get_conversation_history(conversation, limit=10)
@@ -297,6 +404,7 @@ class AgentService:
'type': 'llm_chunk',
'content': chunk.content,
'accumulated_content': accumulated_content,
'model_id': model_id,
}
if chunk.is_finished:
@@ -317,6 +425,7 @@ class AgentService:
yield {
'type': 'answer',
'content': final_answer,
'model_id': model_id,
}
# 更新助手消息
@@ -325,6 +434,7 @@ class AgentService:
assistant_msg.status = 'completed'
assistant_msg.total_tokens = total_tokens
assistant_msg.elapsed_time = elapsed_time
assistant_msg.reasoning_steps = reasoning_steps.copy()
# 更新对话统计
conversation.message_count = (conversation.message_count or 0) + 2
@@ -352,6 +462,10 @@ class AgentService:
logger.exception(f'Agent chat error: {e}')
assistant_msg.status = 'failed'
assistant_msg.error_message = str(e)
error_step = self._event_to_reasoning_step({'type': 'error', 'content': str(e)})
if error_step:
reasoning_steps.append(error_step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
await self._db.commit()
yield {
'type': 'error',
@@ -438,6 +552,7 @@ class AgentService:
)
self._db.add(assistant_msg)
await self._db.flush()
reasoning_steps: List[Dict[str, Any]] = []
yield {
'type': 'start',
@@ -507,6 +622,25 @@ class AgentService:
output = (event.get('outputs') or {}).get('output', '')
if output:
final_content = output
step = self._event_to_reasoning_step(event)
if step:
existing_index = next(
(
index
for index, item in enumerate(reasoning_steps)
if item.get('node_id')
and item.get('node_id') == step.get('node_id')
and item.get('type') == 'node_start'
and step.get('type') == 'node_complete'
),
-1,
)
if existing_index >= 0:
reasoning_steps[existing_index] = step
else:
reasoning_steps.append(step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
yield event
else:
@@ -573,6 +707,25 @@ class AgentService:
output = (event.get('outputs') or {}).get('output', '')
if output:
final_content = output
step = self._event_to_reasoning_step(event)
if step:
existing_index = next(
(
index
for index, item in enumerate(reasoning_steps)
if item.get('node_id')
and item.get('node_id') == step.get('node_id')
and item.get('type') == 'node_start'
and step.get('type') == 'node_complete'
),
-1,
)
if existing_index >= 0:
reasoning_steps[existing_index] = step
else:
reasoning_steps.append(step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
yield event
@@ -585,6 +738,7 @@ class AgentService:
assistant_msg.status = 'completed'
assistant_msg.elapsed_time = elapsed_time
assistant_msg.total_tokens = total_tokens
assistant_msg.reasoning_steps = reasoning_steps.copy()
# 更新对话统计
conversation.message_count = (conversation.message_count or 0) + 2
@@ -614,6 +768,10 @@ class AgentService:
logger.exception(f'Dialog flow error: {e}')
assistant_msg.status = 'failed'
assistant_msg.error_message = str(e)
error_step = self._event_to_reasoning_step({'type': 'error', 'content': str(e)})
if error_step:
reasoning_steps.append(error_step)
assistant_msg.reasoning_steps = reasoning_steps.copy()
await self._db.commit()
yield {
'type': 'error',
backend-fastapi/ai_platform/services/llm_service.py
+45 -7
View File
@@ -29,6 +29,38 @@ class LLMService:
def __init__(self, db: Optional[AsyncSession] = None):
self._provider_cache: Dict[str, BaseLLMProvider] = {}
self._db = db
@staticmethod
def _is_missing_model_id(model_id: Optional[str]) -> bool:
if model_id is None:
return True
return str(model_id).strip().lower() in {"", "none", "null", "undefined"}
async def resolve_chat_model_id(self, model_id: Optional[str]) -> str:
"""解析 chat 模型。未指定时使用当前启用的默认 chat 模型。"""
if not self._is_missing_model_id(model_id):
return str(model_id)
if not self._db:
raise ValueError("未找到可用的 chat 模型，请先在模型配置中启用一个模型")
from ai_platform.models import LLMModel
result = await self._db.execute(
select(LLMModel)
.where(
LLMModel.is_deleted == False,
LLMModel.is_active == True,
LLMModel.model_type == "chat",
)
.order_by(LLMModel.sort.desc(), LLMModel.sys_create_datetime.desc())
)
model = result.scalars().first()
if not model:
raise ValueError("未找到可用的 chat 模型，请先在模型配置中启用一个模型")
logger.info("No model_id supplied, fallback to default chat model %s", model.id)
return str(model.id)
async def _get_provider_async(self, model_id: str) -> tuple:
"""
@@ -45,6 +77,8 @@ class LLMService:
if not self._db:
raise ValueError("数据库会话未初始化")
model_id = await self.resolve_chat_model_id(model_id)
# 查询模型
result = await self._db.execute(
select(LLMModel).where(
@@ -319,14 +353,18 @@ class LLMService:
async def get_provider():
return await self._get_provider_async(model_id)
loop = asyncio.get_event_loop()
if loop.is_running():
# 如果已有事件循环在运行，使用线程池
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(asyncio.run, get_provider())
provider, model_name = future.result()
try:
loop = asyncio.get_event_loop()
except RuntimeError:
provider, model_name = asyncio.run(get_provider())
else:
provider, model_name = loop.run_until_complete(get_provider())
if loop.is_running():
# 如果已有事件循环在运行，使用线程池
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(asyncio.run, get_provider())
provider, model_name = future.result()
else:
provider, model_name = loop.run_until_complete(get_provider())
# 使用获取到的 provider 进行流式调用
yield from self.chat_stream_sync_with_provider(
backend-fastapi/ai_platform/services/workflow_service.py
+199 -7
View File
@@ -50,10 +50,48 @@ def _node_label_from_map(node_map: Dict[str, Any], node_id: str) -> str:
return data.get('label') or node.get('label') or node_id
def _safe_preview(value: Any, max_length: int = 220) -> str:
if value is None or value == '':
return ''
if isinstance(value, str):
text = value
else:
import json
try:
text = json.dumps(value, ensure_ascii=False)
except TypeError:
text = str(value)
text = text.strip()
return text if len(text) <= max_length else f'{text[:max_length]}...'
def _node_collaboration_mode(node_type: str, data: Dict[str, Any]) -> str:
if data.get('agent_code'):
return 'agent'
if node_type == 'parallel':
return 'parallel'
if node_type == 'subflow':
return 'subflow'
if node_type == 'intent':
return 'router'
if node_type in ('confirm', 'question', 'choice'):
return 'human_input'
if node_type == 'llm':
return 'llm'
return 'workflow'
def _node_collaboration_metadata(node_map: Dict[str, Any], node_id: str, node_type: str) -> dict:
node = node_map.get(node_id) or {}
data = node.get('data') or {}
metadata: Dict[str, Any] = {}
metadata: Dict[str, Any] = {
'node_id': node_id,
'node_type': node_type,
'node_label': _node_label_from_map(node_map, node_id),
'collaboration_role': data.get('label') or node_id,
'collaboration_mode': _node_collaboration_mode(node_type, data),
}
agent_code = (data.get('agent_code') or '').strip()
if agent_code:
@@ -83,15 +121,96 @@ def _node_collaboration_metadata(node_map: Dict[str, Any], node_id: str, node_ty
if branch.get('id')
]
if node_type in ('template', 'llm') and agent_code:
metadata['collaboration_role'] = data.get('label') or node_id
if node_type == 'subflow':
metadata['collaboration_role'] = data.get('label') or node_id
return metadata
def _build_communication_metadata(
node_map: Dict[str, Any],
node_id: str,
node_type: str,
*,
event_type: str,
status: Optional[str] = None,
branch_id: Optional[str] = None,
branch_label: Optional[str] = None,
target_node_id: Optional[str] = None,
output: Any = None,
error: Any = None,
waiting_config: Optional[Dict[str, Any]] = None,
) -> dict:
actor = _node_collaboration_metadata(node_map, node_id, node_type)
message = _safe_preview(error or output)
if waiting_config:
message = (
waiting_config.get('title')
or waiting_config.get('question')
or waiting_config.get('content')
or message
)
communication: Dict[str, Any] = {
'event': event_type,
'status': status or '',
'actor': {
key: actor.get(key)
for key in (
'node_id',
'node_type',
'node_label',
'agent_code',
'agent_name',
'collaboration_role',
'collaboration_mode',
)
if actor.get(key)
},
'message': _safe_preview(message),
}
if branch_id:
communication['channel'] = 'parallel_branch'
communication['branch_id'] = branch_id
communication['branch_label'] = branch_label or branch_id
elif node_type == 'subflow':
communication['channel'] = 'subflow'
elif node_type in ('confirm', 'question', 'choice') or event_type == 'waiting_input':
communication['channel'] = 'human_input'
else:
communication['channel'] = 'workflow_edge'
if target_node_id:
target_type = (node_map.get(target_node_id) or {}).get('type', '')
target = _node_collaboration_metadata(node_map, target_node_id, target_type)
communication['target'] = {
key: target.get(key)
for key in (
'node_id',
'node_type',
'node_label',
'agent_code',
'agent_name',
'collaboration_role',
'collaboration_mode',
)
if target.get(key)
}
return communication
def _resolve_handoff_target(
current_node_id: str,
result: Optional[NodeResult],
edge_map: Dict[str, List[str]],
parallel_edge_map: Dict[str, Dict[str, str]],
node_map: Dict[str, Any],
) -> Optional[str]:
if result and result.next_node_id:
if result.next_node_id in parallel_edge_map.get(current_node_id, {}):
return parallel_edge_map[current_node_id][result.next_node_id]
if result.next_node_id in node_map:
return result.next_node_id
next_nodes = edge_map.get(current_node_id, [])
return next_nodes[0] if next_nodes else None
def _make_execution_log_entry(
node_map: Dict[str, Any],
node_id: str,
@@ -100,8 +219,25 @@ def _make_execution_log_entry(
) -> dict:
metadata = dict(extra.pop('metadata', {}) or {})
collaboration = _node_collaboration_metadata(node_map, node_id, node_type)
branch_id = extra.get('branch') or extra.get('branch_id')
branch_label = extra.get('branch_label')
if branch_id:
collaboration['branch_id'] = branch_id
collaboration['branch_label'] = branch_label or branch_id
if collaboration:
metadata['collaboration'] = collaboration
metadata['communication'] = _build_communication_metadata(
node_map,
node_id,
node_type,
event_type='node_complete',
status=extra.get('status'),
branch_id=branch_id,
branch_label=branch_label,
target_node_id=extra.get('target_node_id'),
output=extra.get('output'),
error=extra.get('error'),
)
return {
'node_id': node_id,
'node_type': node_type,
@@ -133,6 +269,10 @@ def _event_collaboration_metadata(
'model_id',
'output_variable',
'subflow_name',
'prompt_tokens',
'completion_tokens',
'confidence',
'matched_branch',
):
value = result_metadata.get(key)
if value is not None and value != '':
@@ -159,8 +299,27 @@ def _merge_event_collaboration(
result: Optional[NodeResult] = None,
) -> Dict[str, Any]:
collaboration = _event_collaboration_metadata(node_map, node_id, node_type, result)
if event.get('branch_id'):
collaboration['branch_id'] = event.get('branch_id')
collaboration['branch_label'] = event.get('branch_label') or event.get('branch_id')
if collaboration:
event['collaboration'] = collaboration
event_error = None
if event.get('type') == 'error':
event_error = event.get('error_message') or event.get('message') or event.get('content')
event['communication'] = _build_communication_metadata(
node_map,
node_id,
node_type,
event_type=event.get('type') or '',
status=event.get('status'),
branch_id=event.get('branch_id'),
branch_label=event.get('branch_label'),
target_node_id=event.get('target_node_id'),
output=(event.get('outputs') or {}).get('output') if isinstance(event.get('outputs'), dict) else event.get('outputs'),
error=event_error,
waiting_config=event.get('waiting_config') or event.get('config'),
)
return event
@@ -653,6 +812,13 @@ class AIWorkflowService:
start_time = time.time()
result = await node_instance.execute_async(context)
elapsed = int((time.time() - start_time) * 1000)
target_node_id = _resolve_handoff_target(
current_node_id,
result,
edge_map,
parallel_edge_map,
node_map,
)
# 记录日志
logs.append(_make_execution_log_entry(
@@ -666,6 +832,7 @@ class AIWorkflowService:
tokens_used=result.tokens_used,
metadata=_node_result_metadata(result),
inputs=copy.deepcopy(node_inputs),
target_node_id=target_node_id,
))
total_tokens += result.tokens_used
@@ -819,6 +986,13 @@ class AIWorkflowService:
start_time = time.time()
result = await node_instance.execute_async(branch_context)
elapsed = int((time.time() - start_time) * 1000)
target_node_id = _resolve_handoff_target(
current_id,
result,
edge_map,
parallel_edge_map,
node_map,
)
branch_logs.append(_make_execution_log_entry(
node_map,
@@ -832,6 +1006,7 @@ class AIWorkflowService:
metadata=_node_result_metadata(result),
branch=branch_id,
branch_label=branch_labels.get(branch_id, branch_id),
target_node_id=target_node_id,
))
branch_tokens += result.tokens_used
@@ -1072,6 +1247,7 @@ class AIWorkflowService:
'output': result.output,
'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
},
'target_node_id': target_node_id,
}, result), node_map, current_id, node_type, result)
if not result.success:
@@ -1782,6 +1958,13 @@ class AIWorkflowService:
result = await node_instance.execute_async(context)
elapsed = int((time.time() - start_time) * 1000)
target_node_id = _resolve_handoff_target(
current_node_id,
result,
edge_map,
parallel_edge_map,
node_map,
)
# 处理节点事件（如消息节点发送消息）
if result.events:
@@ -1815,7 +1998,9 @@ class AIWorkflowService:
status='waiting',
output=result.output,
elapsed_time=elapsed,
metadata=_node_result_metadata(result),
inputs=copy.deepcopy(node_inputs),
target_node_id=target_node_id,
)
logs.append(log_entry)
@@ -1838,6 +2023,7 @@ class AIWorkflowService:
'node_id': current_node_id,
'node_type': node_type,
'config': result.waiting_config,
'waiting_config': result.waiting_config,
}, node_map, current_node_id, node_type, result)
# 暂停工作流执行，等待用户输入后续流
@@ -1855,6 +2041,7 @@ class AIWorkflowService:
tokens_used=result.tokens_used,
metadata=_node_result_metadata(result),
inputs=copy.deepcopy(node_inputs),
target_node_id=target_node_id,
)
logs.append(log_entry)
@@ -1882,6 +2069,7 @@ class AIWorkflowService:
'output': result.output,
'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
},
'target_node_id': target_node_id,
}, result), node_map, current_node_id, node_type, result)
if not result.success:
@@ -2335,6 +2523,7 @@ class AIWorkflowService:
output=result.output,
elapsed_time=elapsed,
metadata=_node_result_metadata(result),
target_node_id=target_node_id,
)
logs.append(log_entry)
@@ -2353,6 +2542,7 @@ class AIWorkflowService:
'node_id': current_node_id,
'node_type': node_type,
'config': result.waiting_config,
'waiting_config': result.waiting_config,
}, node_map, current_node_id, node_type, result)
# 如果在循环中，添加迭代信息
if loop_state:
@@ -2371,6 +2561,7 @@ class AIWorkflowService:
elapsed_time=elapsed,
tokens_used=result.tokens_used,
metadata=_node_result_metadata(result),
target_node_id=target_node_id,
)
logs.append(log_entry)
@@ -2401,6 +2592,7 @@ class AIWorkflowService:
'output': result.output,
'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
},
'target_node_id': target_node_id,
}, result), node_map, current_node_id, node_type, result)
# 如果在循环中，添加迭代信息
if loop_state: