mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 12:25:29 +08:00
319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""WeKnora Agent default runner implementation.
|
|
|
|
Ported from LangBot's legacy weknora-api provider runner and adapted to
|
|
Runner protocol v1.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import typing
|
|
|
|
from langbot_plugin.api.definition.components.runner.runner import Runner
|
|
from langbot_plugin.api.entities.builtin.provider.message import FunctionCall, Message, MessageChunk, ToolCall
|
|
from langbot_plugin.api.entities.builtin.runner import (
|
|
RunnerContext,
|
|
RunnerResult,
|
|
)
|
|
from pkg.errors import WeKnoraAPIError, WeKnoraConfigError
|
|
from pkg.scoped_identity import scoped_identity
|
|
from pkg.weknora_client import AsyncWeKnoraClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DefaultRunner(Runner):
|
|
"""Real Runner for WeKnora API.
|
|
|
|
Configuration is read from ctx.config using the legacy weknora-api field names
|
|
so host-side migration can map old pipeline configs directly.
|
|
|
|
Runtime state:
|
|
- external.session_id: WeKnora session id for the current LangBot conversation.
|
|
"""
|
|
|
|
def _validate_config(self, ctx: RunnerContext) -> dict[str, typing.Any]:
|
|
config = ctx.config or {}
|
|
|
|
app_type = config.get("app-type", "agent")
|
|
valid_app_types = ["chat", "agent"]
|
|
if app_type not in valid_app_types:
|
|
raise WeKnoraConfigError(
|
|
f"Invalid app-type: {app_type}. Must be one of {valid_app_types}",
|
|
code="weknora.config_invalid",
|
|
)
|
|
|
|
api_key = str(config.get("api-key", "")).strip()
|
|
if not api_key:
|
|
raise WeKnoraConfigError("api-key is required", code="weknora.config_invalid")
|
|
|
|
base_url = str(config.get("base-url", "")).strip()
|
|
if not base_url:
|
|
raise WeKnoraConfigError("base-url is required", code="weknora.config_invalid")
|
|
|
|
knowledge_base_ids = config.get("knowledge-base-ids", [])
|
|
if knowledge_base_ids is None:
|
|
knowledge_base_ids = []
|
|
if not isinstance(knowledge_base_ids, list):
|
|
raise WeKnoraConfigError("knowledge-base-ids must be a list", code="weknora.config_invalid")
|
|
|
|
if any(not isinstance(item, str) or not item.strip() for item in knowledge_base_ids):
|
|
raise WeKnoraConfigError("knowledge-base-ids must contain nonblank strings", code="weknora.config_invalid")
|
|
default_agent = "builtin-quick-answer" if app_type == "chat" else "builtin-smart-reasoning"
|
|
agent_id = config.get("agent-id", default_agent)
|
|
if agent_id is not None and not isinstance(agent_id, str):
|
|
raise WeKnoraConfigError("agent-id must be a string or null", code="weknora.config_invalid")
|
|
|
|
return {
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"app_type": app_type,
|
|
"agent_id": agent_id,
|
|
"knowledge_base_ids": list(knowledge_base_ids),
|
|
"web_search_enabled": bool(config.get("web-search-enabled", False)),
|
|
"timeout": float(config.get("timeout", 120)),
|
|
"base_prompt": str(config.get("base-prompt", "")),
|
|
}
|
|
|
|
def _get_user_tag(self, ctx: RunnerContext) -> str:
|
|
actor = ctx.actor
|
|
if actor and actor.actor_id:
|
|
return scoped_identity(self, ctx, f"{actor.actor_type}_{actor.actor_id}")
|
|
raise WeKnoraConfigError(
|
|
"vendor user identity requires a trusted Host actor identity",
|
|
code="weknora.identity_unavailable",
|
|
)
|
|
|
|
def _get_input_text(self, ctx: RunnerContext, base_prompt: str) -> str:
|
|
text = ctx.input.to_text()
|
|
return text if text else base_prompt
|
|
|
|
def _should_stream(self, ctx: RunnerContext) -> bool:
|
|
configured = (ctx.config or {}).get("streaming")
|
|
if configured is not None:
|
|
return bool(configured)
|
|
return bool(ctx.delivery.supports_streaming or ctx.runtime.metadata.get("streaming_supported", False))
|
|
|
|
async def _ensure_session_id(
|
|
self,
|
|
ctx: RunnerContext,
|
|
client: AsyncWeKnoraClient,
|
|
timeout: float,
|
|
user_tag: str,
|
|
) -> tuple[str, bool]:
|
|
session_id = ctx.state.conversation.get("external.session_id")
|
|
if session_id:
|
|
return str(session_id), False
|
|
|
|
session_id = await client.create_session(
|
|
title=f"IM Chat - {user_tag}",
|
|
timeout=min(30, timeout),
|
|
)
|
|
return session_id, True
|
|
|
|
async def run(self, ctx: RunnerContext) -> typing.AsyncGenerator[RunnerResult, None]:
|
|
"""Run the WeKnora app."""
|
|
try:
|
|
user_tag = self._get_user_tag(ctx) # Validate identity before constructing any upstream client.
|
|
config = self._validate_config(ctx)
|
|
except WeKnoraConfigError as e:
|
|
yield RunnerResult.run_failed(ctx.run_id, error=e.message, code=e.code)
|
|
return
|
|
|
|
client = AsyncWeKnoraClient(
|
|
api_key=config["api_key"],
|
|
base_url=config["base_url"],
|
|
)
|
|
|
|
try:
|
|
session_id, _created = await self._ensure_session_id(ctx, client, config["timeout"], user_tag)
|
|
input_text = self._get_input_text(ctx, config["base_prompt"])
|
|
|
|
if config["app_type"] == "agent":
|
|
results = self._run_runner_chat(ctx, client, config, session_id, input_text, user_tag)
|
|
else:
|
|
results = self._run_knowledge_chat(ctx, client, config, session_id, input_text, user_tag)
|
|
|
|
async for result in results:
|
|
yield result
|
|
|
|
yield RunnerResult.state_updated(
|
|
ctx.run_id,
|
|
"external.session_id",
|
|
session_id,
|
|
scope="conversation",
|
|
)
|
|
yield RunnerResult.run_completed(ctx.run_id)
|
|
except WeKnoraAPIError as e:
|
|
yield RunnerResult.run_failed(
|
|
ctx.run_id,
|
|
error=e.message,
|
|
code=e.code,
|
|
retryable=getattr(e, "retryable", False),
|
|
)
|
|
return
|
|
except Exception as e:
|
|
logger.exception(f"WeKnora runner unexpected error: {e}")
|
|
yield RunnerResult.run_failed(
|
|
ctx.run_id,
|
|
error=f"WeKnora runner error: {e}",
|
|
code="weknora.unexpected_error",
|
|
)
|
|
return
|
|
|
|
async def _run_runner_chat(
|
|
self,
|
|
ctx: RunnerContext,
|
|
client: AsyncWeKnoraClient,
|
|
config: dict[str, typing.Any],
|
|
session_id: str,
|
|
input_text: str,
|
|
user_tag: str,
|
|
) -> typing.AsyncGenerator[RunnerResult, None]:
|
|
pending_answer = ""
|
|
message_count = 0
|
|
is_final = False
|
|
saw_chunk = False
|
|
should_stream = self._should_stream(ctx)
|
|
|
|
async for chunk in client.agent_chat(
|
|
session_id=session_id,
|
|
query=input_text,
|
|
user=user_tag,
|
|
agent_id=config["agent_id"],
|
|
knowledge_base_ids=config["knowledge_base_ids"],
|
|
web_search_enabled=config["web_search_enabled"],
|
|
timeout=config["timeout"],
|
|
):
|
|
saw_chunk = True
|
|
logger.debug(f"WeKnora agent chunk: {chunk}")
|
|
|
|
response_type = chunk.get("response_type", "")
|
|
content = chunk.get("content", "")
|
|
done = bool(chunk.get("done", False))
|
|
|
|
if response_type == "tool_call":
|
|
tool_data = chunk.get("data", {})
|
|
tool_name = tool_data.get("tool_name", "") if isinstance(tool_data, dict) else ""
|
|
if tool_name:
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(
|
|
role="assistant",
|
|
tool_calls=[
|
|
ToolCall(
|
|
id=str(chunk.get("id", "")),
|
|
type="function",
|
|
function=FunctionCall(
|
|
name=str(tool_name),
|
|
arguments=json.dumps(tool_data.get("arguments", {})),
|
|
),
|
|
)
|
|
],
|
|
),
|
|
)
|
|
continue
|
|
|
|
if response_type == "answer":
|
|
message_count += 1
|
|
if content:
|
|
if len(pending_answer) + len(str(content)) > 1024 * 1024:
|
|
raise WeKnoraAPIError("WeKnora response exceeds the runtime limit")
|
|
pending_answer += str(content)
|
|
if done:
|
|
is_final = True
|
|
if should_stream and pending_answer and (message_count % 8 == 0 or is_final):
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=pending_answer, is_final=is_final),
|
|
)
|
|
continue
|
|
|
|
if response_type == "error":
|
|
raise WeKnoraAPIError(f"WeKnora service error: {content}", code="weknora.api_error")
|
|
|
|
if not saw_chunk:
|
|
raise WeKnoraAPIError("WeKnora API returned no response", code="weknora.empty_response")
|
|
|
|
if not pending_answer:
|
|
raise WeKnoraAPIError("WeKnora API returned an empty answer", code="weknora.empty_response")
|
|
|
|
if pending_answer and (not should_stream or not is_final):
|
|
if should_stream:
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=pending_answer, is_final=True),
|
|
)
|
|
else:
|
|
yield RunnerResult.message_completed(
|
|
ctx.run_id,
|
|
Message(role="assistant", content=pending_answer),
|
|
)
|
|
|
|
async def _run_knowledge_chat(
|
|
self,
|
|
ctx: RunnerContext,
|
|
client: AsyncWeKnoraClient,
|
|
config: dict[str, typing.Any],
|
|
session_id: str,
|
|
input_text: str,
|
|
user_tag: str,
|
|
) -> typing.AsyncGenerator[RunnerResult, None]:
|
|
pending_answer = ""
|
|
message_count = 0
|
|
is_final = False
|
|
saw_chunk = False
|
|
should_stream = self._should_stream(ctx)
|
|
|
|
async for chunk in client.knowledge_chat(
|
|
session_id=session_id,
|
|
query=input_text,
|
|
user=user_tag,
|
|
agent_id=config["agent_id"],
|
|
knowledge_base_ids=config["knowledge_base_ids"],
|
|
timeout=config["timeout"],
|
|
):
|
|
saw_chunk = True
|
|
logger.debug(f"WeKnora chat chunk: {chunk}")
|
|
|
|
response_type = chunk.get("response_type", "")
|
|
content = chunk.get("content", "")
|
|
done = bool(chunk.get("done", False))
|
|
|
|
if response_type == "answer":
|
|
message_count += 1
|
|
if content:
|
|
if len(pending_answer) + len(str(content)) > 1024 * 1024:
|
|
raise WeKnoraAPIError("WeKnora response exceeds the runtime limit")
|
|
pending_answer += str(content)
|
|
if done:
|
|
is_final = True
|
|
if should_stream and pending_answer and (message_count % 8 == 0 or is_final):
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=pending_answer, is_final=is_final),
|
|
)
|
|
continue
|
|
|
|
if response_type == "error":
|
|
raise WeKnoraAPIError(f"WeKnora service error: {content}", code="weknora.api_error")
|
|
|
|
if not saw_chunk:
|
|
raise WeKnoraAPIError("WeKnora API returned no response", code="weknora.empty_response")
|
|
|
|
if not pending_answer:
|
|
raise WeKnoraAPIError("WeKnora API returned an empty answer", code="weknora.empty_response")
|
|
|
|
if pending_answer and (not should_stream or not is_final):
|
|
if should_stream:
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=pending_answer, is_final=True),
|
|
)
|
|
else:
|
|
yield RunnerResult.message_completed(
|
|
ctx.run_id,
|
|
Message(role="assistant", content=pending_answer),
|
|
)
|