攀岩者

OpenRath：像搭积木一样构建多Agent系统

作者：

climbing

在

清华团队开源 · PyTorch风格API · 多Agent多Session协作框架

一、OpenRath是什么？

OpenRath是一个类PyTorch风格的多Agent多Session运行时框架，由清华团队开发。

简单说：它让你用写PyTorch神经网络的方式，来构建复杂的AI Agent系统。

为什么需要OpenRath？

想象一下，你要管理一个AI团队：

  • Agent A负责搜索信息
  • Agent B负责分析数据
  • Agent C负责生成报告
  • 它们需要共享上下文、分工协作、并行工作

传统方式就像让一群人用对讲机乱成一团，OpenRath则给了你一套清晰的组织架构。

核心理念：多Agent × 多Session

大多数Agent框架是”单Agent单Session”（像ChatGPT对话）。OpenRath支持：

模式 说明 场景

|——|——|——|

单Agent单Session 普通聊天 简单问答
多Agent单Session 多个Agent协作 团队协作
单Agent多Session 一个Agent处理多个任务 批量处理
多Agent多Session 多个Agent处理多个任务 复杂业务系统

二、PyTorch类比：秒懂核心概念

OpenRath最大的特点是用PyTorch的思维方式设计Agent系统：

PyTorch概念 OpenRath对应 作用

|————-|————-|——|

Tensor（张量） Session（会话） 承载所有状态数据
nn.Linear（线性层） Agent（智能体） 处理输入，产出输出
nn.Module（模块） Workflow（工作流） 组合多个Agent
Function（函数） Tool（工具） Agent可调用的能力
Device（设备） Sandbox（沙箱） 工具执行的环境
控制流 Selector（选择器） LLM驱动的动态路由
持久化状态 Memory（记忆） Agent的长期记忆

一句话理解：

  • Session = 流动的数据（像Tensor）
  • Agent = 处理数据的层（像nn.Linear）
  • Workflow = 组合层的网络（像nn.Module）

三、快速开始

3.1 安装

`bash

# 基础安装

pip install openrath

# 带可选依赖

pip install “openrath[opensandbox]” # 云端沙箱

pip install “openrath[openviking]” # 记忆后端

`

3.2 最小示例：Hello Agent

`python

from rath import Agent, Session, Workflow

# 创建一个Agent（像定义一个神经网络层）

agent = Agent(

name=”helper”,

prompt=”你是一个有用的助手，请简洁回答问题。”

)

# 创建一个Session（像创建一个Tensor）

session = Session()

session.add_user_message(“你好，请介绍一下自己”)

# 运行Agent（像前向传播）

result = agent.forward(session)

# 查看结果

print(result.last_message())

`

3.3 带工具的Agent

`python

from rath import Agent, Tool, Session

# 定义一个工具（像定义一个函数）

@Tool(

name=”search”,

description=”搜索网络信息”,

parameters={

“query”: {“type”: “string”, “description”: “搜索关键词”}

}

)

def search(query: str) -> str:

# 这里调用实际的搜索API

return f”搜索结果：关于'{query}’的信息…”

# 创建Agent并绑定工具

agent = Agent(

name=”researcher”,

prompt=”你是一个研究员，使用搜索工具查找信息。”,

tools=[search]

)

session = Session()

session.add_user_message(“帮我搜索一下Python最新版本”)

result = agent.forward(session)

print(result.last_message())

`

四、核心组件详解

4.1 Session：会话状态

Session是OpenRath的核心，就像PyTorch的Tensor。

`python

from rath import Session

# 创建Session

session = Session()

# 添加消息

session.add_system_message(“你是一个Python专家”)

session.add_user_message(“解释一下装饰器”)

session.add_assistant_message(“装饰器是…”)

session.add_tool_result(“search”, “搜索结果…”)

# Session操作（像Tensor操作）

forked = session.fork() # 分支（创建副本）

merged = session.merge(other) # 合并两个Session

compressed = session.compress() # 压缩（减少token）

detached = session.detach() # 断开（独立副本）

`

Session的魔力：

  • 分支：一个Session可以分成多个分支，每个分支独立发展
  • 合并：多个Session可以合并，综合结果
  • 追踪：每个Session都有血缘关系，可以追溯来源

4.2 Agent：智能体

Agent是处理Session的”层”，像nn.Linear：

`python

from rath import Agent

# 基础Agent

agent = Agent(

name=”summarizer”,

prompt=”请将以下内容总结成3个要点。”

)

# 带记忆的Agent

agent = Agent(

name=”assistant”,

prompt=”你是一个有用的助手。”,

memory=my_memory, # 绑定长期记忆

tools=[search, calculator] # 绑定工具

)

# 前向传播（处理Session）

output_session = agent.forward(input_session)

`

