一切皆插件，真能落地吗

DeepSeek Harness 是 deepseek-ai 在 2026 年推出的 LLM 调用框架，核心理念是”Everything is a Plugin”——把模型、工具、记忆、回调全部抽象为插件，通过统一调度器串联。它不是又一个 Agent 框架，而是更底层的”调用编排层”。截至 2026/08/14，GitHub Trending 显示其 star 数突破 37.5k，单日增长 2k+，登顶 AI 类目榜首。它的爆火源于一个反共识判断：在 LangChain、LlamaIndex 还在堆 Agent 抽象时，DeepSeek 选择回归”插件即函数”的 Unix 哲学，开发者终于能用一行 pipeline.register() 替换整段胶水代码。

1. 插件怎么注册才不会被调度器忽略？

现象：很多开发者在写第一个插件时，会直接 pipeline.add(plugin)，结果运行时报 PluginNotFound。在 Issue #142 里，有人贴出 200 行的复现代码，问题仅仅是没显式声明 name 字段。

根因：Harness 的调度器按 Plugin.name 字段建立哈希索引，如果用默认的类名做 name，多个同名插件会互相覆盖；异步插件如果没继承 AsyncPlugin 基类，调度器会直接静默丢弃，不会抛错。

解法：显式声明 name，并继承正确的基类：

from harness import AsyncPlugin, Pipeline

class TranslatePlugin(AsyncPlugin):
    name = "translate-zh"  # 必须显式声明
    version = "0.1.0"

    async def run(self, ctx):
        ctx.output["text"] = await self.translate(ctx.input["text"])
        return ctx

pipeline = Pipeline()
pipeline.register(TranslatePlugin())

2. 插件之间如何共享状态而不破坏隔离？

现象：插件 A 想读插件 B 的输出，结果发现 ctx.output 永远是空。这是 HN 上被讨论最多的设计争议，有人甚至怀疑这是”设计缺陷”。

根因：Harness 默认每个插件拿到的是隔离 ctx，输出通过 Pipeline.bus 事件总线广播，而不是直接 dict 赋值。设计者认为这是为了避免插件间隐式耦合，但新人很容易踩坑。

解法：订阅事件总线而非直接读 ctx：

class SentimentPlugin(AsyncPlugin):
    name = "sentiment"

    async def run(self, ctx):
        async for event in self.bus.subscribe("translate-zh"):
            text = event.payload["text"]
            ctx.output["sentiment"] = await self.analyze(text)
        return ctx

3. 流式响应在插件链里为什么会”卡住”？

现象：开发者把单插件跑得通，但接入流水线后 SSE 流停在第一 token，要等 30 秒才看到完整输出。Reddit r/LocalLLaMA 用户 @karman 抱怨：“这跟非流式有什么区别？”

根因：插件链默认是 batch 模式，所有上游插件 finish 后下游才启动。要开启流式必须显式声明 stream=True 并使用 StreamingPlugin 基类，否则 yield 会被缓存。

解法：

from harness import StreamingPlugin

class StreamLLMPlugin(StreamingPlugin):
    name = "stream-llm"
    stream = True

    async def stream_run(self, ctx):
        async for chunk in self.llm.stream(ctx.input["prompt"]):
            await self.emit(chunk)  # 推送到下游

4. 插件报错时如何定位是哪个环节出问题？

现象：Pipeline 抛出 PipelineAbort，堆栈指向调度器，看不到具体是哪个插件。社区里有个梗：“Harness 报错就像黑盒，祈祷吧。”

根因：Harness 的默认 logger 是 silent，必须手动挂载 trace。日志系统与调度器解耦虽然性能更好，但调试体验极差。

解法：注册一个高优先级 trace 插件拦截全链路：

from harness import Plugin

class TracePlugin(Plugin):
    name = "trace"
    priority = -999  # 最先执行

    async def run(self, ctx):
        print(f"[{ctx.current_plugin}] input={ctx.input}")
        try:
            return await ctx.next()
        except Exception as e:
            print(f"FAIL at {ctx.current_plugin}: {e}")
            raise

5. 能否用 Harness 完全替换 LangChain Agent？

现象：很多迁移用户在 Twitter/X 上问”Harness 是不是 LangChain 杀手”，讨论热度比项目本身还高。

根因：两者抽象层级不同。Harness 是调用编排，LangChain Agent 是决策循环。如果你的场景是固定 pipeline（总结→翻译→分类），Harness 完胜；如果是动态决策（Agent 自己选工具），目前还需自实现 ReAct 循环。

解法：渐进式迁移。先把无状态插件迁过去，再把需要决策的部分包成 DecisionPlugin，内部依然调 LangChain：

class DecisionPlugin(AsyncPlugin):
    name = "agent-decision"

    async def run(self, ctx):
        from langchain.agents import initialize_agent
        agent = initialize_agent(self.tools, self.llm)
        ctx.output["decision"] = await agent.arun(ctx.input["query"])
        return ctx

Sources

• GitHub Trending 2026-08-14: deepseek-ai/deepseek-harness
• GitHub Issue #142: “PluginNotFound on first registration”
• Hacker News: “Show HN: DeepSeek Harness – Everything is a Plugin”
• r/LocalLLaMA: “Stream stalls after first token in pipeline mode”
• Twitter/X: @karman “Harness is not LangChain killer, here’s why”