轻量多 Agent 框架：约 700 行实现多 Agent 系统的核心原语——类型化工具调用、Swarm 式 handoff 路由、map-reduce 团队、护栏、记忆。零框架依赖，全部 11 个测试由确定性 MockLLM 驱动，不碰网络。
LangChain 解决的是"什么都有"，agent-mesh 想回答的是"Agent 循环的本质是什么"。整个框架只有五个概念：LLM 接口、工具注册表、Agent 循环、handoff、编排器。读懂这 700 行，任何 Agent 框架（OpenAI Swarm / LangGraph / Claude Agent SDK）的骨架都是它的超集。
route: triage -> billing | handoffs: 1
answer: A-77 已发货、可退款，已为你提交退款申请。
triage agent 调用 transfer_to_billing → Swarm 切换到 billing agent → billing 调用 lookup_order 工具 → 观察结果 → 生成最终回答。整条链路的每一步都在测试里被断言。
| 模块 | 是什么 | 关键设计 |
|---|---|---|
tools.py |
@tool 装饰器 |
类型注解 + docstring 自动生成 JSON Schema（与 OpenAI tools API 同契约）；工具异常被封装成 error 字符串返回给模型，让它观察-重试而不是崩溃 |
agent.py |
Agent 循环 | reason → tool_calls → observe → repeat；max_turns 防死循环；transcript 保持 OpenAI 消息格式，可直接喂给下一个 agent |
handoff.py + swarm.py |
Swarm 路由 | handoff 本身是一个终端工具：执行即跳出循环，由 Swarm 读取路由；只扫描本轮新增 transcript，避免历史 handoff 被重复消费；max_handoffs 防乒乓 |
guardrails.py |
护栏管道 | block（拒绝）/ transform（改写）两种语义：正则拦截注入、信用卡号脱敏、长度上限，可自由组合 |
memory.py |
会话记忆 | 滑动窗口 + 超窗折叠为运行摘要（summarizer 注入式，测试用桩） |
另有 Team（map-reduce）：planner 拆任务 → N 个 worker 并行执行（接口同构，换 asyncio.gather 即真并行）→ reducer 汇总。
uv sync && uv run pytest # 11 passed，零网络
# 接真实模型（任何 OpenAI 兼容端点：OpenAI / DeepSeek / vLLM / Ollama）
export OPENAI_API_KEY=sk-...
export AGENT_MESH_MODEL=deepseek-chat
uv run python examples/customer_support.py "我想退款，订单 A-77"三十行搭一个客服 swarm：
from agent_mesh import Agent, OpenAICompatibleLLM, Swarm, tool
from agent_mesh.tools import ToolRegistry
llm = OpenAICompatibleLLM()
@tool
def lookup_order(order_id: str) -> str:
"""Look up an order by id."""
return orders.get(order_id, "not found")
billing_reg = ToolRegistry()
billing_reg.register(lookup_order)
triage_reg = ToolRegistry()
triage = Agent("triage", "判断问题类型并移交给 billing 或 tech", llm,
tools=triage_reg, handoffs=["billing", "tech"])
billing = Agent("billing", "账务客服：先查单再答复", llm, tools=billing_reg)
tech = Agent("tech", "技术支持", llm)
result = Swarm([triage, billing, tech], entry="triage").run("A-77 想退款")
print(result.route, result.final_text)MockLLM 回放脚本化响应并记录每次调用的消息与工具列表——所以测试能断言的是框架行为本身：工具 schema 是否正确生成、异常是否被封装、handoff 后 transcript 是否完整传递、护栏是否按序生效。LLM 的"聪明"是端点的事，框架的正确性是自己的事。OpenAICompatibleLLM 用同一个接口（complete(messages, tools)）对接真实端点，API key 只从环境变量读取。
Q：MockLLM 是什么？为什么要用它做测试？
MockLLM 是一个 BaseLLM 的确定性实现（agent_mesh/llm.py）：传入一组脚本化的 LLMResponse（或返回响应的可调用对象），每次 complete() 按序弹出一条，并记录每轮收到的消息数与可用工具清单供断言；脚本耗尽会直接失败。这让整个测试套件零网络、可重复——测的是框架行为（schema 生成、异常封装、transcript 传递、护栏顺序），不是模型的聪明程度。换上 OpenAICompatibleLLM 即可无缝切到真实端点，二者实现同一接口。
Q：它与 LangChain / OpenAI Swarm 是什么关系？
不是替代品，是"最小骨架"。agent-mesh 只实现五个原语（LLM 接口、工具注册表、Agent 循环、handoff、编排器），handoff 语义与 OpenAI Swarm 同构（transfer_to_X 终端工具 + 路由器读 transcript）。适合用来理解 Agent 框架的运行原理、做原型与教学；需要生态集成（向量库、RAG 组件、可视化追踪）时，LangChain/LangGraph 是更完整的选择。
Q：工具 schema 自动生成怎么用？
给一个带类型注解和 docstring 的函数加 @tool 装饰器即可：参数类型从注解映射为 JSON Schema（支持 int / float / bool / str / list / dict 六种），docstring（或 @tool(description=...)）成为工具描述，带默认值的参数不进 required。用 ToolRegistry.register() 注册、schemas() 导出、dispatch() 调用；工具抛出的异常会被封装成 error: ... 字符串返回给模型，供它观察后重试。
Q：怎么接真实模型？
OpenAICompatibleLLM 对接任何 OpenAI chat 兼容端点（OpenAI / DeepSeek / Qwen / vLLM / Ollama）：AGENT_MESH_MODEL、OPENAI_BASE_URL、OPENAI_API_KEY 三个环境变量分别控制模型名、端点地址与密钥（密钥只从环境读取，不写入代码）。HTTP 层用 httpx 直发、不依赖官方 SDK，httpx 是可选依赖——不装也不影响框架核心与测试运行。
Q：Team（map-reduce）的 worker 是真并行吗？
当前是串行执行——这是刻意选择，保证结果确定、断言可写。接口与 asyncio.gather 同构：需要真并行时，把 Team.run 里的 for 循环换成 gather 即可，planner / worker / reducer 的签名无需改动。
Q：会话记忆（Memory）怎么接入？
Memory 是独立组件，不与 Agent 循环自动绑定：它保留最近 window 条消息（默认 20），超窗的旧消息交给注入的 summarize 回调折叠成运行摘要（可注入真实 LLM 或测试桩），render() 返回"摘要 + 窗口"组成的消息列表。是否接入、在哪一轮接入，由你的调用链决定。
- 无异步与流式支持：核心循环全部是同步阻塞调用；
OpenAICompatibleLLM用 httpx 同步 POST（超时 120 秒），不支持stream；Team的 worker 串行执行。需要真并行要自行改写为 asyncio。 - 护栏与记忆不自动接线：
run_pipeline()与Memory是独立组件，Agent/Swarm不会自动调用它们——输入输出护栏和会话记忆都需要你在自己的调用链里显式接入。 - 工具参数注解仅覆盖基础类型：只映射
int/str/bool/float/list/dict六种；Optional、Literal、dataclass、嵌套泛型等复杂注解会退化为"string"。 - 内置 LLM 适配只有 OpenAI 兼容一种：Anthropic 原生 API、Gemini 等需自行实现
BaseLLM.complete()；HTTP 请求失败直接抛出（raise_for_status），没有重试与退避，也不统计 token 用量。 - 规模防护是浅层的：
Agent默认max_turns=8、Swarm默认max_handoffs=4，超限即返回错误文本或停止路由；没有循环检测、成本预算、人工审批等更细的防护机制。 - 无持久化：
Swarm.run每次从空白 transcript 开始，跨次运行的会话状态不落盘；多轮对话需要你自行配合Memory或外部存储管理。
MIT