Agent特点：

  • 轻量：一个Agent只做一件事
  • 可复用：同一个Agent可以用在不同Workflow中
  • 可组合：多个Agent可以串联或并联

4.3 Workflow：工作流

Workflow是组合Agent的容器，像nn.Module：

`python

from rath import Workflow, Agent

# 定义Agent

researcher = Agent(name=”researcher”, prompt=”搜索信息…”)

analyzer = Agent(name=”analyzer”, prompt=”分析数据…”)

writer = Agent(name=”writer”, prompt=”撰写报告…”)

# 组合成Workflow

class ReportWorkflow(Workflow):

def __init__(self):

self.researcher = researcher

self.analyzer = analyzer

self.writer = writer

def forward(self, session):

# 1. 搜索

session = self.researcher.forward(session)

# 2. 分析

session = self.analyzer.forward(session)

# 3. 写报告

session = self.writer.forward(session)

return session

# 使用Workflow

workflow = ReportWorkflow()

result = workflow.forward(initial_session)

`

4.4 Tool：工具

Tool是Agent可以调用的函数：

`python

from rath import Tool

# 定义工具

@Tool(

name=”calculator”,

description=”计算数学表达式”,

parameters={

“expression”: {“type”: “string”, “description”: “数学表达式”}

}

)

def calculator(expression: str) -> str:

try:

result = eval(expression)

return str(result)

except Exception as e:

return f”计算错误: {e}”

# 工具会自动注册到Agent

agent = Agent(

name=”math_helper”,

prompt=”你是一个数学助手，使用calculator工具计算。”,

tools=[calculator]

)

`

4.5 Sandbox：沙箱

Sandbox是工具执行的环境：

`python

from rath import Sandbox

# 本地沙箱（默认）

local_sandbox = Sandbox(type=”local”)

# 云端沙箱（需要opensandbox）

cloud_sandbox = Sandbox(type=”opensandbox”, config={…})

# 工具在沙箱中执行

@Tool(sandbox=cloud_sandbox)

def run_code(code: str) -> str:

# 代码在云端安全执行

…

`

4.6 Memory：记忆

Memory是Agent的长期记忆：

`python

from rath import Memory

# 创建记忆

memory = Memory(type=”local”) # 或 “openviking” 云端

# Agent使用记忆

agent = Agent(

name=”assistant”,

prompt=”你是一个助手。”,

memory=memory

)

# 记忆会自动：

# – recall：在需要时回忆过去的信息

# – commit：在对话后保存新的信息

`

4.7 Selector：选择器

Selector是LLM驱动的动态路由：

`python

from rath import Selector, Workflow

# 定义多个Workflow

coding_workflow = Workflow(name=”coding”, …)

research_workflow = Workflow(name=”research”, …)

writing_workflow = Workflow(name=”writing”, …)

# 创建Selector（让LLM决定走哪条路）

selector = Selector(

name=”task_router”,

prompt=”根据用户任务类型，选择合适的工作流：”,

options=[coding_workflow, research_workflow, writing_workflow]

)

# 运行时，LLM会自动选择最合适的Workflow

result = selector.forward(session)

`

五、实战案例

5.1 案例1：智能客服系统

`python

from rath import Agent, Workflow, Tool, Session

# 定义工具

@Tool(name=”query_order”, description=”查询订单状态”)

def query_order(order_id: str) -> str:

# 查询数据库

return f”订单{order_id}状态：已发货”

@Tool(name=”refund”, description=”申请退款”)

def refund(order_id: str, reason: str) -> str:

# 处理退款

return f”订单{order_id}退款申请已提交”

# 定义Agent

classifier = Agent(

name=”classifier”,

prompt=”判断用户问题类型：订单查询、退款申请、其他”

)

order_agent = Agent(

name=”order_agent”,

prompt=”处理订单相关问题”,

tools=[query_order]

)

refund_agent = Agent(

name=”refund_agent”,

prompt=”处理退款相关问题”,

tools=[refund]

)

# 定义Workflow

class CustomerServiceWorkflow(Workflow):

def forward(self, session):

# 1. 分类

session = classifier.forward(session)

category = session.get_metadata(“category”)

# 2. 路由到对应Agent

if category == “订单查询”:

return order_agent.forward(session)

elif category == “退款申请”:

return refund_agent.forward(session)

else:

return session.add_assistant_message(“抱歉，我无法处理这个问题。”)

# 使用

workflow = CustomerServiceWorkflow()

session = Session()

session.add_user_message(“我的订单12345到哪了？”)

result = workflow.forward(session)

print(result.last_message())

`

5.2 案例2：研究助手（多Agent协作）

