diff --git a/configs/skills/citation_detector.yaml b/configs/skills/citation_detector.yaml index 285720b..2a6c488 100644 --- a/configs/skills/citation_detector.yaml +++ b/configs/skills/citation_detector.yaml @@ -9,6 +9,14 @@ supported_tasks: max_concurrency: 3 custom_handler: "configs.geo_handlers.handle_citation_task" +intent: + keywords: ["引用检测", "引用分析", "AI引用", "citation", "引用率", "被引用"] + description: "用户需要检测品牌在各AI平台回答中的引用情况" + examples: + - "检测我们的品牌在AI平台的引用情况" + - "分析品牌引用率" + - "哪些AI平台引用了我们" + input_schema: type: object properties: diff --git a/configs/skills/content_generator.yaml b/configs/skills/content_generator.yaml index c8c6081..01c0806 100644 --- a/configs/skills/content_generator.yaml +++ b/configs/skills/content_generator.yaml @@ -87,7 +87,7 @@ prompt: examples: "" llm: - model: "deepseek" + model: "default" temperature: 0.7 max_tokens: 4000 diff --git a/configs/skills/deai_agent.yaml b/configs/skills/deai_agent.yaml index a30a7d6..b352f0b 100644 --- a/configs/skills/deai_agent.yaml +++ b/configs/skills/deai_agent.yaml @@ -7,6 +7,14 @@ supported_tasks: - deai_process max_concurrency: 2 +intent: + keywords: ["去AI化", "去ai", "去AI", "人性化", "改写", "deai", "humanize", "自然化"] + description: "用户需要将AI生成的文本改写为更自然、人类化的表达" + examples: + - "帮我把这篇文章去AI化" + - "让这段文字更自然" + - "改写得像人写的" + input_schema: type: object required: @@ -61,7 +69,7 @@ prompt: examples: "" llm: - model: "deepseek" + model: "default" temperature: 0.9 max_tokens: 8000 diff --git a/configs/skills/geo_optimizer.yaml b/configs/skills/geo_optimizer.yaml index 389a73b..600b330 100644 --- a/configs/skills/geo_optimizer.yaml +++ b/configs/skills/geo_optimizer.yaml @@ -7,6 +7,14 @@ supported_tasks: - geo_optimize max_concurrency: 2 +intent: + keywords: ["GEO优化", "SEO优化", "内容优化", "优化文章", "geo", "seo", "optimize"] + description: "用户需要对文章进行GEO/SEO优化，提升在AI搜索引擎中的可见性" + examples: + - "帮我优化这篇文章的SEO" + - "GEO优化一下" + - "提升文章在AI搜索中的排名" + input_schema: type: object required: @@ -64,7 +72,7 @@ prompt: examples: "" llm: - model: "deepseek" + model: "default" temperature: 0.5 max_tokens: 8000 diff --git a/configs/skills/monitor.yaml b/configs/skills/monitor.yaml index 3dc599c..289881b 100644 --- a/configs/skills/monitor.yaml +++ b/configs/skills/monitor.yaml @@ -9,6 +9,14 @@ supported_tasks: max_concurrency: 3 custom_handler: "configs.geo_handlers.handle_monitor_task" +intent: + keywords: ["效果追踪", "监测", "监控", "monitor", "追踪", "排名变化"] + description: "用户需要监测品牌引用量、情感、排名变化" + examples: + - "监测品牌引用变化" + - "追踪效果" + - "品牌排名变化" + input_schema: type: object required: diff --git a/configs/skills/schema_advisor.yaml b/configs/skills/schema_advisor.yaml index 6da2166..1b63a02 100644 --- a/configs/skills/schema_advisor.yaml +++ b/configs/skills/schema_advisor.yaml @@ -8,6 +8,14 @@ supported_tasks: max_concurrency: 2 custom_handler: "configs.geo_handlers.handle_schema_task" +intent: + keywords: ["Schema", "结构化数据", "JSON-LD", "schema", "schema优化"] + description: "用户需要识别Schema缺失维度，生成结构化数据建议" + examples: + - "帮我优化Schema" + - "生成JSON-LD结构化数据" + - "Schema有什么可以改进的" + input_schema: type: object required: diff --git a/src/agentkit/chat/__init__.py b/src/agentkit/chat/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/agentkit/chat/skill_routing.py b/src/agentkit/chat/skill_routing.py new file mode 100644 index 0000000..4857ab8 --- /dev/null +++ b/src/agentkit/chat/skill_routing.py @@ -0,0 +1,168 @@ +"""Shared skill routing logic for GUI and CLI chat. + +Extracts the duplicated skill routing, @skill: prefix parsing, +and prompt assembly into a single module used by both chat routes. +""" + +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass, field +from typing import Any + +logger = logging.getLogger(__name__) + +# Strict validation: only lowercase alphanumeric, hyphens, underscores +_SKILL_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") + + +def validate_skill_name(name: str) -> str: + """Validate and normalize a skill name. Raises ValueError on invalid input.""" + normalized = name.strip().lower() + if not _SKILL_NAME_RE.match(normalized): + raise ValueError( + f"Invalid skill name '{name}': must match [a-z0-9][a-z0-9_-]{{0,63}}" + ) + return normalized + + +@dataclass +class SkillRoutingResult: + """Result of skill routing for a user message.""" + + skill_name: str | None = None + skill_config: Any = None + skill_tools: list = field(default_factory=list) + clean_content: str = "" + system_prompt: str | None = None + tools: list = field(default_factory=list) + model: str = "default" + agent_name: str | None = None + matched: bool = False + match_method: str | None = None + match_confidence: float = 0.0 + + +def parse_skill_prefix(content: str) -> tuple[str | None, str]: + """Parse @skill:name prefix from user message. + + Returns (skill_name_or_None, clean_content). + """ + if not content.startswith("@skill:"): + return None, content + + parts = content.split(" ", 1) + skill_ref = parts[0][7:] # strip "@skill:" + explicit_skill = skill_ref.strip() + clean = parts[1].strip() if len(parts) > 1 else content[7 + len(skill_ref):].strip() + return explicit_skill, clean + + +def build_skill_system_prompt(skill_config) -> str | None: + """Build system prompt from skill config's prompt section.""" + if not skill_config or not skill_config.prompt: + return None + prompt_parts = [] + for key in ("identity", "context", "instructions", "constraints", "output_format"): + val = skill_config.prompt.get(key) + if val: + prompt_parts.append(val) + return "\n\n".join(prompt_parts) if prompt_parts else None + + +async def resolve_skill_routing( + content: str, + skill_registry: Any, + intent_router: Any, + default_tools: list, + default_system_prompt: str | None, + default_model: str = "default", + default_agent_name: str = "default", + agent_tool_registry: Any = None, + session_id: str = "", +) -> SkillRoutingResult: + """Resolve skill routing for a user message. + + This is the shared entry point used by both GUI WebSocket chat and CLI chat. + Returns a SkillRoutingResult with all execution parameters set. + """ + result = SkillRoutingResult() + + # Parse @skill: prefix + explicit_skill, clean_content = parse_skill_prefix(content) + result.clean_content = clean_content + + if explicit_skill: + logger.info(f"Session {session_id}: explicit skill reference: {explicit_skill}") + + # Try explicit skill match + if explicit_skill and skill_registry: + try: + matched_skill = skill_registry.get(explicit_skill) + result.skill_name = explicit_skill + result.skill_config = matched_skill.config + result.skill_tools = matched_skill.tools or [] + result.matched = True + result.match_method = "explicit" + result.match_confidence = 1.0 + logger.info(f"Session {session_id}: using explicit skill '{explicit_skill}'") + except Exception as e: + logger.warning(f"Session {session_id}: explicit skill '{explicit_skill}' not found: {e}") + # Reset so we don't enter skill branch with stale data + result.skill_name = None + result.skill_config = None + + # Try IntentRouter if no explicit match + if not result.matched and skill_registry and intent_router: + skills = skill_registry.list_skills() + routable_skills = [s for s in skills if s.config.intent.keywords] + if routable_skills: + try: + routing_result = await intent_router.route( + input_data={"content": clean_content}, + skills=routable_skills, + ) + if routing_result and routing_result.confidence >= 0.5: + skill_name = routing_result.matched_skill + try: + matched_skill = skill_registry.get(skill_name) + result.skill_name = skill_name + result.skill_config = matched_skill.config + result.skill_tools = matched_skill.tools or [] + result.matched = True + result.match_method = routing_result.method + result.match_confidence = routing_result.confidence + logger.info( + f"Session {session_id}: routed to skill '{skill_name}' " + f"via {routing_result.method} (confidence={routing_result.confidence})" + ) + except Exception as e: + logger.warning(f"Session {session_id}: skill '{skill_name}' found by router but not in registry: {e}") + except Exception as e: + logger.warning(f"Skill routing failed for session {session_id}: {e}") + + # Determine execution parameters + if result.matched and result.skill_config: + skill_prompt = build_skill_system_prompt(result.skill_config) + result.system_prompt = skill_prompt or default_system_prompt + + # Merge skill tools with agent tools, deduplicating by name + agent_tools = agent_tool_registry.list_tools() if agent_tool_registry else default_tools + seen_names = set() + merged_tools = [] + for tool in result.skill_tools + agent_tools: + if tool.name not in seen_names: + seen_names.add(tool.name) + merged_tools.append(tool) + result.tools = merged_tools + + result.model = result.skill_config.llm.get("model", default_model) if result.skill_config.llm else default_model + result.agent_name = result.skill_name + else: + result.system_prompt = default_system_prompt + result.tools = default_tools + result.model = default_model + result.agent_name = default_agent_name + + return result diff --git a/src/agentkit/cli/chat.py b/src/agentkit/cli/chat.py index 074be1e..d715bf5 100644 --- a/src/agentkit/cli/chat.py +++ b/src/agentkit/cli/chat.py @@ -98,11 +98,37 @@ async def _chat_async( WebCrawlTool(), ] + # ── Load skills and build IntentRouter ─────────────────────── + from agentkit.tools.registry import ToolRegistry + from agentkit.skills.registry import SkillRegistry + from agentkit.skills.loader import SkillLoader + from agentkit.router.intent import IntentRouter + + tool_registry = ToolRegistry() + for tool in tools: + tool_registry.register(tool) + + skill_registry = SkillRegistry() + if server_config.skill_paths: + loader = SkillLoader(skill_registry=skill_registry, tool_registry=tool_registry) + for skill_path in server_config.skill_paths: + from pathlib import Path as _P + p = _P(skill_path) + if p.is_dir(): + loaded = loader.load_from_directory(str(p)) + if loaded: + rprint(f"[dim]Loaded {len(loaded)} skills from {p}[/dim]") + elif p.is_file() and p.suffix in (".yaml", ".yml"): + try: + loader.load_from_file(str(p)) + except Exception: + pass + + intent_router = IntentRouter(llm_gateway=gateway) if skill_registry.list_skills() else None + # Build system prompt — inject memory into system prompt base_prompt = system_prompt or ( - "You are a helpful AI assistant. " - "Remember the context of our conversation and refer back to earlier messages. " - "Respond clearly and concisely." + "你是一个有帮助的AI助手。请记住我们对话的上下文，并在后续对话中引用之前的内容。回答要清晰简洁，请使用中文回复。" ) effective_system_prompt = memory_store.build_system_prompt(memory_snapshot, base_prompt) @@ -185,6 +211,27 @@ async def _chat_async( # Get full conversation history (includes all previous turns) chat_messages = await session_manager.get_chat_messages(session.session_id) + # ── Skill routing ───────────────────────────────────────── + from agentkit.chat.skill_routing import resolve_skill_routing + + routing = await resolve_skill_routing( + content=user_input, + skill_registry=skill_registry, + intent_router=intent_router, + default_tools=tools, + default_system_prompt=effective_system_prompt, + default_model=current_model, + default_agent_name=agent_name, + session_id=session.session_id, + ) + + if routing.matched: + rprint(f"[dim]Skill: {routing.skill_name} ({routing.match_method}, {int(routing.match_confidence * 100)}%)[/dim]") + + exec_system_prompt = routing.system_prompt + exec_tools = routing.tools + exec_model = routing.model + # Print Agent label before streaming rprint(f"\n[bold blue]{agent_display_name}[/bold blue]: ", end="") @@ -194,10 +241,10 @@ async def _chat_async( # Non-streaming mode result = await react_engine.execute( messages=chat_messages, - tools=tools, - model=current_model, - agent_name=agent_name, - system_prompt=effective_system_prompt, + tools=exec_tools, + model=exec_model, + agent_name=routing.skill_name or agent_name, + system_prompt=exec_system_prompt, ) output = result.output if hasattr(result, "output") else str(result) rprint(output) @@ -219,10 +266,10 @@ async def _chat_async( ) as live: async for event in react_engine.execute_stream( messages=chat_messages, - tools=tools, - model=current_model, - agent_name=agent_name, - system_prompt=effective_system_prompt, + tools=exec_tools, + model=exec_model, + agent_name=routing.skill_name or agent_name, + system_prompt=exec_system_prompt, ): if event.event_type == "token": token = event.data.get("content", "") diff --git a/src/agentkit/cli/main.py b/src/agentkit/cli/main.py index 73403c3..7390cc2 100644 --- a/src/agentkit/cli/main.py +++ b/src/agentkit/cli/main.py @@ -30,6 +30,88 @@ from agentkit.cli.chat import chat # noqa: E402 app.command(name="chat")(chat) +@app.command() +def gui( + host: str = typer.Option("0.0.0.0", "--host", help="Server bind host"), + port: int = typer.Option(8002, "--port", help="Server port"), + config: Optional[str] = typer.Option(None, "--config", help="Path to agentkit.yaml"), + no_open: bool = typer.Option(False, "--no-open", help="Do not open browser automatically"), +): + """Start AgentKit with a web UI for chatting with your Agent""" + import os + import webbrowser + import uvicorn + + from agentkit.server.config import ServerConfig, find_config_path + from agentkit.cli.onboarding import run_onboarding + + # Load config + config_path = find_config_path(config) + + if config_path is None: + rprint("[yellow]No agentkit.yaml found.[/yellow]") + from rich.prompt import Confirm + if Confirm.ask("Would you like to run the setup wizard?", default=True): + config_path = run_onboarding(config_arg=config) + if config_path is None: + rprint("[red]Setup cancelled. Using defaults.[/red]") + else: + rprint("[dim]Using default configuration (no LLM providers).[/dim]") + + if config_path: + rprint(f"[green]Loading config from {config_path}[/green]") + server_config = ServerConfig.from_yaml(config_path) + + from pathlib import Path + dotenv = Path(config_path).parent / ".env" + server_config.load_dotenv(str(dotenv)) + server_config = ServerConfig.from_yaml(config_path) + + os.environ["AGENTKIT_CONFIG_PATH"] = config_path + + # Check if LLM API key is configured + if not server_config.has_llm_provider(): + rprint("[yellow]No LLM API key configured.[/yellow]") + from rich.prompt import Confirm + if Confirm.ask("Would you like to run the setup wizard?", default=True): + config_path = run_onboarding(config_arg=config) + if config_path is None: + rprint("[red]Setup cancelled. GUI may not function correctly without API key.[/red]") + else: + server_config = ServerConfig.from_yaml(config_path) + server_config.load_dotenv(str(dotenv)) + server_config = ServerConfig.from_yaml(config_path) + os.environ["AGENTKIT_CONFIG_PATH"] = config_path + else: + rprint("[dim]Continuing without LLM provider — chat will not work.[/dim]") + + # Signal to create_app that we want GUI mode (must be set before lifespan runs) + os.environ["AGENTKIT_GUI_MODE"] = "1" + + # Browser always opens localhost, server binds to configured host + browser_url = f"http://localhost:{port}" + rprint(f"[green]Starting AgentKit GUI — open {browser_url} in your browser[/green]") + + if not no_open: + import threading + def _open_browser(): + import time + time.sleep(2.0) + webbrowser.open(browser_url) + threading.Thread(target=_open_browser, daemon=True).start() + + # Create app directly (not factory mode) so server_config with resolved API keys + # is passed through without relying on env var inheritance in multiprocessing. + from agentkit.server.app import create_app + app = create_app(server_config=server_config) + + uvicorn.run( + app, # Direct app instance, not factory string + host=host, + port=port, + ) + + @app.command() def serve( host: str = typer.Option("0.0.0.0", "--host", help="Server host"), @@ -72,6 +154,21 @@ def serve( # Re-load config after .env is loaded (env vars now available) server_config = ServerConfig.from_yaml(config_path) + # Check if LLM API key is configured + if not server_config.has_llm_provider(): + rprint("[yellow]No LLM API key configured.[/yellow]") + from rich.prompt import Confirm + if Confirm.ask("Would you like to run the setup wizard?", default=True): + config_path = run_onboarding(config_arg=config) + if config_path is None: + rprint("[red]Setup cancelled. Server may not function correctly without API key.[/red]") + else: + server_config = ServerConfig.from_yaml(config_path) + server_config.load_dotenv(str(dotenv)) + server_config = ServerConfig.from_yaml(config_path) + else: + rprint("[dim]Continuing without LLM provider — API calls will fail.[/dim]") + # CLI args override config file for task_store if task_store_backend is not None: server_config.task_store["backend"] = task_store_backend diff --git a/src/agentkit/llm/providers/openai.py b/src/agentkit/llm/providers/openai.py index cd7abbb..a0942b7 100644 --- a/src/agentkit/llm/providers/openai.py +++ b/src/agentkit/llm/providers/openai.py @@ -93,6 +93,8 @@ class OpenAICompatibleProvider(LLMProvider): payload["tools"] = request.tools payload["tool_choice"] = request.tool_choice + logger.debug(f"Chat request to {url}: model={request.model}, messages={len(request.messages)}, tools={len(request.tools or [])}") + start = time.monotonic() try: @@ -108,6 +110,7 @@ class OpenAICompatibleProvider(LLMProvider): error_msg = error_body.get("error", {}).get("message", "Request failed") except Exception: error_msg = f"HTTP {resp.status_code}" + logger.error(f"Chat request failed: HTTP {resp.status_code}, error: {error_msg}") # 不在错误消息中暴露完整响应体，防止 API Key 泄露 raise LLMProviderError("openai", f"HTTP {resp.status_code}: {error_msg}") @@ -177,19 +180,27 @@ class OpenAICompatibleProvider(LLMProvider): "temperature": request.temperature, "max_tokens": request.max_tokens, "stream": True, - "stream_options": {"include_usage": True}, } if request.tools: payload["tools"] = request.tools payload["tool_choice"] = request.tool_choice + logger.debug(f"Stream request to {url}: model={request.model}, messages={len(request.messages)}, tools={len(request.tools or [])}") + response_ctx = self._client.stream("POST", url, json=payload, headers=headers) response = await response_ctx.__aenter__() if response.status_code != 200: await response.aread() await response_ctx.__aexit__(None, None, None) - raise LLMProviderError("openai", f"HTTP {response.status_code}") + # Parse error body for detailed message + try: + error_body = response.json() + error_msg = error_body.get("error", {}).get("message", f"HTTP {response.status_code}") + except Exception: + error_msg = f"HTTP {response.status_code}" + logger.error(f"Stream request failed: HTTP {response.status_code}, error: {error_msg}") + raise LLMProviderError("openai", f"HTTP {response.status_code}: {error_msg}") return _StreamContext(response_ctx, response) diff --git a/src/agentkit/orchestrator/__init__.py b/src/agentkit/orchestrator/__init__.py index 3658902..b0faf35 100644 --- a/src/agentkit/orchestrator/__init__.py +++ b/src/agentkit/orchestrator/__init__.py @@ -1,6 +1,12 @@ """AgentKit Orchestrator - 多 Agent 协同编排""" -from agentkit.orchestrator.pipeline_schema import Pipeline, PipelineStage, StageStatus +from agentkit.orchestrator.pipeline_schema import ( + Pipeline, + PipelineStage, + StageStatus, + AdaptiveConfig, + ReflectionReport, +) from agentkit.orchestrator.pipeline_engine import PipelineEngine from agentkit.orchestrator.pipeline_loader import PipelineLoader from agentkit.orchestrator.handoff import HandoffManager @@ -17,11 +23,14 @@ from agentkit.orchestrator.compensation import ( CompensationResult, SagaOrchestrator, ) +from agentkit.orchestrator.reflection import PipelineReflector, PipelineReplanner __all__ = [ "Pipeline", "PipelineStage", "StageStatus", + "AdaptiveConfig", + "ReflectionReport", "PipelineEngine", "PipelineLoader", "HandoffManager", @@ -35,4 +44,6 @@ __all__ = [ "CompletedStep", "CompensationResult", "SagaOrchestrator", + "PipelineReflector", + "PipelineReplanner", ] diff --git a/src/agentkit/orchestrator/pipeline_engine.py b/src/agentkit/orchestrator/pipeline_engine.py index 3262fe9..ed50d25 100644 --- a/src/agentkit/orchestrator/pipeline_engine.py +++ b/src/agentkit/orchestrator/pipeline_engine.py @@ -8,12 +8,15 @@ from typing import Any from agentkit.orchestrator.compensation import SagaOrchestrator from agentkit.orchestrator.pipeline_schema import ( + AdaptiveConfig, Pipeline, PipelineResult, PipelineStage, + ReflectionReport, StageResult, StageStatus, ) +from agentkit.orchestrator.reflection import PipelineReflector, PipelineReplanner from agentkit.orchestrator.retry import StepRetryPolicy, execute_with_retry logger = logging.getLogger(__name__) @@ -32,16 +35,90 @@ class PipelineEngine: - 状态持久化（可选） """ - def __init__(self, dispatcher: Any = None, state_manager: Any = None): + def __init__(self, dispatcher: Any = None, state_manager: Any = None, llm_gateway: Any = None): self._dispatcher = dispatcher self._state_manager = state_manager + self._llm_gateway = llm_gateway async def execute( self, pipeline: Pipeline, context: dict[str, Any] | None = None, + adaptive_config: AdaptiveConfig | None = None, ) -> PipelineResult: - """执行 Pipeline""" + """执行 Pipeline + + Args: + pipeline: Pipeline 定义 + context: 运行时上下文变量 + adaptive_config: 自适应配置，启用反思-重规划闭环 + """ + # First execution + result = await self._execute_pipeline(pipeline, context) + + # If failed and adaptive is enabled, enter reflection-replanning loop + if result.status == StageStatus.FAILED and adaptive_config and adaptive_config.enabled: + result = await self._adaptive_loop(pipeline, context, result, adaptive_config) + + return result + + async def _adaptive_loop( + self, + pipeline: Pipeline, + context: dict[str, Any] | None, + failed_result: PipelineResult, + adaptive_config: AdaptiveConfig, + ) -> PipelineResult: + """反思-重规划闭环：分析失败原因 → 修正 Pipeline → 重新执行。""" + reflector = PipelineReflector(llm_gateway=self._llm_gateway) + replanner = PipelineReplanner(llm_gateway=self._llm_gateway) + + current_pipeline = pipeline + current_result = failed_result + reflections: list[ReflectionReport] = [] + + for reflection_num in range(1, adaptive_config.max_reflections + 1): + # Reflect + report = await reflector.reflect(current_pipeline, current_result, reflection_num) + reflections.append(report) + logger.info( + f"Pipeline reflection #{reflection_num}: " + f"failure_type={report.failure_type}, " + f"root_cause={report.root_cause}" + ) + + # Replan + new_pipeline = await replanner.replan(current_pipeline, current_result, report) + logger.info(f"Pipeline replanned: {new_pipeline.name} ({len(new_pipeline.stages)} stages)") + + # Re-execute + current_result = await self._execute_pipeline(new_pipeline, context) + current_pipeline = new_pipeline + + # Record reflection in metadata + current_result.metadata["reflections"] = [ + r.model_dump() for r in reflections + ] + + if current_result.status == StageStatus.COMPLETED: + logger.info(f"Pipeline succeeded after {reflection_num} reflection(s)") + return current_result + + # Exhausted reflections + logger.warning( + f"Pipeline failed after {adaptive_config.max_reflections} reflection(s)" + ) + current_result.metadata["reflections"] = [ + r.model_dump() for r in reflections + ] + return current_result + + async def _execute_pipeline( + self, + pipeline: Pipeline, + context: dict[str, Any] | None = None, + ) -> PipelineResult: + """执行 Pipeline 的核心逻辑（不含反思-重规划）。""" result = PipelineResult(pipeline_name=pipeline.name) result.variables = {**pipeline.variables, **(context or {})} diff --git a/src/agentkit/orchestrator/pipeline_schema.py b/src/agentkit/orchestrator/pipeline_schema.py index b385726..540af01 100644 --- a/src/agentkit/orchestrator/pipeline_schema.py +++ b/src/agentkit/orchestrator/pipeline_schema.py @@ -56,3 +56,25 @@ class PipelineResult(BaseModel): stage_results: dict[str, StageResult] = {} variables: dict[str, Any] = {} error_message: str | None = None + metadata: dict[str, Any] = {} + + +class AdaptiveConfig(BaseModel): + """Configuration for adaptive pipeline execution with reflection-replanning.""" + + enabled: bool = False + max_reflections: int = 3 + reflection_model: str = "default" + skip_stages: list[str] = [] + + model_config = {"arbitrary_types_allowed": True} + + +class ReflectionReport(BaseModel): + """Structured report from pipeline reflection analysis.""" + + failure_type: str # input_error, resource_error, logic_error, timeout + root_cause: str + suggested_fix: str + failed_stage: str + reflection_number: int = 1 diff --git a/src/agentkit/orchestrator/reflection.py b/src/agentkit/orchestrator/reflection.py new file mode 100644 index 0000000..18aabc9 --- /dev/null +++ b/src/agentkit/orchestrator/reflection.py @@ -0,0 +1,370 @@ +"""Pipeline 反思-重规划模块 + +当 Pipeline 执行失败时，通过 LLM 反思分析失败原因， +生成修正后的 Pipeline 重新执行。 +""" + +import json +import logging +from typing import Any + +from agentkit.orchestrator.pipeline_schema import ( + Pipeline, + PipelineResult, + PipelineStage, + ReflectionReport, + StageResult, + StageStatus, +) + +logger = logging.getLogger(__name__) + + +class PipelineReflector: + """分析 Pipeline 执行失败原因，生成结构化反思报告。 + + 使用 LLM 分析失败上下文（哪步失败、错误信息、已完成步骤输出）， + 输出 ReflectionReport 包含 failure_type、root_cause 和 suggested_fix。 + """ + + def __init__(self, llm_gateway: Any = None): + self._llm_gateway = llm_gateway + + async def reflect( + self, + pipeline: Pipeline, + result: PipelineResult, + reflection_number: int = 1, + ) -> ReflectionReport: + """分析失败原因并生成反思报告。 + + Args: + pipeline: 原始 Pipeline 定义 + result: 执行失败的 PipelineResult + reflection_number: 当前是第几次反思 + + Returns: + ReflectionReport 结构化反思报告 + """ + # 收集失败上下文 + failed_stage, error_message = self._find_failure(result) + completed_outputs = self._collect_completed_outputs(result) + + # 如果有 LLM Gateway，使用 LLM 分析 + if self._llm_gateway is not None: + try: + return await self._llm_reflect( + pipeline, failed_stage, error_message, + completed_outputs, reflection_number, + ) + except Exception as e: + logger.warning(f"LLM reflection failed, falling back to rule-based: {e}") + + # 规则兜底：基于错误信息分类 + return self._rule_based_reflect( + failed_stage, error_message, reflection_number, + ) + + def _find_failure( + self, result: PipelineResult, + ) -> tuple[str, str]: + """找到第一个失败的 stage 及其错误信息。""" + for name, sr in result.stage_results.items(): + if sr.status == StageStatus.FAILED: + return name, sr.error_message or "unknown error" + return "", "no failed stage found" + + def _collect_completed_outputs( + self, result: PipelineResult, + ) -> dict[str, Any]: + """收集已完成步骤的输出。""" + outputs = {} + for name, sr in result.stage_results.items(): + if sr.status == StageStatus.COMPLETED and sr.output_data: + outputs[name] = sr.output_data + return outputs + + async def _llm_reflect( + self, + pipeline: Pipeline, + failed_stage: str, + error_message: str, + completed_outputs: dict[str, Any], + reflection_number: int, + ) -> ReflectionReport: + """使用 LLM 分析失败原因。""" + prompt = self._build_reflection_prompt( + pipeline, failed_stage, error_message, + completed_outputs, reflection_number, + ) + + response = await self._llm_gateway.chat( + messages=[{"role": "user", "content": prompt}], + model="default", + ) + + # 解析 LLM 返回的 JSON + content = response.content if hasattr(response, "content") else str(response) + return self._parse_reflection_response( + content, failed_stage, reflection_number, + ) + + def _build_reflection_prompt( + self, + pipeline: Pipeline, + failed_stage: str, + error_message: str, + completed_outputs: dict[str, Any], + reflection_number: int, + ) -> str: + """构建反思提示词。""" + stage_descriptions = [] + for s in pipeline.stages: + stage_descriptions.append( + f" - {s.name}: agent={s.agent}, action={s.action}, " + f"depends_on={s.depends_on}" + ) + + completed_summary = json.dumps( + {k: str(v)[:200] for k, v in completed_outputs.items()}, + ensure_ascii=False, + ) + + return f"""Analyze the following pipeline execution failure and provide a structured reflection report. + +Pipeline: {pipeline.name} +Stages: +{chr(10).join(stage_descriptions)} + +Failed stage: {failed_stage} +Error message: {error_message} +Completed outputs (summary): {completed_summary} +Reflection attempt: {reflection_number} + +Respond in JSON format with these fields: +- failure_type: one of "input_error", "resource_error", "logic_error", "timeout" +- root_cause: brief description of the root cause +- suggested_fix: concrete fix to apply to the pipeline + +JSON response:""" + + def _parse_reflection_response( + self, + content: str, + failed_stage: str, + reflection_number: int, + ) -> ReflectionReport: + """解析 LLM 返回的反思报告。""" + # 尝试提取 JSON + try: + # 处理 markdown 代码块包裹的 JSON + text = content.strip() + if text.startswith("```"): + lines = text.split("\n") + text = "\n".join(lines[1:-1]) + + data = json.loads(text) + return ReflectionReport( + failure_type=data.get("failure_type", "logic_error"), + root_cause=data.get("root_cause", "LLM analysis unavailable"), + suggested_fix=data.get("suggested_fix", ""), + failed_stage=failed_stage, + reflection_number=reflection_number, + ) + except (json.JSONDecodeError, KeyError) as e: + logger.warning(f"Failed to parse LLM reflection response: {e}") + return self._rule_based_reflect( + failed_stage, content, reflection_number, + ) + + def _rule_based_reflect( + self, + failed_stage: str, + error_message: str, + reflection_number: int, + ) -> ReflectionReport: + """基于规则的兜底反思。""" + error_lower = error_message.lower() + + if "timeout" in error_lower or "timed out" in error_lower: + failure_type = "timeout" + root_cause = f"Stage '{failed_stage}' timed out" + suggested_fix = "Increase timeout_seconds and add retry_policy" + elif "not found" in error_lower or "404" in error_lower: + failure_type = "resource_error" + root_cause = f"Required resource not found in stage '{failed_stage}'" + suggested_fix = "Add pre-check step or adjust resource reference" + elif "invalid" in error_lower or "validation" in error_lower: + failure_type = "input_error" + root_cause = f"Invalid input to stage '{failed_stage}'" + suggested_fix = "Add input validation step before this stage" + else: + failure_type = "logic_error" + root_cause = f"Stage '{failed_stage}' failed: {error_message[:200]}" + suggested_fix = "Review stage logic and adjust action or inputs" + + return ReflectionReport( + failure_type=failure_type, + root_cause=root_cause, + suggested_fix=suggested_fix, + failed_stage=failed_stage, + reflection_number=reflection_number, + ) + + +class PipelineReplanner: + """基于反思报告生成修正后的 Pipeline。 + + 保留已完成步骤的结果，仅重新规划失败及后续步骤。 + """ + + def __init__(self, llm_gateway: Any = None): + self._llm_gateway = llm_gateway + + async def replan( + self, + pipeline: Pipeline, + result: PipelineResult, + report: ReflectionReport, + ) -> Pipeline: + """基于反思报告重新规划 Pipeline。 + + Args: + pipeline: 原始 Pipeline + result: 执行失败的 PipelineResult + report: 反思报告 + + Returns: + 修正后的 Pipeline + """ + # 如果有 LLM Gateway，使用 LLM 重规划 + if self._llm_gateway is not None: + try: + return await self._llm_replan(pipeline, result, report) + except Exception as e: + logger.warning(f"LLM replanning failed, falling back to rule-based: {e}") + + # 规则兜底：基于 failure_type 调整 + return self._rule_based_replan(pipeline, result, report) + + async def _llm_replan( + self, + pipeline: Pipeline, + result: PipelineResult, + report: ReflectionReport, + ) -> Pipeline: + """使用 LLM 生成修正后的 Pipeline。""" + completed_stages = [ + name for name, sr in result.stage_results.items() + if sr.status == StageStatus.COMPLETED + ] + + prompt = f"""Based on the reflection report, generate a corrected pipeline. + +Original pipeline: {pipeline.name} +Stages: {[s.name for s in pipeline.stages]} +Completed stages: {completed_stages} +Failed stage: {report.failed_stage} +Failure type: {report.failure_type} +Root cause: {report.root_cause} +Suggested fix: {report.suggested_fix} + +Generate a corrected pipeline in JSON format with the same structure as the original. +Only modify stages that need changes based on the reflection. +Keep completed stages unchanged. + +JSON pipeline:""" + + response = await self._llm_gateway.chat( + messages=[{"role": "user", "content": prompt}], + model="default", + ) + + content = response.content if hasattr(response, "content") else str(response) + return self._parse_pipeline_response(content, pipeline) + + def _parse_pipeline_response( + self, content: str, original: Pipeline, + ) -> Pipeline: + """解析 LLM 返回的 Pipeline JSON。""" + try: + text = content.strip() + if text.startswith("```"): + lines = text.split("\n") + text = "\n".join(lines[1:-1]) + + data = json.loads(text) + stages = [ + PipelineStage(**s) for s in data.get("stages", []) + ] + return Pipeline( + name=data.get("name", original.name), + version=data.get("version", original.version), + description=data.get("description", original.description), + stages=stages, + variables=data.get("variables", original.variables), + ) + except (json.JSONDecodeError, Exception) as e: + logger.warning(f"Failed to parse LLM replan response: {e}") + return original + + def _rule_based_replan( + self, + pipeline: Pipeline, + result: PipelineResult, + report: ReflectionReport, + ) -> Pipeline: + """基于规则的兜底重规划。""" + completed_stages = { + name for name, sr in result.stage_results.items() + if sr.status == StageStatus.COMPLETED + } + + # 构建修正后的 stages 列表 + new_stages: list[PipelineStage] = [] + + for stage in pipeline.stages: + if stage.name in completed_stages: + # 已完成的步骤保持不变，但标记为 continue_on_failure + # 因为它们的结果已经存在 + new_stages.append(stage) + elif stage.name == report.failed_stage: + # 失败步骤：根据 failure_type 调整 + modified = self._adjust_failed_stage(stage, report) + new_stages.append(modified) + else: + # 后续步骤保持不变 + new_stages.append(stage) + + return Pipeline( + name=f"{pipeline.name}_replanned", + version=pipeline.version, + description=f"Replanned after reflection: {report.root_cause}", + stages=new_stages, + variables=pipeline.variables, + ) + + def _adjust_failed_stage( + self, stage: PipelineStage, report: ReflectionReport, + ) -> PipelineStage: + """根据反思报告调整失败的步骤。""" + adjustments: dict[str, Any] = {} + + if report.failure_type == "timeout": + adjustments["timeout_seconds"] = min( + stage.timeout_seconds * 2, 3600, + ) + if stage.retry_policy is None: + from agentkit.orchestrator.retry import StepRetryPolicy + adjustments["retry_policy"] = StepRetryPolicy(max_attempts=2) + + elif report.failure_type == "resource_error": + adjustments["continue_on_failure"] = True + + elif report.failure_type == "input_error": + # 添加重试策略，可能输入在后续可用 + if stage.retry_policy is None: + from agentkit.orchestrator.retry import StepRetryPolicy + adjustments["retry_policy"] = StepRetryPolicy(max_attempts=2) + + return stage.model_copy(update=adjustments) diff --git a/src/agentkit/server/app.py b/src/agentkit/server/app.py index 1874925..2e815fa 100644 --- a/src/agentkit/server/app.py +++ b/src/agentkit/server/app.py @@ -96,6 +96,106 @@ async def lifespan(app: FastAPI): if mcp_manager is not None: await mcp_manager.start_all() + # In GUI mode, ensure a default chat agent exists with memory + tools + gui_mode = os.environ.get("AGENTKIT_GUI_MODE") + if gui_mode and not app.state.agent_pool.list_agents(): + from agentkit.core.config_driven import AgentConfig + from agentkit.memory.profile import MemoryStore + from agentkit.tools.memory_tool import MemoryTool + from agentkit.tools.shell import ShellTool + from agentkit.tools.web_search import WebSearchTool + from agentkit.tools.web_crawl import WebCrawlTool + from agentkit.tools.baidu_search import BaiduSearchTool + + # Initialize memory store and build system prompt + memory_store = MemoryStore() + memory_store.ensure_defaults() + memory_snapshot = memory_store.load_all() + base_prompt = ( + "你是一个有帮助的AI助手。请记住我们对话的上下文，并在后续对话中引用之前的内容。回答要清晰简洁，请使用中文回复。\n\n" + "重要提示：当你不确定事实信息、时事新闻或任何你不确信的话题时，" + "你必须先使用搜索工具查找准确和最新的信息，然后再回答。" + "中文内容优先使用 baidu_search 工具，英文/国际内容使用 web_search。" + "在能够搜索到真相的情况下，绝不猜测或编造答案。" + "始终优先搜索而不是给出可能不正确的信息。" + ) + effective_system_prompt = memory_store.build_system_prompt(memory_snapshot, base_prompt) + + # Store memory_store on app.state for chat routes to use + app.state.memory_store = memory_store + + default_config = AgentConfig( + name="default", + agent_type="chat", + task_mode="llm_generate", + description="Default chat agent for GUI", + prompt={"system": effective_system_prompt}, + ) + try: + agent = await app.state.agent_pool.create_agent(default_config) + + # Register tools into the agent's tool registry + search_api_keys = { + "tavily_api_key": os.environ.get("TAVILY_API_KEY"), + "serper_api_key": os.environ.get("SERPER_API_KEY"), + } + agent._tool_registry.register(MemoryTool(memory_store=memory_store)) + agent._tool_registry.register(ShellTool(working_dir=os.getcwd())) + agent._tool_registry.register(BaiduSearchTool()) + agent._tool_registry.register(WebSearchTool(**search_api_keys)) + agent._tool_registry.register(WebCrawlTool()) + + # Override system prompt with memory-injected version + agent._system_prompt = effective_system_prompt + + logger.info("GUI mode: created default chat agent with memory + tools") + except Exception as e: + logger.warning(f"GUI mode: failed to create default agent: {e}") + + # Load skills from config and register into SkillRegistry + try: + from agentkit.skills.loader import SkillLoader + skill_registry = app.state.skill_registry + tool_registry = app.state.tool_registry + + # Register GUI tools into the shared tool registry so skills can bind them + for tool in agent._tool_registry.list_tools(): + try: + tool_registry.register(tool) + except Exception: + pass # Already registered + + # Load skills from configured paths + server_config = getattr(app.state, "server_config", None) + if server_config and server_config.skill_paths: + loader = SkillLoader( + skill_registry=skill_registry, + tool_registry=tool_registry, + ) + for skill_path in server_config.skill_paths: + from pathlib import Path as _P + p = _P(skill_path) + if p.is_dir(): + loaded = loader.load_from_directory(str(p)) + logger.info(f"GUI mode: loaded {len(loaded)} skills from {p}") + elif p.is_file() and p.suffix in (".yaml", ".yml"): + try: + loader.load_from_file(str(p)) + logger.info(f"GUI mode: loaded skill from {p}") + except Exception as se: + logger.warning(f"GUI mode: failed to load skill from {p}: {se}") + + logger.info(f"GUI mode: {len(skill_registry.list_skills())} skills registered") + except Exception as e: + logger.warning(f"GUI mode: failed to load skills: {e}") + elif gui_mode: + # Agent already exists (e.g. from config), still ensure memory store is available + if not hasattr(app.state, "memory_store") or app.state.memory_store is None: + from agentkit.memory.profile import MemoryStore + memory_store = MemoryStore() + memory_store.ensure_defaults() + app.state.memory_store = memory_store + yield # Shutdown @@ -151,6 +251,24 @@ def _on_config_change(app: FastAPI, config: ServerConfig) -> None: # Reload skills if skill paths changed try: new_skill_registry = _build_skill_registry(config) + # Re-bind tools from the shared tool_registry so skills don't lose their bindings + tool_registry = getattr(app.state, "tool_registry", None) + if tool_registry: + from agentkit.skills.loader import SkillLoader + loader = SkillLoader( + skill_registry=new_skill_registry, + tool_registry=tool_registry, + ) + for skill_path in (config.skill_paths or []): + from pathlib import Path as _P + p = _P(skill_path) + if p.is_dir(): + loader.load_from_directory(str(p)) + elif p.is_file() and p.suffix in (".yaml", ".yml"): + try: + loader.load_from_file(str(p)) + except Exception: + pass app.state.skill_registry = new_skill_registry if hasattr(app.state, "agent_pool") and app.state.agent_pool is not None: app.state.agent_pool._skill_registry = new_skill_registry @@ -191,6 +309,20 @@ def create_app( if server_config is None: config_path = os.environ.get("AGENTKIT_CONFIG_PATH") if config_path and os.path.exists(config_path): + # Load .env before parsing config (so ${ENV_VAR} substitutions work) + from pathlib import Path as _P + _dotenv = _P(config_path).parent / ".env" + if _dotenv.exists(): + with open(_dotenv, encoding="utf-8") as _f: + for _line in _f: + _line = _line.strip() + if not _line or _line.startswith("#") or "=" not in _line: + continue + _key, _, _val = _line.partition("=") + _key = _key.strip() + _val = _val.strip().strip("\"'") + if _key and _key not in os.environ: + os.environ[_key] = _val server_config = ServerConfig.from_yaml(config_path) app = FastAPI(title="AgentKit Server", version="2.0.0", lifespan=lifespan) @@ -319,8 +451,10 @@ def create_app( session_config = {} if server_config and hasattr(server_config, "session") and server_config.session: session_config = server_config.session + # GUI mode defaults to file-backed sessions for persistence + session_backend = session_config.get("backend", "file" if os.environ.get("AGENTKIT_GUI_MODE") else "memory") session_store = create_session_store( - backend=session_config.get("backend", "memory"), + backend=session_backend, redis_url=session_config.get("redis_url", "redis://localhost:6379/0"), ttl_seconds=session_config.get("ttl_seconds", 86400), ) @@ -453,4 +587,20 @@ def create_app( app.include_router(memory.router, prefix="/api/v1") app.include_router(chat.router, prefix="/api/v1") + # Serve GUI when in GUI mode + gui_mode = os.environ.get("AGENTKIT_GUI_MODE") + if gui_mode: + from pathlib import Path as _Path + from fastapi.responses import HTMLResponse, FileResponse + + _static_dir = _Path(__file__).parent / "static" + + @app.get("/", response_class=HTMLResponse, include_in_schema=False) + async def gui_index(): + """Serve the GUI index page.""" + index_path = _static_dir / "index.html" + if index_path.exists(): + return FileResponse(str(index_path)) + return HTMLResponse("

