解耦：将 examples 兼容逻辑迁移到 Host 适配层

框架内核仅保留 ProjectEnvironmentResolver 与通用内容质量端口；LegacyExampleResolver 和旧 SourceQualityEvaluator 的动态加载移到 examples/legacy_adapters，由 api_server 组合根注入。增加 project_name 兼容、评分器注入和安装包无业务 import 测试。

agent/agent/tools/builtin/content/platforms/aigc_channel.py

@@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional
 import httpx
 
 from agent.tools.models import ToolResult
+from agent.tools.builtin.content.quality import get_post_evaluator
 from agent.tools.utils.image import build_image_grid, encode_base64, load_images
 from agent.tools.builtin.content.registry import (
     PlatformDef, ParamSpec, register_platform,
@@ -161,12 +162,7 @@ async def search(
     # 构建概览摘要
     summary_list = []
 
-    # 动态导入评价模块
-    try:
-        from examples.process_pipeline.script.evaluate_source_quality import SourceQualityEvaluator
-        evaluator = SourceQualityEvaluator()
-    except ImportError:
-        evaluator = None
+    evaluator = get_post_evaluator()
 
     # 视频帖在评分前先并发探测 mp4 duration（HTTP Range，不下载视频流），
     # 让 evaluator 用真实时长替代 body 长度作为内容信号。

agent/agent/tools/builtin/content/platforms/x.py

@@ -12,6 +12,7 @@ import httpx
 from agent.tools.models import ToolResult
 from agent.tools.utils.image import build_image_grid, encode_base64, load_images
 from agent.tools.builtin.content.registry import PlatformDef, register_platform
+from agent.tools.builtin.content.quality import get_post_evaluator
 
 CRAWLER_URL = "http://crawler.aiddit.com/crawler/x/keyword"
 COMMENT_URL = "http://crawler.aiddit.com/crawler/x/comment"
@@ -38,12 +39,7 @@ async def search(
         result_data = data.get("data", {})
         tweets = result_data.get("data", []) if isinstance(result_data, dict) else []
 
-        # 动态导入评价模块
-        try:
-            from examples.process_pipeline.script.evaluate_source_quality import SourceQualityEvaluator
-            evaluator = SourceQualityEvaluator()
-        except ImportError:
-            evaluator = None
+        evaluator = get_post_evaluator()
 
         # 视频帖在评分前先并发探测 mp4 duration（HTTP Range，不下载视频流），
         # 让 evaluator 用真实时长替代 body 长度作为内容信号。

agent/agent/tools/builtin/content/platforms/youtube.py

@@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional
 import httpx
 
 from agent.tools.models import ToolResult
+from agent.tools.builtin.content.quality import get_post_evaluator
 from agent.tools.utils.image import build_image_grid, encode_base64, load_images
 from agent.tools.builtin.content.registry import (
     PlatformDef, ParamSpec, register_platform,
@@ -182,12 +183,7 @@ async def search(
         for v in videos:
             _normalize_youtube_post(v)
 
-        # 动态导入评价模块
-        try:
-            from examples.process_pipeline.script.evaluate_source_quality import SourceQualityEvaluator
-            evaluator = SourceQualityEvaluator()
-        except ImportError:
-            evaluator = None
+        evaluator = get_post_evaluator()
 
         # 概览
         summary_list = []

agent/agent/tools/builtin/content/quality.py

@@ -0,0 +1,41 @@
+"""Host-injected, business-neutral content quality adapter."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Dict, Optional, Protocol
+
+
+logger = logging.getLogger(__name__)
+
+
+class PostEvaluator(Protocol):
+    def evaluate_post(self, post: Dict[str, Any]) -> Dict[str, Any]: ...
+
+
+_post_evaluator_factory: Optional[Callable[[], PostEvaluator]] = None
+
+
+def set_post_evaluator_factory(
+    factory: Optional[Callable[[], PostEvaluator]],
+) -> None:
+    """Inject an optional business evaluator from the Host composition root."""
+    global _post_evaluator_factory
+    _post_evaluator_factory = factory
+
+
+def get_post_evaluator() -> Optional[PostEvaluator]:
+    if _post_evaluator_factory is None:
+        return None
+    try:
+        return _post_evaluator_factory()
+    except Exception as exc:
+        logger.warning("Host post evaluator is unavailable: %s", exc)
+        return None
+
+
+__all__ = [
+    "PostEvaluator",
+    "set_post_evaluator_factory",
+    "get_post_evaluator",
+]

agent/agent/trace/project_environment.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import importlib
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Protocol
 
@@ -22,31 +21,4 @@ class ProjectEnvironmentResolver(Protocol):
     ) -> Optional[ProjectEnvironment]: ...
 
 
-class LegacyExampleResolver:
-    """Compatibility adapter retained by the host, outside the API kernel."""
-
-    async def resolve(
-        self,
-        project_name: str,
-        messages: Optional[List[Dict[str, Any]]] = None,
-    ) -> Optional[ProjectEnvironment]:
-        module_name = f"examples.{project_name}.run"
-        try:
-            module = importlib.import_module(module_name)
-        except ImportError as exc:
-            if getattr(exc, "name", None) == module_name:
-                return None
-            raise
-        initializer = getattr(module, "init_project_env", None)
-        if initializer is None:
-            return None
-        if messages is None:
-            runner, resolved_messages, config = await initializer()
-        else:
-            runner, resolved_messages, config = await initializer(messages)
-        return ProjectEnvironment(runner, resolved_messages, config)
-
-
-__all__ = [
-    "ProjectEnvironment", "ProjectEnvironmentResolver", "LegacyExampleResolver",
-]
+__all__ = ["ProjectEnvironment", "ProjectEnvironmentResolver"]

agent/api_server.py

@@ -25,7 +25,7 @@ from agent.trace.run_api import (
     set_runner,
     set_project_environment_resolver,
 )
-from agent.trace.project_environment import LegacyExampleResolver
+from examples.legacy_adapters import LegacyExampleResolver, configure_legacy_content_adapters
 from agent.trace.websocket import router as ws_router, set_trace_store as set_ws_trace_store
 from agent.trace.examples_api import router as examples_router
 from agent.trace.logs_websocket import router as logs_router, setup_websocket_logging
@@ -86,6 +86,7 @@ runner = AgentRunner(
 )
 set_runner(runner)
 set_project_environment_resolver(LegacyExampleResolver())
+configure_legacy_content_adapters()
 
 
 # ===== 注册路由 =====

agent/examples/legacy_adapters.py

@@ -0,0 +1,47 @@
+"""Legacy Host adapters kept outside the installable Agent framework package."""
+
+from __future__ import annotations
+
+import importlib
+from typing import Any, Dict, List, Optional
+
+from agent.tools.builtin.content.quality import set_post_evaluator_factory
+from agent.trace.project_environment import ProjectEnvironment
+
+
+class LegacyExampleResolver:
+    async def resolve(
+        self,
+        project_name: str,
+        messages: Optional[List[Dict[str, Any]]] = None,
+    ) -> Optional[ProjectEnvironment]:
+        module_name = f"examples.{project_name}.run"
+        try:
+            module = importlib.import_module(module_name)
+        except ImportError as exc:
+            if getattr(exc, "name", None) == module_name:
+                return None
+            raise
+        initializer = getattr(module, "init_project_env", None)
+        if initializer is None:
+            return None
+        if messages is None:
+            runner, resolved_messages, config = await initializer()
+        else:
+            runner, resolved_messages, config = await initializer(messages)
+        return ProjectEnvironment(runner, resolved_messages, config)
+
+
+def configure_legacy_content_adapters() -> None:
+    """Preserve the old optional source scorer without coupling the framework."""
+
+    def create_evaluator() -> Any:
+        module = importlib.import_module(
+            "examples.process_pipeline.script.evaluate_source_quality"
+        )
+        return module.SourceQualityEvaluator()
+
+    set_post_evaluator_factory(create_evaluator)
+
+
+__all__ = ["LegacyExampleResolver", "configure_legacy_content_adapters"]

agent/pyproject.toml

@@ -50,3 +50,4 @@ include = ["agent*"]
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]
+pythonpath = ["."]

agent/tests/test_content_adapter_boundary.py

@@ -0,0 +1,77 @@
+import json
+import re
+from pathlib import Path
+
+import pytest
+
+import agent
+from agent.tools.builtin.content.platforms import x
+from agent.tools.builtin.content.quality import set_post_evaluator_factory
+
+
+class FakeEvaluator:
+    def evaluate_post(self, post):
+        return {"total_score": 88, "grade": "A"}
+
+
+class FakeResponse:
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return {
+            "code": 0,
+            "data": {
+                "data": [{
+                    "body_text": "generic post",
+                    "channel_account_name": "author",
+                    "link": "https://example.test/post",
+                }]
+            },
+        }
+
+
+class FakeClient:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    async def post(self, *args, **kwargs):
+        return FakeResponse()
+
+
+@pytest.mark.asyncio
+async def test_host_injected_content_evaluator_preserves_optional_scoring(monkeypatch):
+    import agent.tools.builtin.content.transcription as transcription
+
+    async def noop_probe(*args, **kwargs):
+        return None
+
+    async def no_collage(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(x.httpx, "AsyncClient", lambda *args, **kwargs: FakeClient())
+    monkeypatch.setattr(x, "_build_tweet_collage", no_collage)
+    monkeypatch.setattr(transcription, "probe_durations_for_posts", noop_probe)
+    set_post_evaluator_factory(FakeEvaluator)
+    try:
+        result = await x.search("x", "topic", max_count=1)
+        row = json.loads(result.output)["data"][0]
+        assert row["quality_score"] == 88
+        assert row["quality_grade"] == "A"
+    finally:
+        set_post_evaluator_factory(None)
+
+
+def test_installable_framework_has_no_examples_business_imports():
+    package_root = Path(agent.__file__).resolve().parent
+    forbidden = re.compile(
+        r"(?:from\s+examples\b|import\s+examples\b|import_module\([^\n]*examples)"
+    )
+    violations = []
+    for path in package_root.rglob("*.py"):
+        if forbidden.search(path.read_text(encoding="utf-8")):
+            violations.append(str(path.relative_to(package_root)))
+    assert violations == []

agent/tests/test_host_legacy_adapters.py

@@ -0,0 +1,77 @@
+import importlib
+from types import SimpleNamespace
+
+import pytest
+
+from agent.core.runner import RunConfig
+from agent.trace.project_environment import ProjectEnvironment
+from examples.legacy_adapters import LegacyExampleResolver
+
+
+@pytest.mark.asyncio
+async def test_legacy_example_resolver_success_missing_and_internal_error(monkeypatch):
+    resolver = LegacyExampleResolver()
+    received = []
+
+    async def initializer(messages=None):
+        received.append(messages)
+        return "project-runner", messages or [{"role": "user", "content": "default"}], RunConfig()
+
+    module = SimpleNamespace(init_project_env=initializer)
+    monkeypatch.setattr(importlib, "import_module", lambda name: module)
+    resolved = await resolver.resolve("demo", [{"role": "user", "content": "input"}])
+    assert resolved.runner == "project-runner"
+    assert received[-1][0]["content"] == "input"
+
+    def missing(name):
+        error = ImportError("missing project")
+        error.name = "examples.missing.run"
+        raise error
+
+    monkeypatch.setattr(importlib, "import_module", missing)
+    assert await resolver.resolve("missing") is None
+
+    def internal_error(name):
+        error = ImportError("missing dependency")
+        error.name = "business_dependency"
+        raise error
+
+    monkeypatch.setattr(importlib, "import_module", internal_error)
+    with pytest.raises(ImportError, match="dependency"):
+        await resolver.resolve("broken")
+
+
+@pytest.mark.asyncio
+async def test_run_api_uses_injected_project_environment(monkeypatch):
+    from agent.trace import run_api
+
+    selected = {}
+    project_runner = object()
+
+    class Resolver:
+        async def resolve(self, project_name, messages=None):
+            selected["project_name"] = project_name
+            return ProjectEnvironment(
+                runner=project_runner,
+                messages=[{"role": "user", "content": "resolved"}],
+                default_config=RunConfig(),
+            )
+
+    async def fake_run(messages, config, trace_id_future, runner_instance=None):
+        selected["runner"] = runner_instance
+        selected["messages"] = messages
+        trace_id_future.set_result("project-trace")
+
+    monkeypatch.setattr(run_api, "_project_environment_resolver", Resolver())
+    monkeypatch.setattr(run_api, "_runner", object())
+    monkeypatch.setattr(run_api, "_run_with_trace_signal", fake_run)
+    response = await run_api.create_and_run(
+        run_api.CreateRequest(
+            messages=[{"role": "user", "content": "original"}],
+            project_name="demo",
+        )
+    )
+    assert response.trace_id == "project-trace"
+    assert selected["project_name"] == "demo"
+    assert selected["runner"] is project_runner
+    assert selected["messages"][0]["content"] == "resolved"