`python

from rath import Agent, Workflow, Session

# 定义专业Agent

searcher = Agent(name=”searcher”, prompt=”搜索相关文献和资料”)

reader = Agent(name=”reader”, prompt=”阅读和理解文档内容”)

summarizer = Agent(name=”summarizer”, prompt=”总结关键发现”)

critic = Agent(name=”critic”, prompt=”批判性分析和提出问题”)

# 定义研究Workflow

class ResearchWorkflow(Workflow):

def forward(self, session, topic):

# 1. 搜索资料

session.add_user_message(f”搜索关于{topic}的资料”)

session = searcher.forward(session)

# 2. 并行阅读多个文档

sessions = []

for doc in session.get_documents():

doc_session = session.fork()

doc_session.add_user_message(f”阅读这篇文档：{doc}”)

sessions.append(reader.forward(doc_session))

# 3. 合并阅读结果

merged = Session.merge(sessions)

# 4. 总结

merged.add_user_message(“总结所有发现”)

summary = summarizer.forward(merged)

# 5. 批判性分析

summary.add_user_message(“分析这些发现的局限性和问题”)

final = critic.forward(summary)

return final

# 使用

workflow = ResearchWorkflow()

result = workflow.forward(Session(), “大语言模型的最新进展”)

print(result.last_message())

`

5.3 案例3：代码生成与测试

`python

from rath import Agent, Workflow, Tool, Session

@Tool(name=”run_python”, description=”运行Python代码”)

def run_python(code: str) -> str:

# 在沙箱中安全执行

import subprocess

result = subprocess.run([“python3”, “-c”, code],

capture_output=True, text=True)

return result.stdout if result.returncode == 0 else result.stderr

@Tool(name=”write_file”, description=”写入文件”)

def write_file(path: str, content: str) -> str:

with open(path, ‘w’) as f:

f.write(content)

return f”文件{path}已写入”

# 定义Agent

coder = Agent(

name=”coder”,

prompt=”根据需求编写Python代码”,

tools=[run_python, write_file]

)

tester = Agent(

name=”tester”,

prompt=”编写测试用例并运行测试”,

tools=[run_python]

)

debugger = Agent(

name=”debugger”,

prompt=”分析测试失败原因并修复代码”,

tools=[run_python, write_file]

)

# 定义Workflow

class CodeGenerationWorkflow(Workflow):

def forward(self, session, requirement):

# 1. 生成代码

session.add_user_message(f”需求：{requirement}”)

session = coder.forward(session)

# 2. 测试

session.add_user_message(“为这段代码编写测试”)

session = tester.forward(session)

# 3. 如果测试失败，调试修复

if “FAIL” in session.last_message():

session.add_user_message(“测试失败，请修复代码”)

session = debugger.forward(session)

return session

# 使用

workflow = CodeGenerationWorkflow()

result = workflow.forward(Session(), “实现一个快速排序算法”)

print(result.last_message())

`

六、高级特性

6.1 Session分支与合并

`python

# 分支：一个任务分成多个子任务

main_session = Session()

main_session.add_user_message(“分析这个产品的优缺点”)

# 分支1：分析优点

pros_session = main_session.fork()

pros_session.add_user_message(“只分析优点”)

# 分支2：分析缺点

cons_session = main_session.fork()

cons_session.add_user_message(“只分析缺点”)

# 并行处理

pros_result = pros_agent.forward(pros_session)

cons_result = cons_agent.forward(cons_session)

# 合并结果

final = Session.merge([pros_result, cons_result])

`

6.2 Agent记忆系统

`python

from rath import Agent, Memory

# 创建长期记忆

memory = Memory(type=”local”, path=”./agent_memory”)

# Agent使用记忆

agent = Agent(

name=”personal_assistant”,

prompt=”你是一个个人助手，记住用户的偏好。”,

memory=memory

)

# 第一次对话

session1 = Session()

session1.add_user_message(“我喜欢Python，不喜欢Java”)

agent.forward(session1)

# 第二次对话（Agent会记得之前的偏好）

session2 = Session()

session2.add_user_message(“推荐一个编程语言”)

result = agent.forward(session2)

# Agent会推荐Python，因为它记得你的偏好

`

6.3 动态路由（Selector）

`python

from rath import Selector, Workflow

# 定义不同场景的Workflow

coding_wf = Workflow(name=”coding”, …)

research_wf = Workflow(name=”research”, …)

chat_wf = Workflow(name=”chat”, …)

# Selector让LLM决定走哪条路

router = Selector(

name=”smart_router”,

prompt=”根据用户输入的类型，选择最合适的工作流。”,

options=[coding_wf, research_wf, chat_wf]

)

# 运行时自动路由

session = Session()

session.add_user_message(“帮我写一个Python脚本”)

result = router.forward(session) # LLM会选择coding_wf

`

七、OpenRath vs 其他框架

特性 OpenRath LangChain AutoGPT CrewAI