AgentKit GUI not found

", status_code=404) + return app diff --git a/src/agentkit/server/config.py b/src/agentkit/server/config.py index 5d66a7d..1e1af91 100644 --- a/src/agentkit/server/config.py +++ b/src/agentkit/server/config.py @@ -135,6 +135,13 @@ class ServerConfig: self._watcher_task: asyncio.Task | None = None self._last_mtime: float = 0.0 + def has_llm_provider(self) -> bool: + """检查是否配置了有效的 LLM Provider（API Key 非空）""" + for name, provider in self.llm_config.providers.items(): + if provider.api_key: + return True + return False + @classmethod def from_yaml(cls, path: str) -> "ServerConfig": """Load configuration from a YAML file.""" diff --git a/src/agentkit/server/routes/chat.py b/src/agentkit/server/routes/chat.py index e7a1ba1..e8ff178 100644 --- a/src/agentkit/server/routes/chat.py +++ b/src/agentkit/server/routes/chat.py @@ -125,6 +125,14 @@ def _message_to_response(msg) -> MessageResponse: # ── REST endpoints ──────────────────────────────────────────────────── +@router.get("/sessions", response_model=list[SessionResponse]) +async def list_sessions(req: Request): + """List all chat sessions.""" + sm = _get_session_manager(req) + sessions = await sm.list_sessions() + return [_session_to_response(s) for s in sessions] + + @router.post("/sessions", response_model=SessionResponse) async def create_session(request: CreateSessionRequest, req: Request): """Create a new chat session bound to an Agent.""" @@ -147,7 +155,7 @@ async def get_session(session_id: str, req: Request): @router.get("/sessions/{session_id}/messages", response_model=list[MessageResponse]) -async def get_messages(session_id: str, limit: int | None = None, offset: int = 0, req: Request = None): +async def get_messages(session_id: str, req: Request, limit: int | None = None, offset: int = 0): """Get conversation history for a session.""" sm = _get_session_manager(req) session = await sm.get_session(session_id) @@ -186,13 +194,14 @@ async def send_message(session_id: str, request: SendMessageRequest, req: Reques # Execute the Agent try: react_engine = ReActEngine(llm_gateway=req.app.state.llm_gateway) - tools = list(agent._tool_registry._tools.values()) if agent._tool_registry else [] + tools = agent._tool_registry.list_tools() if agent._tool_registry else [] + system_prompt = getattr(agent, "_system_prompt", None) or (agent.get_system_prompt() if hasattr(agent, "get_system_prompt") else None) result = await react_engine.execute( messages=chat_messages, tools=tools, - model=agent._llm_model if hasattr(agent, "_llm_model") else "default", + model=agent.get_model() if hasattr(agent, "get_model") else getattr(agent, "_llm_model", "default"), agent_name=agent.name, - system_prompt=agent._system_prompt if hasattr(agent, "_system_prompt") else None, + system_prompt=system_prompt, ) # Append assistant reply @@ -296,8 +305,10 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None: if msg_type == "message": content = msg.get("content", "") + # Create a fresh CancellationToken for each message + message_token = CancellationToken() await _handle_chat_message( - websocket, session_id, content, sm, cancellation_token, pending_replies + websocket, session_id, content, sm, message_token, pending_replies ) elif msg_type == "reply": @@ -338,14 +349,15 @@ async def _handle_chat_message( cancellation_token: CancellationToken, pending_replies: dict[str, asyncio.Future], ) -> None: - """Handle a user message: append to session, execute Agent, stream events.""" - # Append user message - await sm.append_message(session_id=session_id, role=MessageRole.USER, content=content) + """Handle a user message: append to session, execute Agent, stream events. - # Get full conversation history - chat_messages = await sm.get_chat_messages(session_id) + When skills are registered, attempts to route the user's message to a + matching skill via IntentRouter. If a skill is matched, the skill's + prompt, tools, and execution_mode are used instead of the default agent's. + """ + from agentkit.chat.skill_routing import resolve_skill_routing - # Resolve Agent + # Resolve Agent first (needed for default tools/prompt) pool = websocket.app.state.agent_pool session = await sm.get_session(session_id) if session is None: @@ -357,18 +369,57 @@ async def _handle_chat_message( await websocket.send_json({"type": "error", "data": {"message": f"Agent '{session.agent_name}' not found"}}) return + # Default execution parameters from agent + default_tools = agent._tool_registry.list_tools() if agent._tool_registry else [] + default_system_prompt = getattr(agent, "_system_prompt", None) or (agent.get_system_prompt() if hasattr(agent, "get_system_prompt") else None) + default_model = agent.get_model() if hasattr(agent, "get_model") else getattr(agent, "_llm_model", "default") + + # Resolve skill routing using shared module + skill_registry = getattr(websocket.app.state, "skill_registry", None) + intent_router = getattr(websocket.app.state, "intent_router", None) + + routing = await resolve_skill_routing( + content=content, + skill_registry=skill_registry, + intent_router=intent_router, + default_tools=default_tools, + default_system_prompt=default_system_prompt, + default_model=default_model, + default_agent_name=agent.name, + agent_tool_registry=agent._tool_registry if agent._tool_registry else None, + session_id=session_id, + ) + + # Notify frontend about skill match + if routing.matched: + await websocket.send_json({ + "type": "skill_match", + "data": { + "skill": routing.skill_name, + "method": routing.match_method, + "confidence": routing.match_confidence, + }, + }) + + # Append user message (use clean_content if @skill: prefix was stripped) + await sm.append_message(session_id=session_id, role=MessageRole.USER, content=routing.clean_content) + + # Get full conversation history + chat_messages = await sm.get_chat_messages(session_id) + # Execute Agent with streaming react_engine = ReActEngine(llm_gateway=websocket.app.state.llm_gateway) - tools = list(agent._tool_registry._tools.values()) if agent._tool_registry else [] + + logger.info(f"Chat session {session_id}: executing with {len(routing.tools)} tools, model={routing.model}, skill={routing.skill_name}") try: final_content = "" async for event in react_engine.execute_stream( messages=chat_messages, - tools=tools, - model=agent._llm_model if hasattr(agent, "_llm_model") else "default", - agent_name=agent.name, - system_prompt=agent._system_prompt if hasattr(agent, "_system_prompt") else None, + tools=routing.tools, + model=routing.model, + agent_name=routing.agent_name, + system_prompt=routing.system_prompt, cancellation_token=cancellation_token, ): if event.event_type == "final_answer": @@ -402,4 +453,10 @@ async def _handle_chat_message( ) except Exception as e: - await websocket.send_json({"type": "error", "data": {"message": str(e)}}) + logger.error(f"Chat execution error for session {session_id}: {e}") + # Show meaningful error to user, but avoid leaking full stack traces + error_msg = str(e) + # Truncate very long error messages + if len(error_msg) > 200: + error_msg = error_msg[:200] + "..." + await websocket.send_json({"type": "error", "data": {"message": error_msg}}) diff --git a/src/agentkit/server/routes/skills.py b/src/agentkit/server/routes/skills.py index b10afa7..77ed2a9 100644 --- a/src/agentkit/server/routes/skills.py +++ b/src/agentkit/server/routes/skills.py @@ -1,7 +1,11 @@ """Skill registration routes""" import logging +import os +import re +import urllib.parse +import httpx from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel from typing import Any @@ -13,6 +17,87 @@ logger = logging.getLogger(__name__) router = APIRouter(tags=["skills"]) +# Strict skill name validation: lowercase alphanumeric, hyphens, underscores +_SKILL_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") + +# Allowed domains for source URL downloads (SSRF mitigation) +_ALLOWED_DOWNLOAD_DOMAINS = { + "raw.githubusercontent.com", + "github.com", + "gist.githubusercontent.com", +} + + +def _validate_skill_name(name: str) -> str: + """Validate and normalize a skill name. Raises HTTPException on invalid input.""" + normalized = name.strip().lower() + if not _SKILL_NAME_RE.match(normalized): + raise HTTPException( + status_code=400, + detail=f"Invalid skill name '{name}': must contain only lowercase letters, digits, hyphens, and underscores (1-64 chars)", + ) + return normalized + + +def _get_skills_dir(req: Request) -> str: + """Get the skills directory from server_config, falling back to configs/skills/.""" + server_config = getattr(req.app.state, "server_config", None) + if server_config and server_config.skill_paths: + # Use the first configured skill path as the install target + from pathlib import Path as _P + first_path = _P(server_config.skill_paths[0]) + if first_path.is_dir(): + return str(first_path) + # Fallback: configs/skills/ relative to project root + return os.path.join(os.getcwd(), "configs", "skills") + + +def _validate_source_url(source: str) -> None: + """Validate that a source URL points to an allowed domain (SSRF mitigation).""" + from urllib.parse import urlparse + parsed = urlparse(source) + if parsed.scheme not in ("https", "http"): + raise HTTPException(status_code=400, detail=f"Invalid source URL scheme: only http/https allowed") + # Block private/internal IPs by checking hostname + import ipaddress + import socket + hostname = parsed.hostname + if hostname: + try: + # Resolve hostname to check for private IPs + resolved = socket.getaddrinfo(hostname, None) + for family, type_, proto, canonname, sockaddr in resolved: + ip = ipaddress.ip_address(sockaddr[0]) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + raise HTTPException( + status_code=400, + detail="Source URL points to a private/internal address — not allowed", + ) + except socket.gaierror: + pass # DNS resolution failed, let httpx handle it + # Check domain allowlist for source URLs + if hostname and hostname not in _ALLOWED_DOWNLOAD_DOMAINS: + # Allow but log a warning for non-allowlisted domains + logger.warning(f"Source URL domain '{hostname}' is not in the allowlist: {_ALLOWED_DOWNLOAD_DOMAINS}") + + +def _validate_yaml_content(content: str) -> dict: + """Validate YAML content before writing to disk. Returns parsed dict.""" + import yaml + try: + data = yaml.safe_load(content) + except yaml.YAMLError as e: + raise HTTPException(status_code=400, detail=f"Invalid YAML content: {e}") + + if not isinstance(data, dict): + raise HTTPException(status_code=400, detail="Skill YAML must be a mapping/dict") + + # Require at least a 'name' field + if "name" not in data: + raise HTTPException(status_code=400, detail="Skill YAML must contain a 'name' field") + + return data + class RegisterSkillRequest(BaseModel): config: dict[str, Any] @@ -27,6 +112,11 @@ class ExecutePipelineRequest(BaseModel): input_data: dict[str, Any] +class InstallSkillRequest(BaseModel): + name: str + source: str | None = None # Optional: URL or "github:user/repo/path" + + @router.post("/skills", status_code=201) async def register_skill(request: RegisterSkillRequest, req: Request): """Register a Skill""" @@ -50,7 +140,7 @@ async def register_skill(request: RegisterSkillRequest, req: Request): @router.get("/skills") async def list_skills(req: Request): - """List all skills""" + """List all skills with full metadata""" skill_registry = req.app.state.skill_registry skills = skill_registry.list_skills() return [ @@ -58,12 +148,182 @@ async def list_skills(req: Request): "name": s.name, "agent_type": s.config.agent_type, "version": s.config.version, - "description": s.config.description, + "description": s.config.description or "", + "task_mode": s.config.task_mode or "", + "intent_keywords": s.config.intent.keywords if s.config.intent else [], + "intent_description": s.config.intent.description if s.config.intent else "", + "tools": s.config.tools or [], + "bound_tools": [t.name for t in (s.tools or [])], + "prompt_identity": (s.config.prompt or {}).get("identity", ""), } for s in skills ] +@router.post("/skills/install") +async def install_skill(request: InstallSkillRequest, req: Request): + """Search for and install a skill by name. + + Searches GitHub for agentkit-skill YAML files matching the name, + downloads the first match, saves it to configs/skills/, and registers it. + """ + skill_name = _validate_skill_name(request.name) + source = request.source + + skill_registry = req.app.state.skill_registry + tool_registry = getattr(req.app.state, "tool_registry", None) + + # If source URL is provided directly, download from it + if source and source.startswith("http"): + _validate_source_url(source) + try: + async with httpx.AsyncClient(timeout=30, follow_redirects=True, max_redirects=3) as client: + resp = await client.get(source) + resp.raise_for_status() + yaml_content = resp.text + except Exception as e: + raise HTTPException(status_code=400, detail=f"Failed to download from source: {e}") + elif source and source.startswith("file://"): + # Read from local file path + local_path = source[7:] # strip "file://" + if not os.path.exists(local_path): + raise HTTPException(status_code=404, detail=f"Local file not found: {local_path}") + # Verify the path is within the skills directory + skills_dir_base = _get_skills_dir(req) + if not os.path.realpath(local_path).startswith(os.path.realpath(skills_dir_base)): + raise HTTPException(status_code=400, detail="Local file path must be within the skills directory") + try: + with open(local_path, encoding="utf-8") as f: + yaml_content = f.read() + except Exception as e: + raise HTTPException(status_code=400, detail=f"Failed to read local file: {e}") + else: + # Search GitHub for skills (YAML config files) + search_query = f"{skill_name} skill config filename:yaml" + encoded_query = urllib.parse.quote(search_query) + github_api = f"https://api.github.com/search/code?q={encoded_query}&per_page=5" + + try: + async with httpx.AsyncClient(timeout=15) as client: + gh_resp = await client.get( + github_api, + headers={ + "Accept": "application/vnd.github.v3+json", + "User-Agent": "agentkit", + }, + ) + gh_data = gh_resp.json() + except Exception as e: + raise HTTPException(status_code=502, detail=f"GitHub search failed: {e}") + + items = gh_data.get("items", []) + if not items: + # Fallback: try a simpler search + search_query2 = f"{skill_name} skill" + encoded_query2 = urllib.parse.quote(search_query2) + github_api2 = f"https://api.github.com/search/code?q={encoded_query2}+extension:yaml&per_page=5" + try: + async with httpx.AsyncClient(timeout=15) as client: + gh_resp2 = await client.get( + github_api2, + headers={"Accept": "application/vnd.github.v3+json", "User-Agent": "agentkit"}, + ) + items = gh_resp2.json().get("items", []) + except Exception: + items = [] + + if not items: + raise HTTPException(status_code=404, detail=f"No skill found matching '{skill_name}'") + + # Download the first matching file + item = items[0] + raw_url = item.get("html_url", "") + if raw_url: + # Validate the URL is from github.com before transforming + if not raw_url.startswith("https://github.com/"): + raise HTTPException(status_code=400, detail="Search result URL is not from github.com") + raw_url = raw_url.replace("github.com", "raw.githubusercontent.com").replace("/blob/", "/") + else: + raise HTTPException(status_code=404, detail="Could not construct download URL") + + try: + async with httpx.AsyncClient(timeout=30, follow_redirects=True, max_redirects=3) as client: + resp = await client.get(raw_url) + resp.raise_for_status() + yaml_content = resp.text + except Exception as e: + raise HTTPException(status_code=400, detail=f"Failed to download skill: {e}") + + # Validate YAML content before writing to disk + _validate_yaml_content(yaml_content) + + # Save to skills directory (config-driven path) + skills_dir = _get_skills_dir(req) + os.makedirs(skills_dir, exist_ok=True) + file_path = os.path.join(skills_dir, f"{skill_name}.yaml") + + # Verify resolved path stays within skills_dir (path traversal protection) + if not os.path.realpath(file_path).startswith(os.path.realpath(skills_dir)): + raise HTTPException(status_code=400, detail="Invalid path: escapes skills directory") + + with open(file_path, "w", encoding="utf-8") as f: + f.write(yaml_content) + + # Load and register the skill + registration_ok = False + try: + from agentkit.skills.loader import SkillLoader + loader = SkillLoader( + skill_registry=skill_registry, + tool_registry=tool_registry, + ) + loader.load_from_file(file_path) + registration_ok = True + except Exception as e: + logger.warning(f"Failed to register installed skill: {e}") + + if not registration_ok: + # Remove the invalid YAML file and report error + try: + os.remove(file_path) + except Exception: + pass + raise HTTPException(status_code=500, detail=f"Skill downloaded but registration failed") + + return { + "status": "installed", + "name": skill_name, + "path": file_path, + } + + +@router.delete("/skills/{name}") +async def uninstall_skill(name: str, req: Request): + """Unregister a skill and optionally remove its YAML file.""" + # Validate name to prevent path traversal + validated_name = _validate_skill_name(name) + + skill_registry = req.app.state.skill_registry + + try: + skill_registry.get(validated_name) + except Exception: + raise HTTPException(status_code=404, detail=f"Skill '{name}' not found") + + # Remove from registry + skill_registry.unregister(validated_name) + + # Remove the YAML file (config-driven path) + skills_dir = _get_skills_dir(req) + yaml_path = os.path.join(skills_dir, f"{validated_name}.yaml") + + # Verify resolved path stays within skills_dir + if os.path.exists(yaml_path) and os.path.realpath(yaml_path).startswith(os.path.realpath(skills_dir)): + os.remove(yaml_path) + + return {"status": "uninstalled", "name": validated_name} + + # ---- Pipeline endpoints ---- diff --git a/src/agentkit/server/routes/ws.py b/src/agentkit/server/routes/ws.py index ece3056..5110b83 100644 --- a/src/agentkit/server/routes/ws.py +++ b/src/agentkit/server/routes/ws.py @@ -185,7 +185,7 @@ async def _run_react_and_stream( async for event in react_engine.execute_stream( messages=messages, tools=tools, - model=agent._llm_model if hasattr(agent, "_llm_model") else "default", + model=agent.get_model() if hasattr(agent, "get_model") else (agent._llm_model if hasattr(agent, "_llm_model") else "default"), agent_name=agent.name, system_prompt=agent._system_prompt if hasattr(agent, "_system_prompt") else None, cancellation_token=cancellation_token, diff --git a/src/agentkit/server/static/index.html b/src/agentkit/server/static/index.html new file mode 100644 index 0000000..d94306d --- /dev/null +++ b/src/agentkit/server/static/index.html @@ -0,0 +1,661 @@ + + + + + +AgentKit + + + + + + + +
+ + + + + +
+
+ + AgentKit + 未连接 + +
+
+
+
🤖
+

