DAG 声明式编排才是多智能体的正确打开方式harness-sdk 工作流实战【免费下载链接】harness-sdkBuild an agent harness and control it end-to-end. Open-source SDK for production AI agents in Python TypeScript - any model, any cloud.项目地址: https://gitcode.com/GitHub_Trending/sdkpython13/harness-sdk多智能体系统正在快速从demo 阶段走向生产环境但一个残酷的事实是绝大多数团队的第一版多智能体应用都是用手工调度拼出来的。所谓手工调度就是在业务代码里写一串agent_a(...)→ 把结果拼成字符串 →agent_b(...)的胶水逻辑。当智能体只有两三个、任务链路是纯串行时这套写法勉强能跑一旦智能体数量上到五个、出现分支汇合diamond、并行扇出、循环迭代手工调度的代码会迅速腐烂成一张无法维护的意大利面条图。这正是 harness-sdk一个同时提供 Python 与 TypeScript SDK 的开源 Agent Harness 框架给出的答案所在用 Graph图抽象把智能体之间的依赖关系声明出来让运行时替你完成调度。本文不打算复述 API 文档而是沿着一条真实的问题链展开手工调度为什么在多智能体场景必然崩盘 → DAG 声明式编排的核心概念与懒调度机制 → 一个可复用的两步依赖工作流示例。全文代码均取自仓库源码你可以直接对照验证。为什么手工调度在多智能体场景必崩先看一段典型的手工调度代码它就在仓库的示例目录里site/docs/examples/python/agents_workflow.pyresearcher_response researcher_agent( fResearch: {user_input}. Use your available tools to gather information from reliable sources. ) research_findings str(researcher_response) analyst_response analyst_agent( fAnalyze these findings about {user_input}:\n\n{research_findings}, ) analysis str(analyst_response) final_report writer_agent( fCreate a report on {user_input} based on this analysis:\n\n{analysis} )这段代码逻辑清晰三个智能体依次执行结果逐级传递。但请留意它暴露出的四个结构性缺陷这些缺陷会在智能体数量增长后指数级放大1. 数据传递是隐式契约而非显式依赖。research_findings通过 f-string 手工拼接进下一次调用。一旦 Researcher 的输出格式变化、或某个环节返回空字符串错误会在下游智能体那里以幻觉式推理的方式爆发而调用链上没有任何一层知道这个契约的存在。2. 执行顺序被硬编码在调用序列里。如果要让 Research 和 Fact-Check 并行执行再汇合到 Report手工调度只能引入asyncio.gather 手动收集结果 手动拼接上下文并行度每提升一点胶水代码就翻一倍。3. 状态与可观测性缺失。手工调度没有执行顺序记录、没有每节点耗时/Token 统计、没有从断点恢复的能力。生产环境里一个智能体超时挂起你只能靠日志大海捞针。4. 失败处理没有统一语义。每个调用点都要单独 try/except且异常后下游是否继续完全取决于你当时怎么想。社区里围绕 harness-sdk 的实战文章反复提到Harness 与 Agent 的本质区别Agent 是单点执行单元Harness 是编排系统。手工调度正是把编排逻辑错误地塞进了业务代码里。而声明式编排的核心洞察是把依赖关系提升为一等公民让调度器替开发者做决策。DAG 声明式编排的核心概念与懒调度机制harness-sdk 的 Graph 编排实现在 strands-py/src/strands/multiagent/graph.pyPython 端与 strands-ts/src/multiagent/graph.tsTypeScript 端。官方文档对其定位很明确见 site/src/content/docs/user-guide/sdk/multi-agent/graph.mdxA Graph gives you deterministic control over how a set of agents runs. You define the nodes (agents, custom nodes, or nested multi-agent systems) and the edges between them. Each node runs according to its edge dependencies, and its output passes as input to the nodes that depend on it.一句话翻译Graph 让你用节点 边声明智能体之间的依赖运行时根据边的依赖关系决定执行顺序并把上游输出自动组装成下游输入。它支持无环DAG与有环两种拓扑因此既能做严格依赖的流水线也能做带退出条件的反馈循环。三个核心构件GraphNode、GraphEdge、GraphBuilder在 strands-py/src/strands/multiagent/graph.py 中三者分工清晰GraphNode一个节点node_id唯一标识executor可以是单个Agent、A2AAgent远程智能体或任意MultiAgentBase另一个 Graph 或 Swarm即嵌套编排dependencies记录它依赖哪些节点。GraphEdge一条有向边from_node→to_node可选condition函数决定这条边是否可穿越条件路由。GraphBuilder建造者模式add_node()注册节点、add_edge()声明依赖、build()时做校验并生成Graph实例。注意build()中的两个关键行为见 strands-py/src/strands/multiagent/graph.py 的GraphBuilder.build与_validate_graph# 未显式指定入口点时自动检测无入边的节点即入口点 if not self.entry_points: self.entry_points {node for node_id, node in self.nodes.items() if not node.dependencies} # 若既没设 max_node_executions 也没设 execution_timeout # 且图中存在环则告警可能无限执行 if self._max_node_executions is None and self._execution_timeout is None: logger.warning(Graph without execution limits may run indefinitely if cycles exist)懒调度ready-driven机制依赖满足才执行Graph 的调度核心是批次就绪驱动而非预先规划全部顺序。主循环在Graph._execute_graph中strands-py/src/strands/multiagent/graph.pyready_nodes list(self.entry_points) # 初始批次入口节点 while ready_nodes: current_batch ready_nodes.copy() ready_nodes.clear() # 并行执行当前批次实时合并事件流 async for event in self._execute_nodes_parallel(current_batch, invocation_state): yield event # 批次结束后只从本批次节点的出边里找新就绪节点 newly_ready self._find_newly_ready_nodes(current_batch) ready_nodes.extend(newly_ready)值得注意的细节是_find_newly_ready_nodes的注释strands-py/src/strands/multiagent/graph.pyOnly evaluates destination nodes of outbound edges from the completed batch, instead of iterating over all nodes in the graph.即调度器不会在每次循环里扫描整张图而是只检查刚完成批次的出边指向的候选节点再逐个用_is_node_ready_with_conditions验证入边条件是否满足。这是一种典型的懒评估节点在依赖未就绪前完全不参与调度依赖一满足立即进入下一批次。它天然支持并行扇出同一批次的多个节点并发执行也让条件路由变得廉价——边条件只有在源节点完成、目标节点成为候选时才被求值。这种机制也带来一个语言层面的行为差异文档在 site/src/content/docs/user-guide/sdk/multi-agent/graph.mdx 的 SDK Differences 一节明确标注依赖解析Python 默认 OR 语义任一入边完成即触发目标TypeScript 默认 AND 语义所有入边源完成才触发对 diamond/join 模式更直观。调度粒度Python 按离散批次执行等整批完成再调度下一批TypeScript 按节点逐个启动受maxConcurrency约束快节点不必等慢兄弟。状态语义Python 节点默认跨重访累积状态除非reset_on_revisitTypeScript 节点默认无状态重访时干净重跑可用preserveContext: true选择累积。这个差异恰恰说明声明式编排把调度策略从业务代码里抽离出来变成可配置的运行时行为——你要 OR 还是要 AND要批次还是逐个是调一行配置的事而不是重写一遍调度逻辑。边界条件条件边、执行限额与可观测性除了基础拓扑Graph 还提供了把确定性焊进编排的手段条件边add_edge(research, analysis, conditiononly_if_research_successful)条件函数接收GraphStatePython 端还支持接收invocation_state做运行时路由如按用户角色、功能开关分流返回布尔值决定是否穿越。执行限额set_max_node_executions()限制总执行次数防环死循环、set_execution_timeout()整体超时、set_node_timeout()单节点超时。结果对象GraphResult提供execution_order真实执行顺序、completed_nodes/failed_nodes、execution_time、accumulated_usageToken 汇总——调度不再是黑盒。输入自动组装_build_node_inputstrands-py/src/strands/multiagent/graph.py会把原始任务 所有已满足依赖节点的输出按固定格式拼接成下游节点的输入省去手工传参Original Task: Analyze the quarterly sales data and create a summary report Inputs from previous nodes: From data_processor: - Agent: Sales data processed successfully. Found 1,247 transactions totaling $89,432. From validator: - Agent: Data validation complete. All records verified, no anomalies detected.一个可复用的两步依赖工作流示例现在把上述概念落成一个可运行的工程示例。我们构建一个研究 → 汇合分析的最小 DAGresearch完成后并行扇出analysis与fact_check两者都完成后汇合到report。完整代码可对照文档 site/src/content/docs/user-guide/sdk/multi-agent/graph.mdx 的 Creating a Graph 一节from strands import Agent from strands.multiagent import GraphBuilder researcher Agent(nameresearcher, system_promptYou are a research specialist...) analyst Agent(nameanalyst, system_promptYou are a data analysis specialist...) fact_checker Agent(namefact_checker, system_promptYou are a fact checking specialist...) report_writer Agent(namereport_writer, system_promptYou are a report writing specialist...) builder GraphBuilder() # 声明节点 builder.add_node(researcher, research) builder.add_node(analyst, analysis) builder.add_node(fact_checker, fact_check) builder.add_node(report_writer, report) # 声明依赖边DAG 拓扑 builder.add_edge(research, analysis) builder.add_edge(research, fact_check) builder.add_edge(analysis, report) builder.add_edge(fact_check, report) # 显式指定入口点省略时会自动检测无入边节点 builder.set_entry_point(research) # 为安全兜底整体 10 分钟超时 builder.set_execution_timeout(600) graph builder.build() result graph(Research the impact of AI on healthcare and create a comprehensive report) print(fStatus: {result.status}) print(fExecution order: {[node.node_id for node in result.execution_order]})这段代码与手工调度版本的差异是本质性的第一拓扑与业务解耦。依赖关系以数据边集合存在而不是藏在调用栈里。想调整流程改add_edge即可调用代码零改动。第二并行是免费的。research完成后analysis与fact_check同批次并行执行Python 端由_execute_nodes_parallel用共享 asyncio 队列实时合并事件流实现你不需要手写gather。第三汇合点有明确的依赖语义。report节点在 Python 端默认 OR 语义下可能被任一上游触发如果你需要严格的双输入齐备语义文档给出了all_dependencies_complete工厂函数配合条件边实现 AND 等待site/src/content/docs/user-guide/sdk/multi-agent/graph.mdx 的 Waiting for All Dependencies 一节。TypeScript 端则天然就是 AND 语义。第四结果可审计。result.execution_order记录了真实执行序列result.accumulated_usage汇总全部 Token 消耗配合stream_async可以实时订阅multiagent_node_start/stop事件做细粒度观测见 strands-py/src/strands/multiagent/graph.py 的事件流实现。进阶把确定性逻辑焊进编排如果想让工作流里掺杂确定性业务逻辑数据校验、格式化、规则判断Graph 支持自定义节点Python 端继承MultiAgentBase实现invoke_async即可把一个纯 Python 函数变成图节点site/src/content/docs/user-guide/sdk/multi-agent/graph.mdx 的 Custom Node Types 一节。其价值文档点得很透Skip LLM calls for deterministic operations——纯逻辑不该浪费 Token也不该承担 LLM 的不确定性。这正是AI 创造 确定性控制混合工作流的标准做法。结语编排层是独立于模型层与业务层的一等公民把手工调度换成 DAG 声明式编排表面上只是换了一种写法实质是把谁先跑、谁等谁、结果怎么传从业务代码中整体剥离交给一个具备依赖解析、条件路由、并行调度、执行限额与状态快照的运行时。社区关于 harness-sdk 的实战文章反复强调编排层应严格区分于模型层与业务层本文的 Graph 实现正是这一原则的落地模型层负责推理能力业务层负责领域逻辑编排层只负责一件事——按照你声明的依赖正确、可控、可观测地把智能体组织起来。如果你正在构建的系统中智能体数量即将突破三五个、流程开始出现分支汇合或迭代循环那么是时候停止在业务代码里手写调度把控制权交给 DAG 了。仓库根目录的 AGENTS.md 和 strands-py/AGENTS.md 还记录了更多关于本仓库开发协作与架构约定的内容可作为继续深入编排层源码的入口。【免费下载链接】harness-sdkBuild an agent harness and control it end-to-end. Open-source SDK for production AI agents in Python TypeScript - any model, any cloud.项目地址: https://gitcode.com/GitHub_Trending/sdkpython13/harness-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
