taolib.harness.agents.graph_agent

taolib.harness.agents.graph_agent#

LangGraph Agent 基类 - 定义基于 StateGraph 的可注册、可组合 Agent 抽象。

本模块提供两个核心抽象：

  • AgentConfig —— 通过 Pydantic v2 定义的 Agent 元配置（名称、版本、 描述、工具集、LLM 模型参数、最大迭代次数等）；

  • GraphAgent —— LangGraph Agent 的抽象基类，子类通过实现 build_graph() 声明节点与边，基类负责图编译、检查点接入、状态回调 广播以及与上层注册表/桥接层的协同。

并附带预置模板 ReActAgent，按 reason → act → observe → reason 循环 组织节点，支持工具调用与条件终止。

装饰器风格示例:

from taolib.harness.core import register_agent

@register_agent(name="planner", version="1.0.0", tags=("plan",))
class PlannerAgent(GraphAgent): ...

LangGraph 在当前环境可能未安装，本模块通过 TYPE_CHECKING 守卫与运行时 探测实现可插拔接入：未安装时仍可使用内置的轻量执行回退。

Attributes#

NodeFn

节点函数签名：接收当前状态字典，返回需要合并的更新字典（同步或异步）。

ToolFn

工具函数签名：任意可调用对象，参数与返回值由具体工具决定。

Classes#

AgentConfig

Agent 元配置。

AgentTool

Agent 包装为工具时的描述对象。

GraphAgent

LangGraph Agent 抽象基类。

GraphSpec

Agent 图结构的声明式规格。

LLMConfig

LLM 模型配置 - 用于 Agent 内部对底层模型的参数化调用。

ReActAgent

预置 ReAct 模式 Agent 模板。

Module Contents#

class taolib.harness.agents.graph_agent.AgentConfig#

Bases: pydantic.BaseModel

Agent 元配置。

用于在注册、装配、序列化阶段声明一个 Agent 的全部静态属性。

description: str#
llm: LLMConfig#
max_iterations: int#
model_config#
name: str#
tags: tuple[str, Ellipsis]#
tools: list[Any]#
version: str#
class taolib.harness.agents.graph_agent.AgentTool#

Agent 包装为工具时的描述对象。

可被任意支持 name / description / func 三元组的工具协议消费 （包括 LangChain StructuredTool、OpenAI function-call 等）。

args_schema: type[pydantic.BaseModel] | None = None#
description: str#
func: collections.abc.Callable[Ellipsis, collections.abc.Awaitable[Any]]#
name: str#
class taolib.harness.agents.graph_agent.GraphAgent(config: AgentConfig | collections.abc.Mapping[str, Any], *, registry: taolib.harness.core.registry.AgentRegistry | None = None, checkpointer: taolib.harness.runtime.checkpointer.HarnessCheckpointer | None = None, auto_register: bool = True)#

Bases: abc.ABC

LangGraph Agent 抽象基类。

生命周期：

  1. 实例化时自动向 AgentRegistry 注册自身（除非显式禁用）；

  2. 首次调用 compile() 时执行 build_graph() 并返回可执行图对象；

  3. invoke() / ainvoke() 为同步/异步执行入口；

  4. as_tool() 将 Agent 包装为可被其他 Agent 调用的工具；

  5. as_step() 将 Agent 适配为 Metaflow Step 的同步可调用。

钩子 on_node_start / on_node_end / on_error 用于观测节点级状态。

构造 Agent。

参数:

  • config -- AgentConfig 实例或可被其校验的字典。

  • registry -- 自定义 Agent 注册表；缺省使用全局默认注册表。

  • checkpointer -- 自定义检查点适配器；缺省构造内存型实例。

  • auto_register -- 是否在构造时自动注册到 registry。

async ainvoke(inputs: collections.abc.Mapping[str, Any] | None = None, config: collections.abc.Mapping[str, Any] | None = None) → dict[str, Any]#

异步执行 Agent。

as_step() → collections.abc.Callable[Ellipsis, dict[str, Any]]#

将 Agent 包装为可在 Metaflow Step 中调用的同步函数（Agent-as-Step）。

返回的可调用接受任意 inputs 字典，内部驱动 invoke()， 从而让 Flow Step 可以无缝复用 Agent 的图执行能力。

