shortgraph 0.1.0

A lightweight, secure, and neutral Python Agent framework.

ShortGraph

ShortGraph 是一个轻量级的、受 LangGraph 启发的 Python 图编排框架，专为构建复杂的 Agent 工作流而设计。

它摒弃了传统链式（Chain）调用的局限性，采用**图（Graph）**作为核心抽象，赋予开发者对 Agent 控制流的极致掌控力。无论是循环、条件分支、并行执行，还是人工介入（HITL），ShortGraph 都能以简洁的代码实现。

---

核心架构 (Core Architecture)

ShortGraph 的设计严格遵循图执行引擎的四大维度：

1. 图执行引擎 (Pregel Engine):
   • 核心运行时，负责调度节点执行、传递状态。
   • 采用 BSP (Bulk Synchronous Parallel) 模型，支持多节点并行与同步。
   • 位于 shortgraph/pregel/ 目录。
2. 状态管理 (State Management):
   • AgentState：在节点间流转的共享状态对象。
   • 支持类似 Redux 的状态更新机制，确保数据流向清晰。
   • 位于 shortgraph/channels/ 目录。
3. 图定义 (Graph Definition):
   • StateGraph：用于定义图的拓扑结构。
   • Node (节点)：执行具体逻辑的函数（或子图）。
   • Edge (边)：定义节点间的流转方向。
   • Conditional Edge (条件边)：基于运行时状态动态决定下一跳。
   • 位于 shortgraph/graph/ 目录。
4. 运行时协议 (Runtime Protocol):
   • 支持同步 invoke 和流式 stream 执行。
   • 支持 Checkpointer 持久化机制，实现“断点续传”和“时光倒流”。

---

🛠️ 安装 (Installation)

目前推荐源码安装：

git clone https://github.com/your-repo/shortgraph.git
cd shortgraph
pip install -r requirements.txt

---

🚀 快速开始 (Quick Start)

1. 定义一个最简单的图

from shortgraph.graph.state import StateGraph
from shortgraph.channels.manager import AgentState

# 1. 定义节点函数
def step_1(state: AgentState):
    print("执行步骤 1")
    state.set_variable("count", 1)
    return state

def step_2(state: AgentState):
    val = state.get_variable("count")
    print(f"执行步骤 2, 当前计数: {val}")
    state.set_variable("count", val + 1)
    return state

# 2. 构建图
workflow = StateGraph()
workflow.add_node("node_1", step_1)
workflow.add_node("node_2", step_2)

# 3. 定义流向
workflow.set_entry_point("node_1")
workflow.add_edge("node_1", "node_2")
workflow.add_edge("node_2", "__end__")

# 4. 编译
app = workflow.compile()

# 5. 运行
state = AgentState()
app.invoke(state)

---

📚 详细教程与 API (Detailed Documentation)

1. 图构建 (StateGraph)

StateGraph 是你的工作流蓝图。

• add_node(name, func, retry=0):
  • name: 节点唯一标识符。
  • func: 接收 AgentState 并返回 AgentState 的函数。
  • retry: (可选) 失败自动重试次数。
• add_edge(start_key, end_key):
  • 定义从 start_key 到 end_key 的确定性流转。
• add_conditional_edges(source_key, condition_func, path_map):
  • 实现逻辑分支。
  • condition_func: 接收 state，返回字符串（意图）。
  • path_map: 字典，将 condition_func 的返回值映射到下一个节点名称。
• set_entry_point(key): 指定图的起始节点。
• compile(checkpointer=..., interrupt_before=...):
  • 将图编译为可执行的 Pregel 应用。

2. 状态管理 (AgentState)

在所有节点间共享的数据容器。

• state.get_variable(key): 获取变量值。
• state.set_variable(key, value): 更新变量值。

3. 模型集成 (LLMFactory)

ShortGraph 内置了对主流大模型的支持，统一封装为 BaseLLM 接口。

from shortgraph.models.factory import LLMFactory

# OpenAI
llm = LLMFactory.create_openai(api_key="sk-...")

# DeepSeek
llm = LLMFactory.create_deepseek(api_key="sk-...")

# Zhipu AI (智谱清言)
llm = LLMFactory.create_zhipu(api_key="...")

# Ollama (本地模型)
llm = LLMFactory.create_ollama(model="llama3")

4. 进阶功能 (Advanced Features)

A. 人工介入 (Human-in-the-loop)

在关键节点前暂停，等待人工审批或修改状态。

from shortgraph.checkpoint.memory import MemoryCheckpoint

# 启用检查点记录
checkpointer = MemoryCheckpoint()

# 编译时指定中断点
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["approval_node"] # 在进入 approval_node 前暂停
)

# 运行（会暂停）
app.invoke(state, thread_id="thread_1")

# ... 人工检查，更新 state ...

# 恢复运行（传入相同的 thread_id）
app.invoke(state, thread_id="thread_1")

B. 自动重试 (Auto-Retry)

对于不稳定的节点（如网络请求），可以设置自动重试。

# 如果失败，自动重试 3 次
workflow.add_node("api_call", call_api_func, retry=3)

C. 子图嵌套 (Subgraphs)

图可以作为节点嵌入到另一个图中，实现模块化复用。

child_app = child_workflow.compile()
parent_workflow.add_node("research_department", child_app)

---

📂 目录结构 (Directory Structure)

shortgraph/
├── pregel/         # [核心] 图执行引擎 (main, loop, algo)
├── graph/          # [定义] 图结构定义 (StateGraph, Node)
├── channels/       # [状态] 状态管理 (AgentState)
├── checkpoint/     # [记忆] 持久化与历史记录
├── models/         # [模型] LLM 工厂 (OpenAI, DeepSeek, Zhipu...)
├── paradigms/      # [范式] 预置 Agent 模式 (SmartAgent, GraphReAct...)
├── tools/          # [工具] 工具定义基类
└── examples/       # [示例] 丰富的实战代码

🧩 示例代码 (Examples)

请查看 examples/ 目录获取更多实战代码：

• examples/showcase.py: SmartAgent 完整演示（天气查询）。
• examples/hitl_agent.py: 人工介入 (HITL) 流程演示。
• examples/retry_agent.py: 自动重试 机制演示。
• examples/subgraph_agent.py: 多 Agent 协作（子图嵌套）演示。
• examples/graph_agent.py: 手动构建 ReAct 图的演示。

---

ShortGraph - Make Agents Simple & Powerful.