xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-07-15 08:56:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add tool call observability

Browse Source
This commit is contained in:
Hyu
2026-07-02 15:52:46 +08:00
parent b1bc05d5d3
commit c809e3d14f
19 changed files with 1351 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/langbot/pkg/api/http/controller/groups/monitoring.py
+45
View File
@@ -138,6 +138,39 @@ class MonitoringRouterGroup(group.RouterGroup):
}
)
@self.route('/tool-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def get_tool_calls() -> str:
"""Get tool call records"""
bot_ids = quart.request.args.getlist('botId')
pipeline_ids = quart.request.args.getlist('pipelineId')
session_ids = quart.request.args.getlist('sessionId')
start_time_str = quart.request.args.get('startTime')
end_time_str = quart.request.args.get('endTime')
limit = int(quart.request.args.get('limit', 100))
offset = int(quart.request.args.get('offset', 0))
start_time = parse_iso_datetime(start_time_str)
end_time = parse_iso_datetime(end_time_str)
tool_calls, total = await self.ap.monitoring_service.get_tool_calls(
bot_ids=bot_ids if bot_ids else None,
pipeline_ids=pipeline_ids if pipeline_ids else None,
session_ids=session_ids if session_ids else None,
start_time=start_time,
end_time=end_time,
limit=limit,
offset=offset,
)
return self.success(
data={
'tool_calls': tool_calls,
'total': total,
'limit': limit,
'offset': offset,
}
)
@self.route('/embedding-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def get_embedding_calls() -> str:
"""Get embedding call records"""
@@ -284,6 +317,16 @@ class MonitoringRouterGroup(group.RouterGroup):
offset=0,
)
# Get tool calls
tool_calls, tool_calls_total = await self.ap.monitoring_service.get_tool_calls(
bot_ids=bot_ids if bot_ids else None,
pipeline_ids=pipeline_ids if pipeline_ids else None,
start_time=start_time,
end_time=end_time,
limit=limit,
offset=0,
)
# Get sessions
sessions, sessions_total = await self.ap.monitoring_service.get_sessions(
bot_ids=bot_ids if bot_ids else None,
@@ -318,12 +361,14 @@ class MonitoringRouterGroup(group.RouterGroup):
'overview': overview,
'messages': messages,
'llmCalls': llm_calls,
'toolCalls': tool_calls,
'embeddingCalls': embedding_calls,
'sessions': sessions,
'errors': errors,
'totalCount': {
'messages': messages_total,
'llmCalls': llm_calls_total,
'toolCalls': tool_calls_total,
'embeddingCalls': embedding_calls_total,
'sessions': sessions_total,
'errors': errors_total,
src/langbot/pkg/api/http/service/maintenance.py
+1
View File
@@ -243,6 +243,7 @@ class MaintenanceService:
tables = {
'messages': persistence_monitoring.MonitoringMessage.id,
'llm_calls': persistence_monitoring.MonitoringLLMCall.id,
'tool_calls': persistence_monitoring.MonitoringToolCall.id,
'embedding_calls': persistence_monitoring.MonitoringEmbeddingCall.id,
'errors': persistence_monitoring.MonitoringError.id,
'sessions': persistence_monitoring.MonitoringSession.session_id,
src/langbot/pkg/api/http/service/monitoring.py
+208
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import uuid
import datetime
import json
import sqlalchemy
from ....core import app
@@ -50,6 +51,12 @@ class MonitoringService:
persistence_monitoring.MonitoringLLMCall.timestamp,
persistence_monitoring.MonitoringLLMCall.id,
),
(
'monitoring_tool_calls',
persistence_monitoring.MonitoringToolCall,
persistence_monitoring.MonitoringToolCall.timestamp,
persistence_monitoring.MonitoringToolCall.id,
),
(
'monitoring_embedding_calls',
persistence_monitoring.MonitoringEmbeddingCall,
@@ -131,6 +138,68 @@ class MonitoringService:
await autocommit_conn.execute(sqlalchemy.text('PRAGMA wal_checkpoint(TRUNCATE)'))
await autocommit_conn.execute(sqlalchemy.text('VACUUM'))
def _serialize_tool_payload(self, payload: object, max_length: int = 20000) -> str | None:
"""Serialize tool arguments/results for monitoring storage."""
if payload is None:
return None
if isinstance(payload, str):
text = payload
else:
try:
text = json.dumps(payload, ensure_ascii=False, default=str)
except Exception:
text = str(payload)
if len(text) <= max_length:
return text
return f'{text[:max_length]}... [truncated {len(text) - max_length} chars]'
async def _get_message_for_tool_context(
self,
message_id: str | None = None,
session_id: str | None = None,
):
if message_id:
result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.select(persistence_monitoring.MonitoringMessage).where(
persistence_monitoring.MonitoringMessage.id == message_id
)
)
row = result.first()
if row:
return row[0]
if not session_id:
return None
user_query = (
sqlalchemy.select(persistence_monitoring.MonitoringMessage)
.where(
sqlalchemy.and_(
persistence_monitoring.MonitoringMessage.session_id == session_id,
persistence_monitoring.MonitoringMessage.role == 'user',
)
)
.order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
.limit(1)
)
result = await self.ap.persistence_mgr.execute_async(user_query)
row = result.first()
if row:
return row[0]
any_query = (
sqlalchemy.select(persistence_monitoring.MonitoringMessage)
.where(persistence_monitoring.MonitoringMessage.session_id == session_id)
.order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
.limit(1)
)
result = await self.ap.persistence_mgr.execute_async(any_query)
row = result.first()
return row[0] if row else None
# ========== Recording Methods ==========
async def record_message(
@@ -220,6 +289,57 @@ class MonitoringService:
return call_id
async def record_tool_call(
self,
tool_name: str,
tool_source: str,
duration: int,
status: str = 'success',
bot_id: str | None = None,
bot_name: str | None = None,
pipeline_id: str | None = None,
pipeline_name: str | None = None,
session_id: str | None = None,
message_id: str | None = None,
arguments: object | None = None,
result: object | None = None,
error_message: str | None = None,
) -> str:
"""Record a tool call."""
context_message = await self._get_message_for_tool_context(message_id=message_id, session_id=session_id)
if context_message:
bot_id = bot_id or context_message.bot_id
bot_name = bot_name or context_message.bot_name
pipeline_id = pipeline_id or context_message.pipeline_id
pipeline_name = pipeline_name or context_message.pipeline_name
session_id = session_id or context_message.session_id
message_id = message_id or context_message.id
call_id = str(uuid.uuid4())
call_data = {
'id': call_id,
'timestamp': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
'tool_name': tool_name,
'tool_source': tool_source,
'duration': max(0, duration),
'status': status,
'bot_id': bot_id or 'unknown',
'bot_name': bot_name or 'Unknown',
'pipeline_id': pipeline_id or 'unknown',
'pipeline_name': pipeline_name or 'Unknown',
'session_id': session_id,
'message_id': message_id,
'arguments': self._serialize_tool_payload(arguments),
'result': self._serialize_tool_payload(result),
'error_message': self._serialize_tool_payload(error_message),
}
await self.ap.persistence_mgr.execute_async(
sqlalchemy.insert(persistence_monitoring.MonitoringToolCall).values(call_data)
)
return call_id
async def record_embedding_call(
self,
model_name: str,
@@ -749,6 +869,58 @@ class MonitoringService:
total,
)
async def get_tool_calls(
self,
bot_ids: list[str] | None = None,
pipeline_ids: list[str] | None = None,
session_ids: list[str] | None = None,
start_time: datetime.datetime | None = None,
end_time: datetime.datetime | None = None,
limit: int = 100,
offset: int = 0,
) -> tuple[list[dict], int]:
"""Get tool calls with filters"""
conditions = []
if bot_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.bot_id.in_(bot_ids))
if pipeline_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.pipeline_id.in_(pipeline_ids))
if session_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.session_id.in_(session_ids))
if start_time:
conditions.append(persistence_monitoring.MonitoringToolCall.timestamp >= start_time)
if end_time:
conditions.append(persistence_monitoring.MonitoringToolCall.timestamp <= end_time)
count_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringToolCall.id))
if conditions:
count_query = count_query.where(sqlalchemy.and_(*conditions))
count_result = await self.ap.persistence_mgr.execute_async(count_query)
total = count_result.scalar() or 0
query = sqlalchemy.select(persistence_monitoring.MonitoringToolCall).order_by(
persistence_monitoring.MonitoringToolCall.timestamp.desc()
)
if conditions:
query = query.where(sqlalchemy.and_(*conditions))
query = query.limit(limit).offset(offset)
result = await self.ap.persistence_mgr.execute_async(query)
tool_calls_rows = result.all()
return (
[
self.ap.persistence_mgr.serialize_model(
persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
)
for row in tool_calls_rows
],
total,
)
async def get_embedding_calls(
self,
start_time: datetime.datetime | None = None,
@@ -971,6 +1143,34 @@ class MonitoringService:
else:
error_llm_calls += 1
# Get tool calls for this session
tool_query = (
sqlalchemy.select(persistence_monitoring.MonitoringToolCall)
.where(persistence_monitoring.MonitoringToolCall.session_id == session_id)
.order_by(persistence_monitoring.MonitoringToolCall.timestamp.asc())
)
tool_result = await self.ap.persistence_mgr.execute_async(tool_query)
tool_rows = tool_result.all()
tool_calls = [
self.ap.persistence_mgr.serialize_model(
persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
)
for row in tool_rows
]
total_tool_calls = len(tool_rows)
success_tool_calls = 0
error_tool_calls = 0
total_tool_duration = 0
for row in tool_rows:
tool_call = row[0] if isinstance(row, tuple) else row
total_tool_duration += tool_call.duration
if tool_call.status == 'success':
success_tool_calls += 1
else:
error_tool_calls += 1
# Get errors for this session
error_query = (
sqlalchemy.select(persistence_monitoring.MonitoringError)
@@ -1014,6 +1214,14 @@ class MonitoringService:
'total_tokens': total_tokens,
'average_duration_ms': int(total_duration / total_llm_calls) if total_llm_calls > 0 else 0,
},
'tool_calls': tool_calls,
'tool_stats': {
'total_calls': total_tool_calls,
'success_calls': success_tool_calls,
'error_calls': error_tool_calls,
'total_duration_ms': total_tool_duration,
'average_duration_ms': int(total_tool_duration / total_tool_calls) if total_tool_calls > 0 else 0,
},
'errors': errors,
'session_duration_seconds': session_duration_seconds,
}
src/langbot/pkg/entity/persistence/monitoring.py
+22
View File
@@ -49,6 +49,28 @@ class MonitoringLLMCall(Base):
message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) # Associated message ID
class MonitoringToolCall(Base):
"""Tool call records"""
__tablename__ = 'monitoring_tool_calls'
id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, index=True)
tool_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
tool_source = sqlalchemy.Column(sqlalchemy.String(50), nullable=False) # native, plugin, mcp, skill
duration = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) # milliseconds
status = sqlalchemy.Column(sqlalchemy.String(50), nullable=False) # success, error
bot_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
bot_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
pipeline_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
arguments = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
result = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
class MonitoringSession(Base):
"""Session tracking records"""
src/langbot/pkg/persistence/migrations/dbm026_monitoring_tool_calls.py
+17
View File
@@ -0,0 +1,17 @@
from langbot.pkg.entity.persistence import monitoring as persistence_monitoring
from .. import migration
@migration.migration_class(26)
class DBMigrateMonitoringToolCalls(migration.DBMigration):
"""Add monitoring_tool_calls table"""
async def upgrade(self):
"""Upgrade"""
async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.create, checkfirst=True)
async def downgrade(self):
"""Downgrade"""
async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.drop, checkfirst=True)
src/langbot/pkg/provider/tools/toolmgr.py
+114 -4
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import typing
import time
from typing import TYPE_CHECKING
import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
@@ -142,21 +143,130 @@ class ToolManager:
return tools
def _get_query_session_id(self, query: pipeline_query.Query) -> str | None:
launcher_type = getattr(query, 'launcher_type', None)
launcher_id = getattr(query, 'launcher_id', None)
if launcher_type is None or launcher_id is None:
return None
launcher_type_value = launcher_type.value if hasattr(launcher_type, 'value') else launcher_type
return f'{launcher_type_value}_{launcher_id}'
async def _record_tool_call(
self,
*,
name: str,
source: str,
parameters: dict,
query: pipeline_query.Query,
duration_ms: int,
status: str,
result: typing.Any = None,
error_message: str | None = None,
) -> None:
monitoring_service = getattr(self.ap, 'monitoring_service', None)
if not monitoring_service:
return
variables = getattr(query, 'variables', {}) or {}
message_id = variables.get('_monitoring_message_id') if isinstance(variables, dict) else None
bot_name = variables.get('_monitoring_bot_name') if isinstance(variables, dict) else None
pipeline_name = variables.get('_monitoring_pipeline_name') if isinstance(variables, dict) else None
try:
await monitoring_service.record_tool_call(
tool_name=name,
tool_source=source,
duration=duration_ms,
status=status,
bot_id=getattr(query, 'bot_uuid', None),
bot_name=bot_name,
pipeline_name=pipeline_name,
session_id=self._get_query_session_id(query),
message_id=message_id,
arguments=parameters,
result=result,
error_message=error_message,
)
except Exception as e:
self.ap.logger.warning(f'Failed to record tool call: {e}')
async def _invoke_tool_with_monitoring(
self,
*,
source: str,
name: str,
parameters: dict,
query: pipeline_query.Query,
invoke: typing.Callable[[], typing.Awaitable[typing.Any]],
) -> typing.Any:
start_time = time.perf_counter()
try:
result = await invoke()
except Exception as e:
duration_ms = int((time.perf_counter() - start_time) * 1000)
await self._record_tool_call(
name=name,
source=source,
parameters=parameters,
query=query,
duration_ms=duration_ms,
status='error',
error_message=str(e),
)
raise
duration_ms = int((time.perf_counter() - start_time) * 1000)
await self._record_tool_call(
name=name,
source=source,
parameters=parameters,
query=query,
duration_ms=duration_ms,
status='success',
result=result,
)
return result
async def execute_func_call(self, name: str, parameters: dict, query: pipeline_query.Query) -> typing.Any:
from langbot.pkg.telemetry import features as telemetry_features
if await self.native_tool_loader.has_tool(name):
telemetry_features.increment(query, 'tool_calls', 'native')
return await self.native_tool_loader.invoke_tool(name, parameters, query)
return await self._invoke_tool_with_monitoring(
source='native',
name=name,
parameters=parameters,
query=query,
invoke=lambda: self.native_tool_loader.invoke_tool(name, parameters, query),
)
if await self.plugin_tool_loader.has_tool(name):
telemetry_features.increment(query, 'tool_calls', 'plugin')
return await self.plugin_tool_loader.invoke_tool(name, parameters, query)
return await self._invoke_tool_with_monitoring(
source='plugin',
name=name,
parameters=parameters,
query=query,
invoke=lambda: self.plugin_tool_loader.invoke_tool(name, parameters, query),
)
if await self.mcp_tool_loader.has_tool(name):
telemetry_features.increment(query, 'tool_calls', 'mcp')
return await self.mcp_tool_loader.invoke_tool(name, parameters, query)
return await self._invoke_tool_with_monitoring(
source='mcp',
name=name,
parameters=parameters,
query=query,
invoke=lambda: self.mcp_tool_loader.invoke_tool(name, parameters, query),
)
if await self.skill_tool_loader.has_tool(name):
telemetry_features.increment(query, 'tool_calls', 'skill')
return await self.skill_tool_loader.invoke_tool(name, parameters, query)
return await self._invoke_tool_with_monitoring(
source='skill',
name=name,
parameters=parameters,
query=query,
invoke=lambda: self.skill_tool_loader.invoke_tool(name, parameters, query),
)
raise ToolNotFoundError(name)
async def shutdown(self):