|——|———-|———–|———|——–|

设计风格 PyTorch风格 链式调用 自主Agent 角色扮演
Session管理 ✅ 分支/合并 ❌ 单Session ❌ 单Session ❌ 单Session
多Agent协作 ✅ 原生支持 ⚠️ 需要额外配置 ⚠️ 有限 ✅ 角色协作
记忆系统 ✅ 内置 ⚠️ 需要集成 ⚠️ 有限 ⚠️ 有限
工具沙箱 ✅ 内置 ❌ 无 ❌ 无 ❌ 无
动态路由 ✅ Selector ⚠️ 需要自定义 ❌ 无 ❌ 无
学习曲线 中等 低 高 低
适合场景 复杂业务系统 快速原型 自主任务 团队协作

OpenRath的优势：

  1. Session分支/合并：唯一支持复杂Session管理的框架
  2. PyTorch风格：对AI开发者非常友好
  3. 工具沙箱：内置安全执行环境
  4. 动态路由：LLM驱动的智能路由

八、最佳实践

8.1 设计原则

  1. 一个Agent只做一件事

`python

# ❌ 错误：一个Agent做太多事

agent = Agent(prompt=”搜索、分析、总结、写报告”)

# ✅ 正确：拆分成多个专注的Agent

searcher = Agent(prompt=”搜索信息”)

analyzer = Agent(prompt=”分析数据”)

summarizer = Agent(prompt=”总结要点”)

writer = Agent(prompt=”撰写报告”)

`

  1. 用Session分支处理并行任务

`python

# ✅ 并行处理多个文档

for doc in documents:

forked = session.fork()

forked.add_user_message(f”处理文档：{doc}”)

results.append(agent.forward(forked))

`

  1. 用Memory避免重复工作

`python

# ✅ Agent记住已经处理过的内容

agent = Agent(prompt=”…”, memory=memory)

`

8.2 性能优化

  1. 压缩长Session

`python

# 当Session太长时，压缩历史

if session.token_count > 4000:

session = session.compress(keep_recent=10)

`

  1. 并行执行独立Agent

`python

import asyncio

# 并行执行多个Agent

async def parallel_agents(session, agents):

tasks = [agent.forward_async(session) for agent in agents]

return await asyncio.gather(*tasks)

`

  1. 缓存重复查询

`python

# 使用Memory缓存常见问题的答案

if memory.has_cache(question):

return memory.get_cache(question)

`

8.3 调试技巧

  1. 查看Session血缘

`python

# 追踪Session是怎么来的

print(session.lineage())

# 输出：root -> fork_1 -> agent_1 -> merge_1

`

  1. 记录Agent调用

`python

# 开启详细日志

import logging

logging.getLogger(“rath”).setLevel(logging.DEBUG)

`

  1. 可视化Workflow

`python

# 生成Workflow的可视化图

workflow.visualize(“workflow.png”)

`

九、常见问题

Q1: OpenRath和LangChain有什么区别？

A: LangChain是链式调用，适合快速原型。OpenRath是PyTorch风格，适合复杂业务系统，特别是需要Session分支/合并的场景。

Q2: OpenRath支持哪些LLM？

A: 支持所有兼容OpenAI API的模型，包括：

  • OpenAI GPT-4/3.5
  • Claude
  • 本地模型（Ollama、vLLM等）
  • 国产模型（通义千问、文心一言等）

Q3: 如何处理长文本？

A: 使用Session的压缩功能：

`python

session = session.compress(

strategy=”sliding_window”,

max_tokens=4000,

keep_recent=10

)

`

Q4: 如何部署到生产环境？

A: OpenRath支持：

  • Docker容器化部署
  • Kubernetes集群部署
  • 云端沙箱（OpenSandbox）

十、资源链接

  • GitHub: https://github.com/Rath-Team/OpenRath
  • 官方文档: https://docs.openrath.com
  • PyPI: https://pypi.org/project/openrath
  • 示例代码: https://github.com/Rath-Team/OpenRath/tree/main/example

十一、总结

OpenRath是一个面向复杂业务场景的多Agent框架，它的PyTorch风格设计让AI开发者感到亲切。

核心优势：

  1. Session分支/合并：唯一支持复杂Session管理
  2. PyTorch风格：学习曲线平缓
  3. 工具沙箱：安全执行环境
  4. 动态路由：智能任务分配

适合场景：

  • 需要多Agent协作的复杂系统
  • 需要并行处理多个任务
  • 需要长期记忆的Agent应用
  • 需要动态路由的智能系统

不适合场景：

  • 简单的聊天机器人
  • 快速原型验证
  • 单Agent单Session应用

💡 一句话总结：如果你想用写PyTorch的方式来构建AI Agent系统，OpenRath是你的最佳选择。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

更多文章