as_tool(*, name: str | None = None, description: str | None = None) → AgentTool#

将 Agent 包装为可被其他 Agent 调用的工具（Agent-as-Tool）。

abstractmethod build_graph() → GraphSpec#

返回 Agent 的图结构声明。

子类应在此声明所有节点函数与边连接，基类负责后续编译。

compile(*, force: bool = False) → _Compiled#

编译图谱声明为可执行对象。

优先使用 LangGraph 的 StateGraph；若环境未安装则回退到内置 _CompiledStubGraph 顺序执行器。

参数:

force -- 是否强制重新编译（忽略缓存）。

invoke(inputs: collections.abc.Mapping[str, Any] | None = None, config: collections.abc.Mapping[str, Any] | None = None) → dict[str, Any]#

同步执行 Agent。

property checkpointer: taolib.harness.runtime.checkpointer.HarnessCheckpointer#

关联的检查点适配器。

config#
config_class: ClassVar[type[AgentConfig]]#
on_error: collections.abc.Callable[[str, BaseException], collections.abc.Awaitable[None] | None] | None = None#
on_node_end: collections.abc.Callable[[str, dict[str, Any]], collections.abc.Awaitable[None] | None] | None = None#
on_node_start: collections.abc.Callable[[str, dict[str, Any]], collections.abc.Awaitable[None] | None] | None = None#
property registry_entry: taolib.harness.core.registry.RegistryEntry[Any] | None#

注册条目（若已注册）。

class taolib.harness.agents.graph_agent.GraphSpec#

Agent 图结构的声明式规格。

子类 GraphAgent.build_graph() 返回本结构后，由基类编译成可执行图。

conditional_edges: list[tuple[str, collections.abc.Callable[[dict[str, Any]], str], dict[str, str]]] = []#
edges: list[tuple[str, str]] = []#
entry_point: str = 'start'#
finish_points: tuple[str, Ellipsis] = ('end',)#
nodes: dict[str, NodeFn]#
class taolib.harness.agents.graph_agent.LLMConfig#

Bases: pydantic.BaseModel

LLM 模型配置 - 用于 Agent 内部对底层模型的参数化调用。

extra: dict[str, Any]#
max_tokens: int | None#
model: str#
model_config#
temperature: float#
top_p: float#
class taolib.harness.agents.graph_agent.ReActAgent(config: AgentConfig | collections.abc.Mapping[str, Any], *, registry: taolib.harness.core.registry.AgentRegistry | None = None, checkpointer: taolib.harness.runtime.checkpointer.HarnessCheckpointer | None = None, auto_register: bool = True)#

Bases: GraphAgent

预置 ReAct 模式 Agent 模板。

构造的图结构如下:

reason ──▶ act ──▶ observe ──▶ reason  (循环直至终止条件)
                          └──▶ end

子类可重写 reason()、act()、observe() 三个方法以注入 具体的 LLM/工具/解析逻辑；缺省实现仅做透传，便于在测试中校验图骨架。

构造 Agent。

参数:

  • config -- AgentConfig 实例或可被其校验的字典。

  • registry -- 自定义 Agent 注册表；缺省使用全局默认注册表。

  • checkpointer -- 自定义检查点适配器；缺省构造内存型实例。

  • auto_register -- 是否在构造时自动注册到 registry。

async act(state: dict[str, Any]) → dict[str, Any]#

执行阶段：调用工具或外部 API。

build_graph() → GraphSpec#

构建标准 reason → act → observe 循环。

async observe(state: dict[str, Any]) → dict[str, Any]#

观察阶段：吸收执行结果并决定是否继续。

async reason(state: dict[str, Any]) → dict[str, Any]#

推理阶段：基于状态生成下一步意图。

type taolib.harness.agents.graph_agent.NodeFn = Callable[[dict[str, Any]], dict[str, Any] | Awaitable[dict[str, Any]]]#

节点函数签名：接收当前状态字典，返回需要合并的更新字典（同步或异步）。

type taolib.harness.agents.graph_agent.ToolFn = Callable[..., Any]#

工具函数签名：任意可调用对象，参数与返回值由具体工具决定。