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

feat(agent-runner): finalize 4.x processor integration

Browse Source
This commit is contained in:
huanghuoguoguo
2026-07-12 15:44:05 +08:00
parent 29689962f3
commit e6384aae5d
109 changed files with 2200 additions and 1204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/langbot/pkg/agent/runner/config_migration.py
+15 -122
View File
@@ -5,27 +5,13 @@ from __future__ import annotations
import typing
LEGACY_RUNNER_ID_MAP: dict[str, str] = {
'local-agent': 'plugin:langbot/local-agent/default',
'dify-service-api': 'plugin:langbot/dify-agent/default',
'n8n-service-api': 'plugin:langbot/n8n-agent/default',
'coze-api': 'plugin:langbot/coze-agent/default',
'dashscope-app-api': 'plugin:langbot/dashscope-agent/default',
'deerflow-api': 'plugin:langbot/deerflow-agent/default',
'langflow-api': 'plugin:langbot/langflow-agent/default',
'tbox-app-api': 'plugin:langbot/tbox-agent/default',
'weknora-api': 'plugin:langbot/weknora-agent/default',
}
class ConfigMigration:
"""Configuration helper for agent runner IDs.
"""Configuration helpers for the current AgentRunner shape.
Responsibilities:
- Resolve runner ID from ai.runner.id
- Migrate legacy ai.runner.runner + ai.<runner-name> blocks
- Extract current Agent/runner config from ai.runner_config
- Keep the current config container shape stable on save
- Extract Agent/runner config from ai.runner_config
- Read current conversation expiry settings
"""
@staticmethod
@@ -38,18 +24,10 @@ class ConfigMigration:
Returns:
Runner ID string, or None if not configured
"""
ai_config = pipeline_config.get('ai', {})
runner_config = ai_config.get('runner', {})
runner_id = runner_config.get('id')
if runner_id:
return runner_id
legacy_runner = runner_config.get('runner')
if isinstance(legacy_runner, str):
return LEGACY_RUNNER_ID_MAP.get(legacy_runner)
return None
ai_config = pipeline_config.get('ai', {}) if isinstance(pipeline_config, dict) else {}
runner = ai_config.get('runner', {}) if isinstance(ai_config, dict) else {}
runner_id = runner.get('id') if isinstance(runner, dict) else None
return runner_id if isinstance(runner_id, str) and runner_id else None
@staticmethod
def resolve_runner_config(
@@ -65,20 +43,10 @@ class ConfigMigration:
Returns:
Runner configuration dict (empty if not found)
"""
ai_config = pipeline_config.get('ai', {})
runner_configs = ai_config.get('runner_config', {})
if runner_id in runner_configs:
return runner_configs[runner_id]
legacy_runner = ConfigMigration._legacy_runner_name_for_id(runner_id)
if legacy_runner and isinstance(ai_config.get(legacy_runner), dict):
return ConfigMigration._normalize_legacy_runner_config(
legacy_runner,
ai_config[legacy_runner],
)
return {}
ai_config = pipeline_config.get('ai', {}) if isinstance(pipeline_config, dict) else {}
runner_configs = ai_config.get('runner_config', {}) if isinstance(ai_config, dict) else {}
runner_config = runner_configs.get(runner_id, {}) if isinstance(runner_configs, dict) else {}
return runner_config if isinstance(runner_config, dict) else {}
@staticmethod
def get_expire_time(pipeline_config: dict[str, typing.Any]) -> int:
@@ -90,82 +58,7 @@ class ConfigMigration:
Returns:
Expire time in seconds (0 means no expiry)
"""
ai_config = pipeline_config.get('ai', {})
runner_config = ai_config.get('runner', {})
return runner_config.get('expire-time', 0)
@staticmethod
def migrate_pipeline_config(pipeline_config: dict[str, typing.Any]) -> dict[str, typing.Any]:
"""Normalize the current config container before saving.
Args:
pipeline_config: Original configuration
Returns:
Configuration with explicit ai.runner and ai.runner_config containers
"""
new_config = dict(pipeline_config)
if 'ai' not in new_config:
return new_config
ai_config = dict(new_config.get('ai', {}))
runner_config = dict(ai_config.get('runner', {}))
runner_configs = dict(ai_config.get('runner_config', {}))
legacy_runner = runner_config.get('runner')
mapped_runner_id = None
if isinstance(legacy_runner, str):
mapped_runner_id = LEGACY_RUNNER_ID_MAP.get(legacy_runner)
if mapped_runner_id and not runner_config.get('id'):
runner_config = {
key: value
for key, value in runner_config.items()
if key != 'runner'
}
runner_config['id'] = mapped_runner_id
if mapped_runner_id and mapped_runner_id not in runner_configs:
legacy_config = ai_config.get(legacy_runner)
if isinstance(legacy_config, dict):
runner_configs[mapped_runner_id] = ConfigMigration._normalize_legacy_runner_config(
legacy_runner,
legacy_config,
)
ai_config['runner'] = runner_config
ai_config['runner_config'] = runner_configs
if mapped_runner_id and legacy_runner in ai_config:
ai_config.pop(legacy_runner, None)
new_config['ai'] = ai_config
return new_config
@staticmethod
def _legacy_runner_name_for_id(runner_id: str) -> str | None:
for legacy_runner, mapped_runner_id in LEGACY_RUNNER_ID_MAP.items():
if mapped_runner_id == runner_id:
return legacy_runner
return None
@staticmethod
def _normalize_legacy_runner_config(
legacy_runner: str,
legacy_config: dict[str, typing.Any],
) -> dict[str, typing.Any]:
"""Normalize legacy runner config blocks to current plugin schema quirks."""
normalized = dict(legacy_config)
if legacy_runner == 'local-agent':
model = normalized.get('model')
if isinstance(model, str):
normalized['model'] = {
'primary': model,
'fallbacks': [],
}
knowledge_base = normalized.pop('knowledge-base', None)
if 'knowledge-bases' not in normalized and isinstance(knowledge_base, str):
normalized['knowledge-bases'] = [] if knowledge_base in {'', '__none__', '__none'} else [knowledge_base]
return normalized
ai_config = pipeline_config.get('ai', {}) if isinstance(pipeline_config, dict) else {}
runner = ai_config.get('runner', {}) if isinstance(ai_config, dict) else {}
expire_time = runner.get('expire-time', 0) if isinstance(runner, dict) else 0
return expire_time if isinstance(expire_time, int) else 0
src/langbot/pkg/agent/runner/host_models.py
+5 -5
View File
@@ -139,9 +139,9 @@ class DeliveryPolicy(pydantic.BaseModel):
class AgentConfig(pydantic.BaseModel):
"""Host-side Agent configuration.
Product-level Agent is the target replacement for Pipeline-owned agent
config. Current Pipeline entry paths can project their config into this
model during migration.
Product-level Agents are independent from Pipelines. A Pipeline entry path
can project its config into this runtime-only model without creating or
updating a persisted Agent.
"""
agent_id: str | None = None
@@ -175,8 +175,8 @@ class AgentConfig(pydantic.BaseModel):
class AgentBinding(pydantic.BaseModel):
"""Binding configuration for mapping events to runners.
This is Host-internal model for event-to-runner binding.
It replaces the old Pipeline runner config role.
This is a Host-internal, runtime-only model for event-to-runner binding.
Projecting Pipeline config into it is not a persistence migration.
"""
binding_id: str
src/langbot/pkg/api/http/service/agent.py
+1 -1
View File
@@ -17,7 +17,7 @@ AGENT_DEFAULT_EVENT_PATTERNS = ['*']
class AgentService:
"""Unified product surface for Agent processors and Pipelines."""
"""Unified processor facade for the peer Agent and Pipeline types."""
ap: app.Application
src/langbot/pkg/api/http/service/pipeline.py
-6
View File
@@ -200,16 +200,10 @@ class PipelineService:
return pipeline_data['uuid']
async def update_pipeline(self, pipeline_uuid: str, pipeline_data: dict) -> None:
from ....agent.runner.config_migration import ConfigMigration
pipeline_data = pipeline_data.copy()
for protected_field in ('uuid', 'for_version', 'stages', 'is_default'):
pipeline_data.pop(protected_field, None)
# Migrate config to new format before saving
if 'config' in pipeline_data:
pipeline_data['config'] = ConfigMigration.migrate_pipeline_config(pipeline_data['config'])
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(persistence_pipeline.LegacyPipeline)
.where(persistence_pipeline.LegacyPipeline.uuid == pipeline_uuid)
src/langbot/pkg/api/mcp/server.py
+15 -15
View File
@@ -164,32 +164,32 @@ class LangBotMCPServer:
await ap.pipeline_service.delete_pipeline(pipeline_uuid)
return _dump({'ok': True})
# ----- Agents -------------------------------------------------- #
# ----- Processors ---------------------------------------------- #
@mcp.tool(description='List product-level processors, including Agents and Pipelines.')
async def list_agents() -> str:
async def list_processors() -> str:
return _dump(await ap.agent_service.get_agents())
@mcp.tool(description='Get a product-level Agent or Pipeline by UUID.')
async def get_agent(agent_uuid: str) -> str:
return _dump(await ap.agent_service.get_agent(agent_uuid))
@mcp.tool(description='Get an Agent or Pipeline processor by UUID.')
async def get_processor(processor_uuid: str) -> str:
return _dump(await ap.agent_service.get_agent(processor_uuid))
@mcp.tool(
description=(
'Create an Agent processor or Pipeline. `agent_data` matches '
'POST /api/v1/agents; set kind to `agent` or `pipeline`. Returns the new UUID and kind.'
'Create an Agent or Pipeline processor. Set `processor_data.kind` to '
'`agent` or `pipeline`. Returns the new UUID and kind.'
)
)
async def create_agent(agent_data: dict) -> str:
return _dump(await ap.agent_service.create_agent(agent_data))
async def create_processor(processor_data: dict) -> str:
return _dump(await ap.agent_service.create_agent(processor_data))
@mcp.tool(description='Update an Agent processor or Pipeline by UUID.')
async def update_agent(agent_uuid: str, agent_data: dict) -> str:
await ap.agent_service.update_agent(agent_uuid, agent_data)
@mcp.tool(description='Update an Agent or Pipeline processor by UUID.')
async def update_processor(processor_uuid: str, processor_data: dict) -> str:
await ap.agent_service.update_agent(processor_uuid, processor_data)
return _dump({'ok': True})
@mcp.tool(description='Delete an Agent processor or Pipeline by UUID.')
async def delete_agent(agent_uuid: str) -> str:
await ap.agent_service.delete_agent(agent_uuid)
@mcp.tool(description='Delete an Agent or Pipeline processor by UUID.')
async def delete_processor(processor_uuid: str) -> str:
await ap.agent_service.delete_agent(processor_uuid)
return _dump({'ok': True})
# ----- Models -------------------------------------------------- #
src/langbot/pkg/persistence/alembic/versions/0005_migrate_runner_config.py
-88
View File
@@ -1,88 +0,0 @@
"""Normalize AgentRunner config containers
Revision ID: 0005_migrate_runner_config
Revises: 0005_add_llm_context_length
Create Date: 2026-05-10
"""
import json
import sqlalchemy as sa
from alembic import op
from langbot.pkg.agent.runner.config_migration import ConfigMigration
revision = '0005_migrate_runner_config'
down_revision = '0005_add_llm_context_length'
branch_labels = None
depends_on = None
def migrate_pipeline_config(config: dict) -> dict:
"""Migrate persisted pipeline config to the AgentRunner plugin shape."""
return ConfigMigration.migrate_pipeline_config(config)
def _load_config(config_value):
if isinstance(config_value, dict):
return config_value
if isinstance(config_value, str):
return json.loads(config_value)
return None
def _update_config(conn, table_name: str, pipeline_uuid: str, migrated_config: dict) -> None:
"""Write JSON config using a dialect-compatible bind."""
config_json = json.dumps(migrated_config)
if conn.dialect.name == 'postgresql':
conn.execute(
sa.text(
f'UPDATE {table_name} '
'SET config = CAST(:config AS JSON) '
'WHERE uuid = :uuid'
),
{'config': config_json, 'uuid': pipeline_uuid},
)
return
conn.execute(
sa.text(f'UPDATE {table_name} SET config = :config WHERE uuid = :uuid'),
{'config': config_json, 'uuid': pipeline_uuid},
)
def upgrade() -> None:
"""Normalize existing pipeline config containers."""
conn = op.get_bind()
inspector = sa.inspect(conn)
table_name = 'legacy_pipelines'
# Check if pipeline table exists (may not exist in fresh install)
if table_name not in inspector.get_table_names():
return
# Get all pipelines
result = conn.execute(sa.text(f'SELECT uuid, config FROM {table_name}'))
pipelines = result.fetchall()
for pipeline_uuid, config_json in pipelines:
if not config_json:
continue
try:
config = _load_config(config_json)
if not isinstance(config, dict):
continue
migrated_config = migrate_pipeline_config(config)
# Only update if config changed
if json.dumps(config, sort_keys=True) != json.dumps(migrated_config, sort_keys=True):
_update_config(conn, table_name, pipeline_uuid, migrated_config)
except Exception:
# Skip invalid configs
continue
def downgrade() -> None:
"""Downgrade is not supported for data migration."""
# No downgrade - keep configs in new format
pass
src/langbot/pkg/persistence/alembic/versions/0009_migrate_routing_to_event_bindings.py
+4 -1
View File
@@ -1,4 +1,7 @@
"""migrate use_pipeline_uuid and pipeline_routing_rules into event_bindings
"""Migrate legacy Bot routes into Pipeline event bindings.
Bindings keep referencing the original Pipeline UUIDs. This migration never
creates Agent rows or copies Pipeline runner configuration.
Revision ID: 0009_migrate_routing_to_event_bindings
Revises: 0008_agent_product_surface
src/langbot/pkg/persistence/alembic/versions/58846a8d7a81_add_event_log_and_transcript_tables.py
+2 -2
View File
@@ -1,7 +1,7 @@
"""add_event_log_and_transcript_tables
Revision ID: 58846a8d7a81
Revises: 0005_migrate_runner_config
Revises: 0005_add_llm_context_length
Create Date: 2026-05-23 15:41:47.030841
"""
from alembic import op
@@ -9,7 +9,7 @@ import sqlalchemy as sa
# revision identifiers
revision = '58846a8d7a81'
down_revision = '0005_migrate_runner_config'
down_revision = '0005_add_llm_context_length'
branch_labels = None
depends_on = None
src/langbot/pkg/persistence/migrations/dbm001_migrate_v3_config.py
-42
View File
@@ -11,7 +11,6 @@ from ...entity.persistence import (
pipeline as persistence_pipeline,
bot as persistence_bot,
)
from ...agent.runner.config_migration import LEGACY_RUNNER_ID_MAP
@migration.migration_class(1)
@@ -114,47 +113,6 @@ class DBMigrateV3Config(migration.DBMigration):
pipeline_config = default_pipeline['config']
# ai
ai_config = pipeline_config.setdefault('ai', {})
runner_name = self.ap.provider_cfg.data['runner']
runner_id = LEGACY_RUNNER_ID_MAP.get(runner_name, '')
ai_config['runner'] = {
'id': runner_id,
}
runner_configs = ai_config.setdefault('runner_config', {})
local_agent_runner_id = LEGACY_RUNNER_ID_MAP['local-agent']
local_agent_config = runner_configs.setdefault(local_agent_runner_id, {})
local_agent_config['model'] = {
'primary': model_uuid,
'fallbacks': [],
}
local_agent_config['prompt'] = [
{
'role': 'system',
'content': self.ap.provider_cfg.data['prompt']['default'],
}
]
runner_configs[LEGACY_RUNNER_ID_MAP['dify-service-api']] = {
'base-url': self.ap.provider_cfg.data['dify-service-api']['base-url'],
'app-type': self.ap.provider_cfg.data['dify-service-api']['app-type'],
'api-key': self.ap.provider_cfg.data['dify-service-api'][
self.ap.provider_cfg.data['dify-service-api']['app-type']
]['api-key'],
'thinking-convert': self.ap.provider_cfg.data['dify-service-api']['options']['convert-thinking-tips'],
'timeout': self.ap.provider_cfg.data['dify-service-api'][
self.ap.provider_cfg.data['dify-service-api']['app-type']
]['timeout'],
}
runner_configs[LEGACY_RUNNER_ID_MAP['dashscope-app-api']] = {
'app-type': self.ap.provider_cfg.data['dashscope-app-api']['app-type'],
'api-key': self.ap.provider_cfg.data['dashscope-app-api']['api-key'],
'references_quote': self.ap.provider_cfg.data['dashscope-app-api'][
self.ap.provider_cfg.data['dashscope-app-api']['app-type']
]['references_quote'],
}
# trigger
pipeline_config['trigger']['group-respond-rules'] = self.ap.pipeline_cfg.data['respond-rules']['default']
pipeline_config['trigger']['access-control'] = self.ap.pipeline_cfg.data['access-control']
src/langbot/pkg/pipeline/pipelinemgr.py
+11 -3
View File
@@ -14,6 +14,7 @@ import langbot_plugin.api.entities.builtin.platform.events as platform_events
import langbot_plugin.api.entities.events as events
from ..utils import importutil
from .config_coercion import coerce_pipeline_config
from ..agent.runner.config_migration import ConfigMigration
import langbot_plugin.api.entities.builtin.provider.session as provider_session
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -94,12 +95,19 @@ class RuntimePipeline:
extensions_prefs = pipeline_entity.extensions_preferences or {}
self.enable_all_plugins = extensions_prefs.get('enable_all_plugins', True)
self.enable_all_mcp_servers = extensions_prefs.get('enable_all_mcp_servers', True)
local_agent_config = (pipeline_entity.config or {}).get('ai', {}).get('local-agent', {})
self.mcp_resource_attachments = local_agent_config.get(
pipeline_config = pipeline_entity.config or {}
runner_config: dict[str, typing.Any] = {}
runner_id = ConfigMigration.resolve_runner_id(pipeline_config) if isinstance(pipeline_config, dict) else None
if runner_id:
resolved_runner_config = ConfigMigration.resolve_runner_config(pipeline_config, runner_id)
if isinstance(resolved_runner_config, dict):
runner_config = resolved_runner_config
self.mcp_resource_attachments = runner_config.get(
'mcp-resources',
extensions_prefs.get('mcp_resources', []),
)
self.mcp_resource_agent_read_enabled = local_agent_config.get(
self.mcp_resource_agent_read_enabled = runner_config.get(
'mcp-resource-agent-read-enabled',
extensions_prefs.get('mcp_resource_agent_read_enabled', True),
)
src/langbot/pkg/platform/botmgr.py
+3 -3
View File
@@ -630,7 +630,7 @@ class RuntimeBot:
text=f'Test event {event_type} dispatched from control plane',
)
test_adapter = SyntheticRouteTestAdapter(self.adapter)
outcome = await self._dispatch_eba_event_to_agent(
outcome = await self._dispatch_eba_event_to_processor(
event,
typing.cast(abstract_platform_adapter.AbstractMessagePlatformAdapter, test_adapter),
)
@@ -1135,9 +1135,9 @@ class RuntimeBot:
except Exception:
await self.logger.error(f'Failed to dispatch platform event to plugins: {traceback.format_exc()}')
await self._dispatch_eba_event_to_agent(event, adapter)
await self._dispatch_eba_event_to_processor(event, adapter)
async def _dispatch_eba_event_to_agent(
async def _dispatch_eba_event_to_processor(
self,
event: platform_events.EBAEvent,
adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,