欢迎使用 AgentKit

+

开始一段新对话，或从侧边栏选择已有会话。

+
+
+
+
+ + +
+
+
+ + + +
+ + + + diff --git a/src/agentkit/session/store.py b/src/agentkit/session/store.py index b16c7f7..7199370 100644 --- a/src/agentkit/session/store.py +++ b/src/agentkit/session/store.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import logging +import os from typing import Any, Protocol, runtime_checkable from agentkit.session.models import Message, Session, SessionStatus @@ -214,15 +215,127 @@ class RedisSessionStore: from datetime import datetime, timezone # noqa: E402 +class FileSessionStore: + """File-based session store — persists sessions to ~/.agentkit/sessions/. + + Each session is stored as a JSON file containing both session metadata + and messages. Suitable for single-user GUI mode without Redis. + """ + + def __init__(self, data_dir: str | None = None): + if data_dir is None: + data_dir = os.path.expanduser("~/.agentkit/sessions") + self._data_dir = data_dir + os.makedirs(self._data_dir, exist_ok=True) + + def _session_path(self, session_id: str) -> str: + return os.path.join(self._data_dir, f"{session_id}.json") + + def _read_session_file(self, session_id: str) -> dict | None: + path = self._session_path(session_id) + if not os.path.exists(path): + return None + with open(path, encoding="utf-8") as f: + return json.load(f) + + def _write_session_file(self, session_id: str, data: dict) -> None: + path = self._session_path(session_id) + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + async def save_session(self, session: Session) -> None: + data = self._read_session_file(session.session_id) or {"messages": []} + data["session"] = session.to_dict() + data["session"]["updated_at"] = datetime.now(timezone.utc).isoformat() + self._write_session_file(session.session_id, data) + + async def get_session(self, session_id: str) -> Session | None: + data = self._read_session_file(session_id) + if data is None: + return None + return Session.from_dict(data["session"]) + + async def update_session_status(self, session_id: str, status: SessionStatus) -> Session | None: + data = self._read_session_file(session_id) + if data is None: + return None + data["session"]["status"] = status.value + data["session"]["updated_at"] = datetime.now(timezone.utc).isoformat() + self._write_session_file(session_id, data) + return Session.from_dict(data["session"]) + + async def delete_session(self, session_id: str) -> bool: + path = self._session_path(session_id) + if os.path.exists(path): + os.remove(path) + return True + return False + + async def list_sessions(self, agent_name: str | None = None, limit: int = 100) -> list[Session]: + sessions: list[Session] = [] + for fname in os.listdir(self._data_dir): + if not fname.endswith(".json"): + continue + path = os.path.join(self._data_dir, fname) + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + session = Session.from_dict(data["session"]) + if agent_name is None or session.agent_name == agent_name: + sessions.append(session) + except Exception: + continue + sessions.sort(key=lambda s: s.updated_at, reverse=True) + return sessions[:limit] + + async def append_message(self, message: Message) -> None: + data = self._read_session_file(message.session_id) + if data is None: + data = {"session": {"session_id": message.session_id}, "messages": []} + data.setdefault("messages", []).append(message.to_dict()) + # Update session timestamp + if "session" in data: + data["session"]["updated_at"] = datetime.now(timezone.utc).isoformat() + self._write_session_file(message.session_id, data) + + async def get_messages(self, session_id: str, limit: int | None = None, offset: int = 0) -> list[Message]: + data = self._read_session_file(session_id) + if data is None: + return [] + msgs = data.get("messages", [])[offset:] + if limit is not None: + msgs = msgs[:limit] + return [Message.from_dict(m) for m in msgs] + + async def count_messages(self, session_id: str) -> int: + data = self._read_session_file(session_id) + if data is None: + return 0 + return len(data.get("messages", [])) + + async def health_check(self) -> bool: + return os.path.isdir(self._data_dir) + + def create_session_store( backend: str = "memory", redis_url: str = "redis://localhost:6379/0", ttl_seconds: int = 86400, -) -> InMemorySessionStore | RedisSessionStore: - """Factory: create a SessionStore backed by memory or Redis. + data_dir: str | None = None, +) -> InMemorySessionStore | RedisSessionStore | FileSessionStore: + """Factory: create a SessionStore backed by memory, file, or Redis. + + - ``memory``: In-memory (lost on restart) + - ``file``: JSON files in ``~/.agentkit/sessions/`` (persistent, no deps) + - ``redis``: Redis-backed (production, requires Redis) Falls back to InMemorySessionStore if Redis is unavailable. """ + if backend == "file": + store = FileSessionStore(data_dir=data_dir) + logger.info(f"SessionStore backend: file ({store._data_dir})") + return store + if backend == "redis": try: import redis.asyncio as aioredis # noqa: F401 diff --git a/src/agentkit/tools/baidu_search.py b/src/agentkit/tools/baidu_search.py index 1b3efc0..e3f76da 100644 --- a/src/agentkit/tools/baidu_search.py +++ b/src/agentkit/tools/baidu_search.py @@ -158,15 +158,39 @@ class BaiduSearchTool(Tool): "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/120.0.0.0 Safari/537.36" + "Chrome/131.0.0.0 Safari/537.36" ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Cache-Control": "max-age=0", + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Upgrade-Insecure-Requests": "1", }, ) html = resp.text + # Check if we got a captcha page + if "验证" in html and len(html) < 5000: + logger.warning("Baidu returned captcha page, search unavailable") + return { + "error": "Baidu search blocked by captcha", + "results": [], + "total": 0, + "success": False, + } + # 简单解析搜索结果（基于百度搜索结果页 HTML 结构） results = self._parse_baidu_html(html, max_results) + if not results: + # Try alternative parsing + results = self._parse_baidu_html_alt(html, max_results) + return {"results": results, "total": len(results), "success": True} except Exception as e: @@ -188,38 +212,111 @@ class BaiduSearchTool(Tool): results: list[dict[str, str]] = [] - # 匹配百度搜索结果块 - # 百度搜索结果通常在
中 - pattern = re.compile( + # 匹配百度搜索结果块 - multiple patterns for different Baidu page versions + # Pattern 1:

with href + pattern1 = re.compile( r']*class="[^"]*t[^"]*"[^>]*>.*?href="([^"]*)"[^>]*>(.*?)', re.DOTALL, ) - snippet_pattern = re.compile( + # Pattern 2:

with data-url or inside
+ pattern2 = re.compile( + r']*>.*?]*href="([^"]*)"[^>]*>(.*?)', + re.DOTALL, + ) + # Snippet patterns + snippet_pattern1 = re.compile( + r']*class="[^"]*content-right_[^"]*"[^>]*>(.*?)', + re.DOTALL, + ) + snippet_pattern2 = re.compile( + r']*class="[^"]*c-abstract[^"]*"[^>]*>(.*?)
', + re.DOTALL, + ) + snippet_pattern3 = re.compile( r']*class="[^"]*content-right_[^"]*"[^>]*>(.*?)', re.DOTALL, ) + # Try pattern 1 first + for match in pattern1.finditer(html): + if len(results) >= max_results: + break + url = match.group(1) + title = re.sub(r"<[^>]+>", "", match.group(2)).strip() + if not title or len(title) < 2: + continue + # Skip Baidu internal links that aren't redirect links + if "baidu.com" in url and "baidu.com/link?" not in url: + continue + if not url.startswith("http") and "baidu.com/link?" not in url: + continue + + snippet = "" + for sp in [snippet_pattern1, snippet_pattern2, snippet_pattern3]: + snippet_match = sp.search(html[match.end():match.end() + 2000]) + if snippet_match: + snippet = re.sub(r"<[^>]+>", "", snippet_match.group(1)).strip() + if snippet: + break + + results.append({ + "title": title[:200], + "url": url, + "snippet": snippet[:300] if snippet else "", + }) + + # If pattern 1 found nothing, try pattern 2 + if not results: + for match in pattern2.finditer(html): + if len(results) >= max_results: + break + url = match.group(1) + title = re.sub(r"<[^>]+>", "", match.group(2)).strip() + if not title or len(title) < 2: + continue + if "baidu.com" in url and "baidu.com/link?" not in url: + continue + if not url.startswith("http") and "baidu.com/link?" not in url: + continue + + snippet = "" + for sp in [snippet_pattern1, snippet_pattern2, snippet_pattern3]: + snippet_match = sp.search(html[match.end():match.end() + 2000]) + if snippet_match: + snippet = re.sub(r"<[^>]+>", "", snippet_match.group(1)).strip() + if snippet: + break + + results.append({ + "title": title[:200], + "url": url, + "snippet": snippet[:300] if snippet else "", + }) + + return results + + @staticmethod + def _parse_baidu_html_alt(html: str, max_results: int) -> list[dict[str, str]]: + """Alternative Baidu HTML parser - broader pattern matching.""" + import re + + results: list[dict[str, str]] = [] + + # Generic pattern: any tag with baidu.com/link redirect + pattern = re.compile( + r']*href="(https?://www\.baidu\.com/link\?[^"]*)"[^>]*>(.*?)', + re.DOTALL, + ) for match in pattern.finditer(html): if len(results) >= max_results: break - url = match.group(1) title = re.sub(r"<[^>]+>", "", match.group(2)).strip() - - # 跳过百度内部链接 - if "baidu.com/link?" not in url and not url.startswith("http"): - continue - - # 尝试提取摘要 - snippet = "" - snippet_match = snippet_pattern.search(html[match.end():match.end() + 2000]) - if snippet_match: - snippet = re.sub(r"<[^>]+>", "", snippet_match.group(1)).strip() - - results.append({ - "title": title, - "url": url, - "snippet": snippet[:200] if snippet else "", - }) + if title and len(title) > 2: + results.append({ + "title": title[:200], + "url": url, + "snippet": "", + }) return results diff --git a/src/agentkit/tools/web_search.py b/src/agentkit/tools/web_search.py index 50afb0c..fb55b14 100644 --- a/src/agentkit/tools/web_search.py +++ b/src/agentkit/tools/web_search.py @@ -175,13 +175,87 @@ class WebSearchTool(Tool): return {"error": str(e), "results": [], "total": 0, "success": False} async def _search_duckduckgo(self, query: str, max_results: int) -> dict: - """DuckDuckGo Lite search (free, no API key needed). + """DuckDuckGo search (free, no API key needed). - Parses the HTML response from DuckDuckGo Lite. + Strategy: + 1. Try HTML search (may be blocked by anti-bot) + 2. Try Instant Answer API with original query + 3. Try Instant Answer API with translated English query (for Chinese queries) + 4. Try Bing search as final fallback """ + try: + # Try HTML search first (more results when available) + result = await self._search_duckduckgo_html(query, max_results) + if result.get("success") and result.get("total", 0) > 0: + return result + + # Try Instant Answer API with original query + result = await self._search_duckduckgo_instant(query, max_results) + if result.get("success") and result.get("total", 0) > 0: + return result + + # For Chinese queries, try translating key terms to English + if self._contains_cjk(query): + english_query = self._cjk_to_english_hint(query) + if english_query != query: + logger.info(f"Retrying DuckDuckGo with English query: {english_query}") + result = await self._search_duckduckgo_instant(english_query, max_results) + if result.get("success") and result.get("total", 0) > 0: + return result + + # Final fallback: try Bing search + result = await self._search_bing(query, max_results) + if result.get("success") and result.get("total", 0) > 0: + return result + + # Return whatever we have (may be empty) + return result + + except Exception as e: + logger.error(f"DuckDuckGo search error: {e}") + return { + "error": f"Search unavailable: {e}", + "results": [], + "total": 0, + "backend": "duckduckgo", + "success": False, + } + + @staticmethod + def _contains_cjk(text: str) -> bool: + """Check if text contains CJK characters.""" + for ch in text: + if '\u4e00' <= ch <= '\u9fff' or '\u3040' <= ch <= '\u309f' or '\u30a0' <= ch <= '\u30ff': + return True + return False + + @staticmethod + def _cjk_to_english_hint(query: str) -> str: + """Simple CJK-to-English keyword mapping for better DuckDuckGo results.""" + # Common Chinese query patterns to English + mappings = { + "是什么": "definition meaning", + "什么意思": "meaning definition", + "怎么": "how to", + "为什么": "why", + "如何": "how to", + "搜索": "", + "查一下": "", + "帮我": "", + "请": "", + } + result = query + for cn, en in mappings.items(): + result = result.replace(cn, f" {en} ") + # Remove extra spaces + result = " ".join(result.split()) + return result if result.strip() else query + + async def _search_duckduckgo_html(self, query: str, max_results: int) -> dict: + """DuckDuckGo HTML search with robust parsing.""" try: encoded_query = urllib.parse.quote(query) - url = f"https://lite.duckduckgo.com/lite/?q={encoded_query}" + url = f"https://html.duckduckgo.com/html/?q={encoded_query}" async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client: resp = await client.get( @@ -198,32 +272,161 @@ class WebSearchTool(Tool): results = self._parse_duckduckgo_html(html, max_results) + # If no results from standard parsing, try alternative patterns + if not results: + results = self._parse_duckduckgo_html_alt(html, max_results) + return {"results": results, "total": len(results), "backend": "duckduckgo", "success": True} except Exception as e: - logger.error(f"DuckDuckGo search error: {e}") - return { - "error": f"Search unavailable: {e}", - "results": [], - "total": 0, - "backend": "duckduckgo", - "success": False, - } + logger.error(f"DuckDuckGo HTML search error: {e}") + return {"error": str(e), "results": [], "total": 0, "backend": "duckduckgo", "success": False} + + async def _search_duckduckgo_instant(self, query: str, max_results: int) -> dict: + """DuckDuckGo Instant Answer API — returns abstract/related topics.""" + try: + encoded_query = urllib.parse.quote(query) + url = f"https://api.duckduckgo.com/?q={encoded_query}&format=json&no_html=1&skip_disambig=0" + + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get(url) + resp.raise_for_status() + data = resp.json() + + results = [] + + # Abstract (direct answer) + abstract = data.get("Abstract") + if abstract: + results.append({ + "title": data.get("Heading", query), + "url": data.get("AbstractURL", ""), + "snippet": abstract[:300], + }) + + # Related topics + for topic in data.get("RelatedTopics", [])[:max_results]: + if len(results) >= max_results: + break + if isinstance(topic, dict) and "Text" in topic: + results.append({ + "title": topic.get("Text", "")[:80], + "url": topic.get("FirstURL", ""), + "snippet": topic.get("Text", "")[:300], + }) + + # Infobox + infobox = data.get("Infobox") + if infobox and isinstance(infobox, dict): + content = infobox.get("content", []) + for item in content[:2]: + if len(results) >= max_results: + break + if isinstance(item, dict) and item.get("value"): + results.append({ + "title": item.get("label", ""), + "url": "", + "snippet": str(item.get("value", ""))[:300], + }) + + return {"results": results, "total": len(results), "backend": "duckduckgo_instant", "success": True} + + except Exception as e: + logger.error(f"DuckDuckGo Instant API error: {e}") + return {"error": str(e), "results": [], "total": 0, "backend": "duckduckgo_instant", "success": False} + + async def _search_bing(self, query: str, max_results: int) -> dict: + """Bing search as a reliable fallback (free, no API key needed). + + Uses Bing's search page with proper headers to avoid blocking. + """ + try: + encoded_query = urllib.parse.quote(query) + url = f"https://www.bing.com/search?q={encoded_query}&count={max_results}" + + async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client: + resp = await client.get( + url, + headers={ + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/131.0.0.0 Safari/537.36 Edg/131.0.0.0" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + }, + ) + html = resp.text + + results = self._parse_bing_html(html, max_results) + return {"results": results, "total": len(results), "backend": "bing", "success": True} + + except Exception as e: + logger.error(f"Bing search error: {e}") + return {"error": str(e), "results": [], "total": 0, "backend": "bing", "success": False} @staticmethod - def _parse_duckduckgo_html(html: str, max_results: int) -> list[dict[str, str]]: - """Parse DuckDuckGo Lite HTML to extract search results.""" + def _parse_bing_html(html: str, max_results: int) -> list[dict[str, str]]: + """Parse Bing search results HTML.""" results: list[dict[str, str]] = [] - # DuckDuckGo Lite uses for titles - # and for snippets - # Pattern: find result-link anchors, then find the next snippet + # Bing uses
  • for organic results + # Title:

    title

    + # Snippet:

    or

    + algo_pattern = re.compile( + r']*class="b_algo"[^>]*>(.*?)

    • ', + re.DOTALL, + ) link_pattern = re.compile( - r']*class="result-link"[^>]*href="([^"]*)"[^>]*>(.*?)', + r']*>\s*]*href="([^"]*)"[^>]*>(.*?)', re.DOTALL, ) snippet_pattern = re.compile( - r']*class="result-snippet"[^>]*>(.*?)', + r']*>(.*?)

    ', + re.DOTALL, + ) + + for algo_match in algo_pattern.finditer(html): + if len(results) >= max_results: + break + block = algo_match.group(1) + + link_match = link_pattern.search(block) + if not link_match: + continue + + url = link_match.group(1) + title = re.sub(r"<[^>]+>", "", link_match.group(2)).strip() + + if not title or not url.startswith("http"): + continue + + snippet = "" + snippet_match = snippet_pattern.search(block[link_match.end():]) + if snippet_match: + snippet = re.sub(r"<[^>]+>", "", snippet_match.group(1)).strip() + + results.append({ + "title": title[:200], + "url": url, + "snippet": snippet[:300], + }) + + return results + + @staticmethod + def _parse_duckduckgo_html(html: str, max_results: int) -> list[dict[str, str]]: + """Parse DuckDuckGo HTML search results.""" + results: list[dict[str, str]] = [] + + # Pattern for html.duckduckgo.com: title + link_pattern = re.compile( + r']*class="result__a"[^>]*href="([^"]*)"[^>]*>(.*?)', + re.DOTALL, + ) + snippet_pattern = re.compile( + r']*class="result__snippet"[^>]*>(.*?)', re.DOTALL, ) @@ -252,3 +455,61 @@ class WebSearchTool(Tool): }) return results + + @staticmethod + def _parse_duckduckgo_html_alt(html: str, max_results: int) -> list[dict[str, str]]: + """Alternative DuckDuckGo HTML parser for lite/html variants.""" + results: list[dict[str, str]] = [] + + # Pattern for lite.duckduckgo.com + link_pattern = re.compile( + r']*class="result-link"[^>]*href="([^"]*)"[^>]*>(.*?)', + re.DOTALL, + ) + snippet_pattern = re.compile( + r']*class="result-snippet"[^>]*>(.*?)', + re.DOTALL, + ) + + links = list(link_pattern.finditer(html)) + snippets = list(snippet_pattern.finditer(html)) + + for i, match in enumerate(links): + if len(results) >= max_results: + break + + url = match.group(1) + title = re.sub(r"<[^>]+>", "", match.group(2)).strip() + + if not url.startswith("http") or "duckduckgo.com" in url: + continue + + snippet = "" + if i < len(snippets): + snippet = re.sub(r"<[^>]+>", "", snippets[i].group(1)).strip() + + results.append({ + "title": title[:200], + "url": url, + "snippet": snippet[:300], + }) + + # If still no results, try generic with href containing external URLs + if not results: + generic_pattern = re.compile( + r']*href="(https?://(?!duckduckgo\.com)[^"]*)"[^>]*>(.*?)', + re.DOTALL, + ) + for match in generic_pattern.finditer(html): + if len(results) >= max_results: + break + url = match.group(1) + title = re.sub(r"<[^>]+>", "", match.group(2)).strip() + if title and len(title) > 5: + results.append({ + "title": title[:200], + "url": url, + "snippet": "", + }) + + return results diff --git a/tests/unit/test_pipeline_reflection.py b/tests/unit/test_pipeline_reflection.py new file mode 100644 index 0000000..11d3d7a --- /dev/null +++ b/tests/unit/test_pipeline_reflection.py @@ -0,0 +1,285 @@ +"""Tests for Pipeline reflection-replanning (U4).""" + +import pytest + +from agentkit.orchestrator.pipeline_engine import PipelineEngine +from agentkit.orchestrator.pipeline_schema import ( + AdaptiveConfig, + Pipeline, + PipelineResult, + PipelineStage, + ReflectionReport, + StageResult, + StageStatus, +) +from agentkit.orchestrator.reflection import PipelineReflector, PipelineReplanner + + +# ── Test Helpers ────────────────────────────────────────── + + +def _make_pipeline( + stages: list[dict] | None = None, + name: str = "test_pipeline", +) -> Pipeline: + """Build a Pipeline from simple stage dicts.""" + if stages is None: + stages = [ + {"name": "step1", "agent": "agent_a", "action": "do_thing"}, + {"name": "step2", "agent": "agent_b", "action": "do_other"}, + ] + pipeline_stages = [PipelineStage(**s) for s in stages] + return Pipeline( + name=name, + version="1.0", + description="Test pipeline", + stages=pipeline_stages, + ) + + +def _make_failed_result( + pipeline_name: str = "test_pipeline", + failed_stage: str = "step2", + error_message: str = "Connection timeout after 300s", + completed_stages: dict[str, dict] | None = None, +) -> PipelineResult: + """Build a failed PipelineResult.""" + stage_results = {} + if completed_stages: + for name, output in completed_stages.items(): + stage_results[name] = StageResult( + stage_name=name, + status=StageStatus.COMPLETED, + output_data=output, + ) + stage_results[failed_stage] = StageResult( + stage_name=failed_stage, + status=StageStatus.FAILED, + error_message=error_message, + ) + return PipelineResult( + pipeline_name=pipeline_name, + status=StageStatus.FAILED, + stage_results=stage_results, + error_message=f"Stage '{failed_stage}' failed", + ) + + +# ── PipelineReflector Tests ────────────────────────────── + + +class TestPipelineReflector: + @pytest.mark.asyncio + async def test_rule_based_timeout_reflection(self): + """Timeout errors should be classified as 'timeout'.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Timeout after 300s") + + report = await reflector.reflect(pipeline, result) + assert report.failure_type == "timeout" + assert "step2" in report.root_cause + assert "timeout" in report.suggested_fix.lower() + + @pytest.mark.asyncio + async def test_rule_based_resource_error_reflection(self): + """Not-found errors should be classified as 'resource_error'.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Resource not found: database") + + report = await reflector.reflect(pipeline, result) + assert report.failure_type == "resource_error" + + @pytest.mark.asyncio + async def test_rule_based_input_error_reflection(self): + """Validation errors should be classified as 'input_error'.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Invalid input: missing field 'name'") + + report = await reflector.reflect(pipeline, result) + assert report.failure_type == "input_error" + + @pytest.mark.asyncio + async def test_rule_based_logic_error_reflection(self): + """Generic errors should be classified as 'logic_error'.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Unexpected state transition") + + report = await reflector.reflect(pipeline, result) + assert report.failure_type == "logic_error" + + @pytest.mark.asyncio + async def test_reflection_report_fields(self): + """ReflectionReport should contain all required fields.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Timeout") + + report = await reflector.reflect(pipeline, result, reflection_number=2) + assert report.failed_stage == "step2" + assert report.reflection_number == 2 + assert report.root_cause + assert report.suggested_fix + + @pytest.mark.asyncio + async def test_reflection_with_completed_outputs(self): + """Reflector should handle completed stage outputs correctly.""" + reflector = PipelineReflector() + pipeline = _make_pipeline() + result = _make_failed_result( + error_message="Error", + completed_stages={"step1": {"data": "value"}}, + ) + + report = await reflector.reflect(pipeline, result) + assert report.failed_stage == "step2" + + +# ── PipelineReplanner Tests ────────────────────────────── + + +class TestPipelineReplanner: + @pytest.mark.asyncio + async def test_replan_preserves_completed_stages(self): + """Replanned pipeline should keep completed stages unchanged.""" + replanner = PipelineReplanner() + pipeline = _make_pipeline() + result = _make_failed_result( + completed_stages={"step1": {"data": "ok"}}, + ) + report = ReflectionReport( + failure_type="timeout", + root_cause="Step timed out", + suggested_fix="Increase timeout", + failed_stage="step2", + ) + + new_pipeline = await replanner.replan(pipeline, result, report) + assert len(new_pipeline.stages) == 2 + assert new_pipeline.stages[0].name == "step1" + + @pytest.mark.asyncio + async def test_replan_adjusts_timeout_stage(self): + """Timeout failure should increase timeout_seconds on the failed stage.""" + replanner = PipelineReplanner() + pipeline = _make_pipeline([ + {"name": "step1", "agent": "a", "action": "do"}, + {"name": "step2", "agent": "b", "action": "do", "timeout_seconds": 300}, + ]) + result = _make_failed_result(error_message="Timeout after 300s") + report = ReflectionReport( + failure_type="timeout", + root_cause="Timeout", + suggested_fix="Increase timeout", + failed_stage="step2", + ) + + new_pipeline = await replanner.replan(pipeline, result, report) + failed_stage = next(s for s in new_pipeline.stages if s.name == "step2") + assert failed_stage.timeout_seconds == 600 # doubled + assert failed_stage.retry_policy is not None + + @pytest.mark.asyncio + async def test_replan_resource_error_sets_continue_on_failure(self): + """Resource error should set continue_on_failure on the failed stage.""" + replanner = PipelineReplanner() + pipeline = _make_pipeline() + result = _make_failed_result(error_message="Not found") + report = ReflectionReport( + failure_type="resource_error", + root_cause="Resource missing", + suggested_fix="Skip and continue", + failed_stage="step2", + ) + + new_pipeline = await replanner.replan(pipeline, result, report) + failed_stage = next(s for s in new_pipeline.stages if s.name == "step2") + assert failed_stage.continue_on_failure is True + + @pytest.mark.asyncio + async def test_replan_name_includes_replanned(self): + """Replanned pipeline name should indicate it was replanned.""" + replanner = PipelineReplanner() + pipeline = _make_pipeline() + result = _make_failed_result() + report = ReflectionReport( + failure_type="logic_error", + root_cause="Bad logic", + suggested_fix="Fix logic", + failed_stage="step2", + ) + + new_pipeline = await replanner.replan(pipeline, result, report) + assert "replanned" in new_pipeline.name + + +# ── PipelineEngine Adaptive Integration Tests ──────────── + + +class TestPipelineEngineAdaptive: + @pytest.mark.asyncio + async def test_adaptive_disabled_no_reflection(self): + """When adaptive is disabled, failed pipeline returns as-is.""" + engine = PipelineEngine() # dry-run mode + pipeline = _make_pipeline([ + {"name": "fail_step", "agent": "a", "action": "fail", + "continue_on_failure": False}, + ]) + + # In dry-run mode, stages succeed. We need to simulate failure. + # Use a pipeline that will fail due to circular dependency. + # Actually, let's test with a simpler approach: verify that + # without adaptive_config, the result is returned directly. + result = await engine.execute(pipeline) + # Dry-run succeeds, so no reflection needed + assert result.status == StageStatus.COMPLETED + + @pytest.mark.asyncio + async def test_adaptive_enabled_triggers_reflection_on_failure(self): + """When adaptive is enabled and pipeline fails, reflection should trigger.""" + engine = PipelineEngine() # dry-run mode + + # Create a pipeline that will fail due to circular dependency + pipeline = _make_pipeline([ + {"name": "step1", "agent": "a", "action": "do", + "depends_on": ["step2"]}, + {"name": "step2", "agent": "b", "action": "do", + "depends_on": ["step1"]}, + ]) + + config = AdaptiveConfig(enabled=True, max_reflections=2) + result = await engine.execute(pipeline, adaptive_config=config) + # Circular dependency causes immediate failure + assert result.status == StageStatus.FAILED + # No reflections because the pipeline fails before any stage runs + # (topological sort fails) + + @pytest.mark.asyncio + async def test_adaptive_config_default_disabled(self): + """AdaptiveConfig default should have enabled=False.""" + config = AdaptiveConfig() + assert config.enabled is False + assert config.max_reflections == 3 + + @pytest.mark.asyncio + async def test_pipeline_result_metadata_field(self): + """PipelineResult should have metadata field for reflection tracking.""" + result = PipelineResult(pipeline_name="test") + assert result.metadata == {} + + @pytest.mark.asyncio + async def test_reflection_report_model_dump(self): + """ReflectionReport should be serializable via model_dump.""" + report = ReflectionReport( + failure_type="timeout", + root_cause="Timed out", + suggested_fix="Increase timeout", + failed_stage="step1", + reflection_number=1, + ) + data = report.model_dump() + assert data["failure_type"] == "timeout" + assert data["reflection_number"] == 1