Node → Workflow → Agent 推导

# core/node.py
import time
from typing import Any, Dict, Optional, Tuple

shared = {}  # 全局状态，所有 Node 之间共享

class Node:
    def __init__(self, max_retries: int = 1, wait: float = 0):
        self.successors: Dict[str, "Node"] = {}
        self._action: str = "default"
        self.max_retries = max_retries
        self.wait = wait

    def exec(self, payload: Any) -> Tuple[str, Any]:
        """子类实现：返回 (action, next_payload)"""
        raise NotImplementedError

    def _exec(self, payload: Any) -> Tuple[str, Any]:
        """内部调用：处理重试逻辑"""
        for attempt in range(self.max_retries):
            try:
                return self.exec(payload)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(self.wait)

    def __rshift__(self, other: "Node") -> "Node":
        """node >> other  —— 连接到下一个节点"""
        self.successors[self._action] = other
        self._action = "default"
        return other

    def __sub__(self, action: str) -> "Node":
        """node - "action"  —— 给连接边打标签"""
        self._action = action
        return self


class Flow:
    def __init__(self, start: Optional[Node] = None):
        self.start = start

    def run(self, payload: Any = None) -> Any:
        """沿有向图走到没有后继节点为止"""
        node = self.start
        while node:
            action, payload = node._exec(payload)
            node = node.successors.get(action)
        return payload

# examples/workflow/main.py
from core.node import Node, Flow, shared
from core.llm import call_llm_simple
from tools.builtins.search import search_web

class InputNode(Node):
    def exec(self, query):
        shared["query"] = query
        return "search", query

class SearchNode(Node):
    def exec(self, query):
        results = search_web(query)
        shared["search_results"] = results
        return "summarize", results

class SummarizeNode(Node):
    def exec(self, results):
        prompt = f"请总结以下搜索结果：\n{results}\n\n原始问题：{shared['query']}"
        answer = call_llm_simple(prompt)
        return "default", answer

# 构图
input_node  = InputNode()
search_node = SearchNode()
summary_node = SummarizeNode()

input_node  - "search"    >> search_node
search_node - "summarize" >> summary_node

# 执行
flow = Flow(input_node)
result = flow.run("Python asyncio 最佳实践")
print(result)

# examples/chatbot/main.py
from core.node import Node, Flow, shared
from core.llm import call_llm

shared["messages"] = []

class ChatNode(Node):
    def exec(self, _):
        response = call_llm(shared["messages"])
        shared["messages"].append(response)
        return "output", response["content"]

class OutputNode(Node):
    def exec(self, text):
        print(f"\nAssistant: {text}\n")
        return "default", None

# 构图
chat_node   = ChatNode()
output_node = OutputNode()
chat_node - "output" >> output_node

# 循环
while True:
    user_input = input("You: ").strip()
    if not user_input:
        continue
    shared["messages"].append({"role": "user", "content": user_input})
    Flow(chat_node).run()

# examples/chatbot_with_tools/main.py
from core.node import Node, Flow, shared
from core.llm import call_llm
from tools.executor import ToolExecutor
from tools.builtins import ALL_TOOLS

shared["messages"] = []
executor = ToolExecutor(ALL_TOOLS)

class ChatNode(Node):
    def exec(self, _):
        response = call_llm(
            shared["messages"],
            tools=[t.to_llm_format() for t in ALL_TOOLS]
        )
        shared["messages"].append(response)

        # 有 tool_calls → 去执行工具
        if response.get("tool_calls"):
            return "tool_call", response["tool_calls"]
        # 没有 → 直接输出
        return "output", response["content"]

class ToolCallNode(Node):
    def exec(self, tool_calls):
        results = executor.execute(tool_calls)
        # 把工具结果追加到对话历史
        for result in results:
            shared["messages"].append(result.to_message())
        # 路由回 ChatNode
        return "chat", None

class OutputNode(Node):
    def exec(self, text):
        print(f"\nAssistant: {text}\n")
        return "default", None

# 构图（注意回路）
chat_node      = ChatNode()
tool_call_node = ToolCallNode()
output_node    = OutputNode()

chat_node      - "tool_call" >> tool_call_node
tool_call_node - "chat"      >> chat_node      # 回路
chat_node      - "output"    >> output_node

# 循环
while True:
    user_input = input("You: ").strip()
    if not user_input:
        continue
    shared["messages"].append({"role": "user", "content": user_input})
    Flow(chat_node).run()

def call_llm_simple(prompt: str) -> str:
    """单轮，string in，string out"""
    ...

def call_llm(
    messages: list,
    tools: list = None,
    system_prompt: str = None
) -> dict:
    """多轮，返回 assistant message dict（含 tool_calls 字段）"""
    ...