在暗夜中开启全景雷达：LSP (语言服务器协议) 桥接

(第 58 篇：Agent 动力学之 LSP 桥接)

前面我们给 Agent 装上了 AST 与搜索， 但它依然像在黑屋里摸象： 只看一个 utils.py， 模型无法 100% 确认 calculate_fee() 在跨越多个目录的引用中， 究竟指向哪个真正的定义， 也无法确定某个符号的类型推导结果。

如果你希望 Agent 拥有“编译器级视野”， 就必须把 IDE 的那套确定性能力接出来： definition、references、hover、diagnostics、code actions。 这条路就是 LSP（Language Server Protocol）桥接。

但注意： LSP 不是一句“启动一个 server”就完事的。 它是一套字节级协议， 外面包着传输 framing， 里面跑着 JSON-RPC 2.0， 再往里才是语言智能。 你要把这一层层协议拆开讲清， 才能写出一个不会在生产环境里碎掉的桥接器。 citeturn0search0turn0search1

1. 不要让模型猜：确定性事实应该来自语言服务器

大模型擅长推理意图， 但不擅长做确定性的静态分析。 在大型项目里，歧义无处不在：

• 字符串歧义：项目里可能存在 10 个同名的 handleRequest 函数。大模型通过 grep 找出来的结果往往包含大量干扰项。
• 上下文缺失：如果模型只读了当前文件，它无法知道引入的 ThirdPartyLibrary 到底暴露了哪些方法。

与其让模型用 token 堆出一个“可能正确”的索引， 不如问一个“必须正确”的事实来源： 语言服务器。 你把它当成编译器的前端， 它会给你： 符号绑定、类型信息、诊断、以及可执行的重构动作。

2. LSP 的真实形态：JSON-RPC 2.0 + Content-Length framing

LSP 是一套基于 JSON-RPC 2.0 的协议， 最常见的传输方式是 stdio， 每条消息使用类似 HTTP 的头部 framing：

1. 头部里必须有 Content-Length: <bytes>。
2. 头部与内容之间用 \r\n\r\n 分隔。
3. 内容部分是一个 JSON 文本，遵守 JSON-RPC 2.0 的 request/response/notification 语义。 citeturn0search0turn0search1

先把“术语”钉死：

• request：有 id，需要 response。
• notification：没有 id，不期待 response。
• response：返回给某个 id 的结果或错误。

这不是小细节。 Agent 桥接器最容易犯的错就是把 LSP 当作“发一个 JSON 等一个 JSON”， 结果遇到：

1. server 的主动推送（diagnostics）。
2. 并发乱序返回（id 对应）。
3. 粘包/拆包（stdout 读取）。

就开始崩溃。

3. 生命周期：initialize 不是开始，shutdown 才是结束

一个最小但正确的生命周期顺序：

1. client -> initialize（request，有 id）。
2. server -> initialize response（携带 capabilities）。
3. client -> initialized（notification）。
4. 后续开始 didOpen/didChange 并消费 diagnostics。
5. client -> shutdown（request）。
6. server -> shutdown response。
7. client -> exit（notification，通常随后关闭进程）。 citeturn0search0

你在工程里要做的是： 把这套生命周期封装成一个“可重复、可观测”的会话对象， 并对错误路径做出明确处理： 超时、崩溃、协议错误、能力缺失。

4. 【核心代码】Headless LSP Client：正确处理 framing 与并发

下面是一份“强调关键点”的桥接器骨架， 它故意不把细节堆满， 但把最容易错的点都写进注释里：

import asyncio
import json
import subprocess
from dataclasses import dataclass
from typing import Any, Optional

