夜雨聆风
SECTION
引言
各类产品界面中已集成形态各异的 AI 能力，例如语音转写纪要、AI 生成内容、拍照答疑、对话输入框等功能。
看似简单的产品界面，实则对应两种截然不同的 AI 应用实现形态：
1. 对话输入框代表 Agent：以大语言模型（LLM）作为决策核心，可自主规划任务、支持多轮交互，天然适配开放式、持续性任务，外在表现为「对话交互」形态。
2. 功能按钮 / API 代表 Graph：例如上文的「录音总结」按钮，其底层 Graph 流程大致为：录音文件输入→LLM 解析并生成摘要→保存录音记录，是一套固定执行流程。Graph 的核心在于流程确定性、任务闭环，依靠预先定义的节点与连线达成指定目标，外在表现为「功能组件」形态。视频生成类 API 同样属于 Graph 形态 AI 应用。
SECTION
核心概念解析
基础定义
1. Graph（流程编排）：由开发者预先定义拓扑结构的流程图，节点可承载代码函数、API 调用、LLM 调用等逻辑，输入输出通常为结构化数据。核心特征是确定性：相同输入下，执行路径与最终输出可完全预判。
2. Agent（智能体）：以 LLM 为核心的执行实体，能够自主规划、决策并完成任务；通过与外部环境（工具、用户、其他智能体）动态交互达成目标，行为具备不确定性。核心特征是自主性。
3. Tool（工具）：可供 Agent 调用的外部能力载体，一般是封装独立功能的函数或 API。工具支持同步 / 异步、有状态 / 无状态等类型，仅负责执行逻辑，不具备自主决策能力。
4. Orchestration（编排）：协调调度多个计算单元（节点、智能体）协同工作的过程；本文语境下特指通过 Graph 实现静态流程预定义。
深度对比表
| Agent智能体 | Graph流程编排 | |
|---|---|---|
| LLM | ||
Key Point
Agent 是由 LLM 全权驱动、具备自主能力的智能实体，通过调用工具使用外部能力；Graph 具备执行确定性，通过拓扑连线串联各类外部能力，节点内可嵌入 LLM 完成生成、判断逻辑。
SECTION
发展历程：从确定性流程到自主智能
2022 年 Langchain 框架首次发布时，大模型行业主流 API 仍是 OpenAI 文本补全接口，仅支持简单「文本输入 - 文本输出」模式。Langchain 最初的定位是「连接 LLM 与外部计算、数据源」，典型链式结构如下。
此后 ReAct（思考 - 行动）范式被正式提出，首次系统性论证 LLM 可通过「思考 - 执行 - 观测」循环与外部世界交互，为 Agent 自主规划能力奠定理论基础。几乎同期，OpenAI 推出对话补全 API，将大模型交互模式从单轮文本输入输出升级为多轮对话；函数调用（Function Calling）能力随之落地，赋予 LLM 标准化调用外部函数、API 的能力。至此，具备多轮对话、自主调用外部能力的应用形态 ——Agent 正式成型。
行业两大技术路线同步演化：
1. 编排框架分支：Langchain 推出 LangGraph，简单链式编排升级为复杂拓扑流程。这类编排框架完美适配 Graph 形态 AI 应用：支持结构化输入、以最终结果为核心输出、对话历史等状态与编排逻辑解耦，灵活编排 LLM、知识库等各类节点组件。
2. 智能体框架分支：AutoGen、CrewAI、谷歌 ADK 等大量 Agent / 多智能体框架涌现。这类框架核心解决「LLM 驱动动态流程」「上下文传递」「记忆管理」「多智能体协作」等问题，与编排框架聚焦的「复杂流程串联外部系统」属于两类不同诉求。
尽管定位不同，编排框架也能实现 ReAct智能体、多智能体等模式，但本质属于模拟实现，好比用 Word 编写代码：功能可实现，但底层设计存在天然适配缺陷。编排框架原生面向确定性Graph 流程，而 Agent 核心是动态变化的思考链路，强行用流程框架承载智能体逻辑，会在输出形态、运行模式上产生适配痛点。
SECTION
融合路径探索：Agent 与 Graph 的关系
以上痛点根源在于底层架构设计差异：专为确定性流程打造的 Graph 框架，很难原生支撑以动态思考链路为核心的自主智能体系统。
Eino 框架同时覆盖 Graph 编排与 Agent 两大场景，技术演进路径为先推出 Graph 编排组件 eino-compose，再独立开发智能体组件 eino-adk。两套模块看似割裂，开发体验无法互通，且该分离架构会长期保留，但部分能力会深度打通。下文从三个维度拆解 Eino 框架内 Agent 与 Graph 的关联。
多智能体场景与流程编排
Agent 与 Graph 存在本质差异，但多智能体场景属于二者交叉领域，无法简单二选一。多智能体即多个智能体通过特定规则协作，对外整体表现为一个完整智能体，其协作规则能否用 Graph 编排实现？我们分三类主流协作模式分析：
1. 层级调用（智能体作为工具）：行业最通用方案（谷歌 ADK 标准范式）：顶层主智能体处理用户交互，遇到代码执行、数据分析等细分任务时，调用专用子智能体。子智能体通常无状态、不与主智能体共享记忆，交互逻辑仅为简单函数调用。主智能体与子智能体仅存在调用与被调用关系，不属于 Graph 编排的节点流转逻辑。
2. 预设固定协作流程：针对成熟标准化协作模式，例如「规划 - 执行 - 重规划」（Plan-Execute-Replan），智能体间交互顺序、角色分工固定。Eino-adk 等框架会将该类模式封装为预制多智能体模板，开发者开箱即用，无需手动配置智能体流转关系。底层 Graph 编排仅作为模板内部实现细节，对开发者完全透明。
多智能体协作可通过「智能体作为工具」简化实现，或使用框架预制固定协作模板；完全动态协作场景则和 Graph 静态编排诉求完全割裂，二者底层编排逻辑不存在交集。
将 Agent 作为 Graph 内节点
换个视角思考：Graph 编排流程中是否有嵌入 Agent 的需求？智能体能否作为 Graph 的一个节点？
先梳理二者核心冲突：
1. Agent 输入来源多元：除上游节点结构化数据外，强依赖自身对话记忆；Graph 节点仅以上游输出作为唯一输入源。
2. Agent 输出为异步完整过程数据，下游节点难以解析复用。
若强行将 Agent 嵌入 Graph 节点，等同于强制一个需要多轮对话、长期记忆、异步流式输出的智能体，适配确定性、同步阻塞的流程节点，实现方案繁琐且架构不优雅。Graph 仅可启动智能体，智能体内部复杂交互逻辑不应阻塞主流程。
实际 Graph 流程中，开发者真正需要的是轻量化「LLM 节点」：在确定流程中接收标准输入，完成意图识别、内容生成，输出结构化数据，为流程注入基础智能能力。
若业务场景确实需要完整 Agent 能力，更合理的方案不是将智能体塞入静态 Graph，而是为 Agent 增加预处理、后处理插件，将业务逻辑内嵌至智能体内部。
小结：直接把 Agent 作为 Graph 节点并非最优解；优先选用轻量化 LLM 节点；确需智能体能力时，通过插件向 Agent 注入业务逻辑。
最优融合方案：Graph 封装为 Agent 的工具
既然微观节点层面直接融合存在诸多缺陷，宏观层面是否存在更简洁的联动方案？答案是依托「工具」作为桥梁。对比 Graph 与 Tool 的核心特征，二者高度匹配：
| Graph流程编排 | Tool | |
|---|---|---|
| LLM |
高度重合的特性意味着：Graph 的运行形态完全适配工具标准，将 Graph 封装为工具逻辑直观、实现简单。绝大多数 Graph 流程都可通过工具机制接入 Agent，扩充智能体能力池，同时完整复用 Graph 的优势：复杂业务拓扑编排、丰富组件生态、流式处理 / 回调 / 断点续跑等框架治理能力。
至此，Agent 与 Graph 两条技术路线的取舍之争形成辩证统一的解决方案。
SECTION
总结
Agent 与 Graph 并非二选一的对立路线，而是两套互补的 AI 应用范式。
1. Graph 是构建稳定、确定性 AI 功能的基础：擅长编排复杂业务逻辑、数据处理流水线、API 调用链路，产出可预测、易维护的工作流。需要功能按钮、稳定后端服务时，优先选用 Graph。
2. Agent 是实现通用自主智能的核心方案：以 LLM 为中心，依托动态规划与工具调用解决开放式复杂任务。需要可对话、自主完成复杂任务的智能助手时，Agent 是核心选型。
二者最佳融合模式：将 Graph 封装为 Agent 的工具。该方案可充分发挥 Graph 强大的流程编排与生态集成能力，扩充 Agent 的工具集。一套完整 RAG 检索链路、数据分析流水线等复杂 Graph 业务，均可封装为智能体的原子能力，由 LLM 在合适时机动态调用。
SECTION
Eino 框架开发者落地指引
1. 使用 eino-compose 编写 Graph 流程，将确定性业务逻辑封装为标准化功能模块；
2. 使用 eino-adk 搭建 Agent智能体，赋予其思考规划、用户交互能力；
3. 将 Graph 封装为工具供给 Agent 调用，实现 1+1>2 的协同效果。
SECTION
代码示例
CODE
package tooladapter
import (
"context"
"encoding/json"
"fmt"
"time"
"yourorg/sdk/audit"
"yourorg/sdk/authz"
"yourorg/sdk/mask"
"yourorg/sdk/observability"
"yourorg/sdk/approval"
"yourorg/schema"
)
// SchemaProvider 编译期类型安全约束，替代运行时反射
type SchemaProvider interface {
ToolSchema() *schema.ToolInfo
Validate(input string) error
}
// SensitiveField 敏感字段标记接口，用于自动脱敏
type SensitiveField interface {
Masked() string
SensitiveFields() []string
}
// InterruptHandler 中断回调接口，对接外部审批流
type InterruptHandler func(ctx context.Context, state *InterruptState) error
// InterruptState 中断状态，包含审批所需上下文
type InterruptState struct {
NodeID string `json:"node_id"`
Input string `json:"input"`
Metadata map[string]interface{} `json:"metadata"`
CreatedAt time.Time `json:"created_at"`
ResolvedAt *time.Time `json:"resolved_at,omitempty"`
ApprovedBy string `json:"approved_by,omitempty"`
}
// GraphCompileOption 编译选项，扩展支持企业级配置
type GraphCompileOption struct {
Timeout time.Duration
MaxConcurrency int
EnableTracing bool
Classification string // 密级：内部/秘密/机密/绝密
Department string // 归属部门
DataCategory string // 数据分类：个人/财务/业务/技术
}
// InvokableGraphTool 企业级适配器
type InvokableGraphTool[I, O any] struct {
graph interface{} // 原始图引用，具体类型由使用者保证
name string
desc string
tInfo *schema.ToolInfo
classification string
department string
dataCategory string
schemaProvider SchemaProvider
maskEngine *mask.Engine
interruptHandler InterruptHandler
tracer *observability.Tracer
authzClient *authz.Client
auditLogger *audit.Logger
}
// NewEnterpriseGraphTool 企业级构造函数，注入所有基础设施依赖
func NewEnterpriseGraphTool[I, O any](
graph interface{},
name, desc string,
provider SchemaProvider,
maskEngine *mask.Engine,
interruptHandler InterruptHandler,
tracer *observability.Tracer,
authzClient *authz.Client,
auditLogger *audit.Logger,
opts ...GraphCompileOption,
) (*InvokableGraphTool[I, O], error) {
if provider == nil {
return nil, fmt.Errorf("schemaProvider is required for compliance")
}
// 从 SchemaProvider 获取编译期确定的工具信息，替代运行时反射
tInfo := provider.ToolSchema()
tInfo.Name = name
tInfo.Description = desc
// 合并编译选项
compileOpt := GraphCompileOption{
Timeout: 30 * time.Second,
MaxConcurrency: 1,
EnableTracing: true,
}
if len(opts) > 0 {
compileOpt = opts[0]
}
return &InvokableGraphTool[I, O]{
graph: graph,
name: name,
desc: desc,
tInfo: tInfo,
classification: compileOpt.Classification,
department: compileOpt.Department,
dataCategory: compileOpt.DataCategory,
schemaProvider: provider,
maskEngine: maskEngine,
interruptHandler: interruptHandler,
tracer: tracer,
authzClient: authzClient,
auditLogger: auditLogger,
}, nil
}
// InvokableRun 企业级执行入口，包含安全与合规全链路
func (g *InvokableGraphTool[I, O]) InvokableRun(ctx context.Context, input string, opts ...tool.Option) (output string, err error) {
// 1. 链路追踪埋点
span := g.tracer.StartSpan(ctx, fmt.Sprintf("graph_tool.%s", g.name))
defer span.End()
// 2. 输入校验与脱敏
if err := g.schemaProvider.Validate(input); err != nil {
g.auditLogger.Log(audit.Entry{
Action: "INPUT_VALIDATION_FAILED",
ToolName: g.name,
Input: g.maskEngine.Mask(input),
Error: err.Error(),
Timestamp: time.Now(),
})
return "", fmt.Errorf("input validation failed: %w", err)
}
// 3. 敏感数据脱敏记录审计日志（输入日志必须脱敏）
safeInput := g.maskEngine.Mask(input)
g.auditLogger.Log(audit.Entry{
Action: "TOOL_INVOKE",
ToolName: g.name,
Input: safeInput,
Department: g.department,
DataCategory: g.dataCategory,
Timestamp: time.Now(),
})
// 4. 权限校验：基于ABAC的策略裁决
decision, err := g.authzClient.Check(ctx, authz.Request{
Subject: getCallerIdentity(ctx), // 从上下文中提取用户身份
Action: "execute",
Resource: fmt.Sprintf("tool:%s", g.name),
Attribute: map[string]string{
"classification": g.classification,
"department": g.department,
"data_category": g.dataCategory,
},
})
if err != nil || !decision.Allowed {
g.auditLogger.Log(audit.Entry{
Action: "ACCESS_DENIED",
ToolName: g.name,
Input: safeInput,
Reason: decision.Reason,
})
return "", fmt.Errorf("access denied: %s", decision.Reason)
}
// 5. 反序列化输入（带类型安全检查）
var typedInput I
if err := json.Unmarshal([]byte(input), &typedInput); err != nil {
return "", fmt.Errorf("input deserialization failed: %w", err)
}
// 6. 执行图，捕获中断
result, err := g.executeWithInterrupt(ctx, typedInput)
if err != nil {
g.auditLogger.Log(audit.Entry{
Action: "EXECUTION_FAILED",
ToolName: g.name,
Input: safeInput,
Error: err.Error(),
})
return "", err
}
// 7. 输出脱敏与序列化
rawOutput, _ := json.Marshal(result)
safeOutput := g.maskEngine.Mask(string(rawOutput))
// 8. 输出审计日志
g.auditLogger.Log(audit.Entry{
Action: "TOOL_COMPLETED",
ToolName: g.name,
Input: safeInput,
Output: safeOutput,
Duration: span.Elapsed(),
})
return string(rawOutput), nil
}
// executeWithInterrupt 支持审批中断的执行逻辑
func (g *InvokableGraphTool[I, O]) executeWithInterrupt(ctx context.Context, input I) (O, error) {
// 模拟图执行过程，实际场景中替换为真实的图运行引擎
// 此处仅为示意，展示中断处理流程
var result O
// 假设图在执行过程中遇到需要人工审批的节点
interruptState := &InterruptState{
NodeID: "approval_node_001",
Input: fmt.Sprintf("%v", input),
Metadata: map[string]interface{}{"action": "transfer_funds", "amount": 50000},
CreatedAt: time.Now(),
}
// 调用外部审批处理器，阻塞等待审批结果
if err := g.interruptHandler(ctx, interruptState); err != nil {
return result, fmt.Errorf("interrupt not approved: %w", err)
}
// 审批通过后继续执行
// result = g.graph.Run(ctx, input)
return result, nil
}
// Info 返回工具元信息，包含分级分类字段
func (g *InvokableGraphTool[I, O]) Info(_ context.Context) (*schema.ToolInfo, error) {
// 扩展 ToolInfo，加入企业级属性
extendedInfo := &schema.ToolInfo{
Name: g.tInfo.Name,
Description: g.tInfo.Description,
Parameters: g.tInfo.Parameters,
Classification: g.classification,
Department: g.department,
DataCategory: g.dataCategory,
}
return extendedInfo, nil
}
// getCallerIdentity 从上下文中提取调用者身份
func getCallerIdentity(ctx context.Context) string {
// 实际实现从 JWT 或 gRPC metadata 中提取
if identity, ok := ctx.Value("x-user-id").(string); ok {
return identity
}
return "anonymous"
}
