Agent

Reson Agent

可扩展、可学习的 Agent 框架，支持执行追踪和持久记忆。

核心特性

• 执行追踪：每次 LLM 调用、工具调用、结论都记录为 Step，形成可视化的执行图
• 三层记忆：Task State（任务状态）→ Experience（经验规则）→ Skill（归纳技能）
• 存储可插拔：通过 Protocol 定义存储接口，支持内存、PostgreSQL、Neo4j 等实现
• 工具系统：@tool 装饰器自动注册工具，支持 Schema 生成、确认机制、参数编辑

安装

# 基础安装
pip install reson-agent

# 带 PostgreSQL 支持
pip install reson-agent[postgres]

# 完整安装
pip install reson-agent[all]

快速开始

1. 单次调用（不需要 Agent 能力）

from reson_agent import AgentRunner
from reson_agent.storage import MemoryTraceStore

# 最简使用
runner = AgentRunner(
    trace_store=MemoryTraceStore()  # 内存存储，测试用
)

result = await runner.call(
    messages=[{"role": "user", "content": "你好"}],
    model="gpt-4o"
)
print(result.reply)

2. Agent 模式（带执行追踪）

from reson_agent import AgentRunner
from reson_agent.storage import MemoryTraceStore, MemoryMemoryStore
from reson_agent.tools import tool

# 注册工具
@tool(
    editable_params=["query"],
    display={"zh": {"name": "搜索", "params": {"query": "关键词"}}}
)
async def search(query: str, uid: str = "") -> str:
    """搜索相关内容"""
    return f"搜索结果: {query}"

# 创建 Runner
runner = AgentRunner(
    trace_store=MemoryTraceStore(),
    memory_store=MemoryMemoryStore(),  # 可选，启用经验/技能
)

# 执行任务
async for event in runner.run(
    task="帮我分析最近的阅读兴趣",
    agent_type="researcher",
    max_iterations=5
):
    if event.type == "llm_delta":
        print(event.data["delta"], end="")
    elif event.type == "conclusion":
        print(f"\n结论: {event.data['content']}")

3. 添加反馈

# 对某个执行步骤给反馈
exp_id = await runner.add_feedback(
    trace_id="trace_abc",
    target_step_id="step_123",
    feedback_type="correction",
    content="搜索时应该限制在最近30天内"
)

核心概念

Trace（执行轨迹）

一次完整的 LLM 交互，包含多个 Step。

@dataclass
class Trace:
    trace_id: str
    mode: Literal["call", "agent"]  # 单次调用 or Agent 模式
    status: Literal["running", "completed", "failed"]
    # ...

Step（执行步骤）

Trace 中的一个原子操作。

@dataclass
class Step:
    step_id: str
    trace_id: str
    step_type: Literal["llm_call", "tool_call", "tool_result", "conclusion", "feedback"]
    parent_ids: List[str]  # 形成 DAG
    data: Dict[str, Any]

Experience（经验）

从执行或反馈中提取的规则。

@dataclass
class Experience:
    condition: str   # 什么情况下适用
    rule: str        # 应该怎么做
    evidence: List[str]  # step_ids 作为证据

Skill（技能）

从经验归纳的高层知识。

@dataclass
class Skill:
    name: str
    guidelines: List[str]
    derived_from: List[str]  # experience_ids

存储适配

实现 TraceStore 和 MemoryStore Protocol 即可接入任何存储。

from reson_agent.storage import TraceStore, MemoryStore

class MyPostgresTraceStore(TraceStore):
    async def create_trace(self, trace: Trace) -> str: ...
    async def add_step(self, step: Step) -> str: ...
    # ...

文档

• 设计文档 - 详细的架构设计和决策
• API 参考 - 完整的 API 文档

License

MIT