class LSPBridge:
    """
    Agent 的语言服务器桥接器（协议层）。
    目标：
    1) 正确发送 JSON-RPC 消息（带 Content-Length）
    2) 正确从 stdout 粘包流里拆出一条条消息
    3) 支持并发 request，并用 id 做回填
    """
    def __init__(self, server_path: str, workspace_root: str):
        self.server_path = server_path
        self.root = workspace_root
        self.proc = None
        self._req_id = 0
        self._pending: dict[int, asyncio.Future] = {}

    async def start(self):
        # 以流模式启动后台语言服务器进程
        self.proc = await asyncio.create_subprocess_exec(
            self.server_path, "--stdio",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL
        )
        # 关键：需要一个后台 reader，持续消费 stdout 并拆包
        asyncio.create_task(self._reader_loop())

        # 发起 initialize（request）
        init_result = await self._send_request("initialize", {
            "rootUri": f"file://{self.root}",
            "capabilities": {}
        })
        # initialized（notification）
        await self._send_notification("initialized", {})
        return init_result

    async def _send_request(self, method: str, params: dict, timeout_s: float = 20.0):
        self._req_id += 1
        request_id = self._req_id
        payload: dict[str, Any] = {
            "jsonrpc": "2.0",
            "id": request_id,
            "method": method,
            "params": params
        }
        fut: asyncio.Future = asyncio.get_running_loop().create_future()
        self._pending[request_id] = fut

        await self._send_payload(payload)
        return await asyncio.wait_for(fut, timeout=timeout_s)

    async def _send_notification(self, method: str, params: dict):
        payload: dict[str, Any] = {"jsonrpc": "2.0", "method": method, "params": params}
        await self._send_payload(payload)

    async def _send_payload(self, payload: dict[str, Any]):
        assert self.proc and self.proc.stdin
        content = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        header = f"Content-Length: {len(content)}\r\n\r\n".encode("ascii")
        self.proc.stdin.write(header + content)
        await self.proc.stdin.drain()

    async def _reader_loop(self):
        assert self.proc and self.proc.stdout
        buf = b""
        while True:
            chunk = await self.proc.stdout.read(4096)
            if not chunk:
                break
            buf += chunk
            # 从 buf 中反复拆包：header -> content_length -> content JSON
            while True:
                sep = buf.find(b"\r\n\r\n")
                if sep < 0:
                    break
                header_bytes = buf[:sep].decode("ascii", errors="replace")
                buf = buf[sep + 4 :]
                content_length = self._parse_content_length(header_bytes)
                if content_length is None:
                    # 协议错误：为了安全，直接丢弃本次缓冲并中止
                    buf = b""
                    break
                if len(buf) < content_length:
                    # 内容不够，继续读
                    break
                content = buf[:content_length]
                buf = buf[content_length:]
                self._handle_message(content)

    def _parse_content_length(self, header: str) -> Optional[int]:
        for line in header.splitlines():
            if line.lower().startswith("content-length:"):
                try:
                    return int(line.split(":", 1)[1].strip())
                except ValueError:
                    return None
        return None

    def _handle_message(self, content: bytes):
        msg = json.loads(content.decode("utf-8", errors="replace"))
        # response
        if "id" in msg and ("result" in msg or "error" in msg):
            request_id = msg["id"]
            fut = self._pending.pop(request_id, None)
            if fut and not fut.done():
                fut.set_result(msg)
            return
        # server notifications（diagnostics 等）
        # 这里应该把它们转成“可复盘事实”，再喂给上层 agent。
        return

    async def get_definition(self, file_path: str, line: int, char: int):
        """精准打击：获取某个变量的真正家园"""
        return await self._send_request("textDocument/definition", {
            "textDocument": {"uri": f"file://{file_path}"},
            "position": {"line": line, "character": char}
        })

5. 把 LSP 变成“证据链”：Agent 不要吃猜测，要吃事实

LSP 的价值不仅是“跳转定义”， 而是把静态事实变成可复盘证据：

1. definition：这个符号解析成哪个文件哪个 range。
2. hover：这个表达式的类型是什么，来自哪里。
3. references：有哪些引用点，数量是多少。
4. diagnostics：当前错误列表（含 code、range、severity）。

你的桥接层应该把这些结果落成“分析包”：

• 记录 request 参数。
• 记录 raw response。
• 记录 server 版本、workspace hash、capabilities。
• 计算一个 content hash，作为后续比较的基准。

这样 verify 回路才是确定性的： 你能比较“修改前后 diagnostics 是否减少”， 而不是让模型凭感觉说“我觉得好了”。

6. 用 LSP 驱动自我纠错：把“报错喂回模型”做成最小闭环

一个可落地的“零 Bug”闭环如下：

1. Agent 通过 AST/编辑器工具做了一次变更（写入）。
2. Runner 立即触发 LSP 的诊断刷新（通常来自 publishDiagnostics 或显式拉取）。
3. 如果出现错误，Runner 不把全量日志丢回模型， 而是把“错误摘要 + 相关符号 hover/definition + 最近一次变更意图”回注。
4. Agent 产出下一次语义意图（可能是 rename、import 修复、类型纠正）。
5. Runner 再次 apply + verify，直到错误收敛或触发断路器。

关键点在于： 模型不是在看“模糊的 stdout”， 它在看“语言服务器的确定性事实”。

你不需要在文章里承诺具体成功率， 你只需要把机制讲清： 事实输入越确定， 闭环越短， 失败越可控。

7. 限制：为什么不能完全依赖 LSP（以及怎么分层）

LSP 很强，但也有软肋：

1. 启动慢：大型项目初始化可能很久。
2. 资源重：频繁 didChange 可能把机器打满。
3. 脏代码：语法破碎时分析质量下降。
4. 能力差异：不同 server 支持的 capabilities 不一致。

因此健康的架构是分层协作：

1. 搜索负责缩小候选范围（便宜、粗）。
2. AST 负责结构定位与可控写入（手术刀）。
3. LSP 负责语义事实与验证证据（编译器视野）。

再补一条安全边界： 不要把 LSP 的 code action 当作“可以直接执行的命令”。 它只是建议。 真正执行必须走 preview + transaction apply， 并且写入权限要遵守你的沙箱策略。

本章精粹（落地要点）

1. LSP 不是“发 JSON”，而是“Content-Length framing + JSON-RPC 2.0 + 生命周期”。 citeturn0search0turn0search1
2. 桥接器最难的不是 API，而是粘包拆包、并发回填、取消与超时。
3. 把 LSP 返回值当作证据链落盘，verify 回路才是确定性的。
4. 架构上让 AST 写、让 LSP 验，避免把“建议”直接当“执行”。

掌握了 LSP，你的 Agent 已经进化成了半个资深架构师。接下来，我们将迈向这趟技术旅程的“最终协议门阁”——【MCP (Model Context Protocol) 革命：如何构建通用的 USB-C 级 Agent 接口？】。我们要开始建立 Agent 的大一统时代了。

(本文完 - 深度解析系列 24)

参考与延伸（写作核验）

• LSP 3.18 规范（传输、生命周期、capabilities 等）。 citeturn0search0
• JSON-RPC 2.0 规范（request/response/notification 语义）。 citeturn0search1