本文为智能体编排架构的概念设计探讨，展示通用的设计思路。正文代码以概念化示例为主，便于快速理解 Agent、Task、Tool、Pipeline 的关系；文末附录提供一份完整、可运行的 Python 示例代码，可作为本地实验起点。

手写工作流脚本太累？试试让三个 AI「同事」替你搞定——研究员查资料→作者写稿→编辑审校。
你只需要定义他们的角色和协作顺序，剩下的由 Deep Agents 架构 自动调度。这是 LLM 应用开发从「单点对话」进化到「多智能体编排」的关键一步。

目录

一、核心概念：四大构建块

概念英文一句话定义示例
智能体Agent拥有独立角色与工具集的 AI 执行单元”内容创作者” Agent
任务Task输入描述 + 期望输出的工作单元”写一篇 500 字科普”
工具Tool封装搜索、计算、API 的标准化接口web_search、calculator
流水线Pipeline多 Agent 的协作流程（串行 / 并行）研究→写作→审核

Agent（智能体）

Agent 是系统的基本执行单元。每个 Agent 拥有独立的角色设定、系统提示词和工具集，相当于一个拥有特定专长的 AI 专家。

Task（任务）

Task 是分配给 Agent 的具体工作单元，包含输入描述、期望输出格式和上下文约束。一个复杂业务需求可被拆解为多个相互依赖的 Task。

Tool（工具）

Tool 是 Agent 与外部世界交互的接口，封装搜索、计算、API 调用、数据库查询等能力，以标准化方式供 Agent 调用。

Pipeline（流水线）

Pipeline 定义了多个 Agent 之间的协作流程和消息传递机制，支持串行（前一个输出作为下一个输入）和并行（多个 Agent 同时工作后汇总）两种模式。

二、快速上手：单 Agent 写作任务

在理解四个基本概念后，我们先从最简单的场景入手：用单个 Agent 完成写作。这相当于拥有一个专职写手——你只需要告诉它角色和任务。

# 以下为概念设计伪代码，展示框架设计思路

# 定义 Agent：一位专业的内容创作者
writer = Agent(
    name="Writer",
    role="内容创作者",
    system_prompt="你是一名专业的技术文章作者，擅长用清晰易懂的语言解释复杂概念。"
)

# 创建任务：明确描述和期望输出
task = Task(
    description="写一篇关于 AI 智能体的介绍文章",
    expected_output="一篇 500 字左右的科普文章",
    context={"target_audience": "技术初学者"}
)

# 运行流水线
pipeline = Pipeline(agents=[writer])
result = pipeline.run(task)

这段代码演示了最简单的单 Agent 流程：定义角色 → 分配任务 → 执行输出。

三、实战进阶：多角色协作流水线

单一 Agent 往往无法完成多步骤的复杂任务。下面的设计展示了”研究助手 → 内容作者 → 编辑审核”的典型协作模式。

场景设定

内容生产涉及三个独立环节：资料搜集 → 内容撰写 → 质量审核。

代码实现（多 Agent 串行）

# 概念设计：多 Agent 串行协作

# 定义三个不同角色的 Agent
researcher = Agent(
    name="Researcher",
    role="研究助手",
    system_prompt="你是一名擅长资料搜集和数据分析的研究员。"
)

writer = Agent(
    name="Writer",
    role="内容作者",
    system_prompt="你擅长将研究结果转化为通俗易懂的文章。"
)

editor = Agent(
    name="Editor",
    role="编辑审核",
    system_prompt="你负责审核文章的事实准确性和语言表达质量。"
)

# 构建串行流水线：前一个 Agent 的输出作为下一个的输入
content_pipeline = Pipeline(
    agents=[researcher, writer, editor],
    flow="sequential"  # 串行执行
)

# 启动整个流水线
result = content_pipeline.run(
    "撰写一篇关于 AI Agent 发展趋势的技术博客"
)

设计要点总结

  • 角色隔离：每个 Agent 只关注自己的职责范围，避免指令冲突
  • 上下文传递：中间结果自动传入下一个环节，无需手动拼接
  • 串行流程：研究结果 → 文章草稿 → 审核意见，环环相扣

四、扩展 Agent 能力：工具集成

Agent 的能力可以通过绑定自定义工具来扩展。以下展示如何为 Agent 添加搜索和计算能力：

# 概念设计：为 Agent 绑定工具

# 模拟搜索功能（实际实现时绑定具体 API）
def search_web(query):
    """调用搜索引擎 API 获取网络信息"""
    # 集成搜索服务（如 Bing Search、SerpAPI 等）
    return ["结果1", "结果2"]

search_tool = Tool(
    name="web_search",
    func=search_web,  # 绑定真实函数，内部调用搜索引擎 API
    description="输入查询字符串，返回前 5 条结果摘要"
)

# 定义计算工具
calc_tool = Tool(
    name="calculator",
    func="执行数学计算，接收数学表达式字符串",
    description="执行数学运算，支持加减乘除等基本运算"
)

# 为 Agent 绑定工具集
research_agent = Agent(
    name="ResearchAgent",
    role="研究专家",
    tools=[search_tool, calc_tool],
    system_prompt="你可以使用搜索和计算工具辅助研究工作。"
)

安全提示：实际实现中，Tool 的 func 通常绑定一个具体的 Python 函数或 API 调用。生产环境需注意：计算功能应使用安全的数学解析库（如 ast.literal_eval 或 sympy）而非 eval；搜索 API 需做好鉴权和访问控制，避免越权调用。

五、集成 Miku 平台：自动化内容生产流水线

Deep Agents 架构的设计理念与内容管理平台的内容创作流程高度契合。以 Miku 平台为例，你可以构建如下流水线：

TopicResearcher 输出《选题分析报告》 → ContentWriter 据此生成文章草稿 → ContentReviewer 生成审核意见

# 概念设计：与内容平台集成的生产流水线

# 定义内容创作团队
topic_researcher = Agent(
    name="TopicResearcher",
    role="选题研究员",
    system_prompt="分析当前技术热点，推荐有价值的选题方向。"
)

content_writer = Agent(
    name="ContentWriter",
    role="文章作者",
    system_prompt="撰写高质量的技术博客文章，注意 SEO 优化和可读性。"
)

content_reviewer = Agent(
    name="ContentReviewer",
    role="内容审核",
    system_prompt="检查文章的事实准确性、表达质量和可读性。"
)

production_pipeline = Pipeline(
    agents=[topic_researcher, content_writer, content_reviewer],
    flow="sequential"
)

通过这样的设计，选题研究、内容撰写、质量审核可以串联起来，为内容平台提供高效的 AI 辅助生产流程。

六、避坑指南：让 Pipeline 更健壮

  1. 角色分工模糊 → 每个 Agent 的 system prompt 只保留一个核心职责，避免职责重叠。
  2. 上下文超限 → 对前一个 Agent 的输出做自动摘要，只传递关键信息，控制输入长度。
  3. 无容错机制 → 设置 max_retries=2，超时则跳过当前 Agent 并输出日志。
  4. 盲目求全 → 先跑通 1 Agent + 1 Tool，再加第二个 Agent，逐步复杂化。
  5. Token 浪费 → 限制 max_output_tokens，缓存相同 prompt 的结果，降低成本。

七、总结与行动建议

从一行 prompt 到一套协作流水线，Deep Agents 让 LLM 应用开发变得像搭乐高。
现在你可以：

  1. 选一个业务环节（比如文章审核）先跑通单 Agent 流程。
  2. 逐步添加角色：助手、检查员、总结员。
  3. 挂载真实工具：搜索、数据库、通知服务。

不会写代码也没关系——本文的概念框架完全可以用自然语言描述给 LLM 辅助生成。
下一步，试试把「选题研究—写作—排版」串起来，让内容生产真正自动化。

附录 A：完整可运行 Python 示例

下面是一份不依赖第三方库的完整示例。它实现了 Tool、Task、Agent、Pipeline 四个核心对象，并演示「研究员 → 作者 → 编辑」的串行协作。

运行方式：

# 方式一：无 API Key 也能运行，会使用本地 mock LLM 输出，适合验证流程
python deep_agents_pipeline_demo.py

# 方式二：接入 OpenAI 兼容接口
export LLM_API_KEY="你的 API Key"
export LLM_MODEL="gpt-4.1-mini"
export LLM_BASE_URL="https://api.openai.com/v1"
python deep_agents_pipeline_demo.py

完整代码：

from __future__ import annotations

import ast
import json
import operator
import os
import textwrap
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class Tool:
    name: str
    description: str
    func: Callable[[str], Any]

    def run(self, tool_input: str) -> str:
        try:
            result = self.func(tool_input)
            return json.dumps(result, ensure_ascii=False, indent=2)
        except Exception as exc:  # noqa: BLE001 - demo code keeps tool errors visible.
            return json.dumps({"error": str(exc)}, ensure_ascii=False)


@dataclass
class Task:
    description: str
    expected_output: str = ""
    context: dict[str, Any] = field(default_factory=dict)


class ChatLLM:
    """Minimal OpenAI-compatible chat client with a local mock fallback."""

    def __init__(
        self,
        model: str | None = None,
        api_key: str | None = None,
        base_url: str | None = None,
        timeout: int = 60,
    ) -> None:
        self.model = model or os.getenv("LLM_MODEL", "gpt-4.1-mini")
        self.api_key = api_key or os.getenv("LLM_API_KEY", "")
        self.base_url = (base_url or os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")).rstrip("/")
        self.timeout = timeout

    def complete(self, messages: list[dict[str, str]]) -> str:
        if not self.api_key:
            return self._mock_complete(messages)

        url = f"{self.base_url}/chat/completions"
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.35,
        }
        request = urllib.request.Request(
            url=url,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                data = json.loads(response.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"LLM request failed: HTTP {exc.code} {body}") from exc

        return data["choices"][0]["message"]["content"].strip()

    def _mock_complete(self, messages: list[dict[str, str]]) -> str:
        system = messages[0]["content"]
        user = messages[-1]["content"]
        if "编辑" in system or "审核" in system:
            return (
                "审核意见：结构清晰，适合初学者。建议补充工具调用的安全边界，"
                "并避免把示例说成已经具备生产级可靠性。"
            )
        if "作者" in system or "写作" in system:
            return (
                "文章草稿：AI Agent 正从单轮问答走向可编排工作流。"
                "在内容生产场景中，可以让研究员收集资料、作者生成草稿、"
                "编辑审核表达与事实。这样的流水线能降低单个提示词的复杂度，"
                "也方便定位错误和逐步扩展工具能力。"
            )
        if "研究员" in system or "研究助手" in system:
            return (
                "研究摘要：AI Agent 的关键趋势包括工具调用、多智能体协作、"
                "长上下文治理、人工审批节点和可观测性。建议文章重点解释"
                "为什么要把复杂任务拆成角色清晰的流水线。"
            )
        return f"Mock response for: {user[:160]}"


class Agent:
    def __init__(
        self,
        name: str,
        role: str,
        system_prompt: str,
        tools: list[Tool] | None = None,
        llm: ChatLLM | None = None,
    ) -> None:
        self.name = name
        self.role = role
        self.system_prompt = system_prompt
        self.tools = {tool.name: tool for tool in tools or []}
        self.llm = llm or ChatLLM()

    def run(self, task: Task, previous_output: str = "") -> str:
        tool_results = self._run_tools(task.context.get("tool_inputs", {}))
        prompt = self._build_prompt(task, previous_output, tool_results)
        return self.llm.complete(
            [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": prompt},
            ]
        )

    def _run_tools(self, tool_inputs: dict[str, str]) -> dict[str, str]:
        results: dict[str, str] = {}
        for tool_name, tool_input in tool_inputs.items():
            tool = self.tools.get(tool_name)
            if tool is None:
                continue
            results[tool_name] = tool.run(str(tool_input))
        return results

    def _build_prompt(self, task: Task, previous_output: str, tool_results: dict[str, str]) -> str:
        return textwrap.dedent(
            f"""
            你的名字：{self.name}
            你的角色：{self.role}

            当前任务：
            {task.description}

            期望输出：
            {task.expected_output or "请给出清晰、可执行的结果。"}

            上游输出：
            {previous_output or "无"}

            任务上下文：
            {json.dumps(task.context, ensure_ascii=False, indent=2)}

            工具结果：
            {json.dumps(tool_results, ensure_ascii=False, indent=2)}
            """
        ).strip()


class Pipeline:
    def __init__(self, agents: list[Agent], flow: str = "sequential") -> None:
        if flow != "sequential":
            raise ValueError("This demo implements only sequential flow.")
        if not agents:
            raise ValueError("Pipeline requires at least one agent.")
        self.agents = agents
        self.flow = flow

    def run(self, task: Task) -> dict[str, Any]:
        previous_output = ""
        trace: list[dict[str, str]] = []

        for agent in self.agents:
            agent_task = Task(
                description=task.description if not previous_output else previous_output,
                expected_output=task.expected_output,
                context={**task.context, "trace": trace},
            )
            output = agent.run(agent_task, previous_output=previous_output)
            trace.append({"agent": agent.name, "role": agent.role, "output": output})
            previous_output = output

        return {"final_output": previous_output, "trace": trace}


def search_web(query: str) -> dict[str, Any]:
    """Demo search tool. Replace this function with Bing, SerpAPI, Tavily, or your own API."""
    return {
        "query": query,
        "results": [
            {
                "title": "Agent workflow trend",
                "summary": "Teams are moving from single prompts to tool-using, observable workflows.",
            },
            {
                "title": "Multi-agent orchestration",
                "summary": "Role separation helps reduce prompt conflicts and makes review easier.",
            },
        ],
    }


def safe_calculator(expression: str) -> float:
    """Small arithmetic evaluator. It intentionally rejects names, calls, and attributes."""
    allowed_binary_ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Mod: operator.mod,
        ast.Pow: operator.pow,
    }
    allowed_unary_ops = {
        ast.UAdd: operator.pos,
        ast.USub: operator.neg,
    }

    def evaluate(node: ast.AST) -> float:
        if isinstance(node, ast.Expression):
            return evaluate(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return float(node.value)
        if isinstance(node, ast.BinOp) and type(node.op) in allowed_binary_ops:
            return allowed_binary_ops[type(node.op)](evaluate(node.left), evaluate(node.right))
        if isinstance(node, ast.UnaryOp) and type(node.op) in allowed_unary_ops:
            return allowed_unary_ops[type(node.op)](evaluate(node.operand))
        raise ValueError(f"Unsupported expression: {ast.dump(node)}")

    tree = ast.parse(expression, mode="eval")
    return evaluate(tree)


def build_content_pipeline() -> Pipeline:
    llm = ChatLLM()
    search_tool = Tool(
        name="web_search",
        description="输入查询字符串，返回资料摘要。",
        func=search_web,
    )
    calculator_tool = Tool(
        name="calculator",
        description="输入数学表达式，返回计算结果。",
        func=safe_calculator,
    )

    researcher = Agent(
        name="Researcher",
        role="研究助手",
        system_prompt="你是一名研究员，负责收集资料、提炼事实、指出不确定信息。",
        tools=[search_tool],
        llm=llm,
    )
    writer = Agent(
        name="Writer",
        role="内容作者",
        system_prompt="你是一名技术文章作者，负责把研究摘要改写成清晰、有结构的中文草稿。",
        llm=llm,
    )
    editor = Agent(
        name="Editor",
        role="编辑审核",
        system_prompt="你是一名编辑，负责检查事实边界、表达质量和可读性，并给出修改建议。",
        tools=[calculator_tool],
        llm=llm,
    )
    return Pipeline([researcher, writer, editor])


def main() -> None:
    task = Task(
        description="撰写一篇关于 AI Agent 发展趋势的技术博客。",
        expected_output="先给研究摘要，再给文章草稿，最后给审核意见。",
        context={
            "target_audience": "技术初学者",
            "style": "清晰、具体、避免夸大",
            "tool_inputs": {
                "web_search": "AI Agent 发展趋势 多智能体 工具调用 可观测性",
                "calculator": "128 * 3",
            },
        },
    )
    result = build_content_pipeline().run(task)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

这份代码故意保持克制：它没有把工具调用、记忆、审批、重试、并行执行一次性塞满，而是先把「角色隔离 + 上下文传递 + 工具结果注入 + 串行编排」跑通。真正接入生产系统时，可以在这个基础上继续增加日志、重试、人工审核、结构化输出校验和持久化存储。

附录 B：真正使用 deepagents 框架

上一份代码是教学版，方便你看清楚 Agent 与 Pipeline 的底层关系。下面这份代码则真正使用 deepagents 框架，通过 create_deep_agent 创建主 Agent，并注册 researcher、writer、editor 三个子 Agent，由主 Agent 通过任务分派完成协作。

安装依赖：

pip install deepagents langchain-openai

环境变量：

export LLM_API_KEY="你的 API Key"
export LLM_MODEL="gpt-4.1-mini"
export LLM_BASE_URL="https://api.openai.com/v1"

完整代码：

from __future__ import annotations

import json
import os
from typing import Any

from deepagents import (
    FilesystemPermission,
    GeneralPurposeSubagentProfile,
    HarnessProfile,
    create_deep_agent,
    register_harness_profile,
)
from langchain_openai import ChatOpenAI


SYSTEM_PROMPT = """你是一个内容生产流水线负责人。

目标：
1. 先把任务交给 researcher，产出研究摘要。
2. 再把研究摘要交给 writer，产出文章草稿。
3. 再把文章草稿交给 editor，产出审核意见。
4. 最后调用 save_pipeline_result 保存结构化结果。

要求：
- 默认使用中文。
- 不要夸大未验证事实。
- 每个角色只做自己的工作。
- 最终回复只返回标题、摘要、审核结论和保存结果。
"""


RESEARCHER_PROMPT = """你是 researcher。
你负责资料搜集、事实提炼和不确定性标注。
你可以调用 search_web 获取演示资料。
输出应包含：关键趋势、可写角度、风险提醒。"""


WRITER_PROMPT = """你是 writer。
你负责把研究摘要写成清晰的中文技术博客草稿。
要求结构清楚、措辞谨慎、适合技术初学者阅读。"""


EDITOR_PROMPT = """你是 editor。
你负责审核文章的事实边界、表达质量、结构完整性和可读性。
输出 PASS 或 REVISE，并给出具体理由。"""


def build_model() -> ChatOpenAI:
    api_key = os.getenv("LLM_API_KEY")
    if not api_key:
        raise RuntimeError("Missing LLM_API_KEY")

    return ChatOpenAI(
        model=os.getenv("LLM_MODEL", "gpt-4.1-mini"),
        api_key=api_key,
        base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
        temperature=0.35,
    )


def search_web(query: str) -> dict[str, Any]:
    """演示搜索工具。生产环境可替换成 Bing、Tavily、SerpAPI 或内部检索服务。"""
    return {
        "query": query,
        "results": [
            {
                "title": "Tool-using agents",
                "summary": "Agent 应用通常会把 LLM、工具调用、状态管理和人工审批组合起来。",
            },
            {
                "title": "Multi-agent workflow",
                "summary": "多角色拆分有助于降低单个提示词复杂度，并让审核环节更清晰。",
            },
            {
                "title": "Production guardrails",
                "summary": "生产级 Agent 需要关注权限控制、日志、重试、成本和可观测性。",
            },
        ],
    }


def save_pipeline_result(title: str, research: str, draft: str, review: str) -> dict[str, Any]:
    """演示保存工具。这里不写文件，只返回结构化结果；接入平台时可替换成数据库或 CMS API。"""
    return {
        "ok": True,
        "title": title,
        "chars": {
            "research": len(research),
            "draft": len(draft),
            "review": len(review),
        },
        "preview": draft[:180],
    }


def deny_filesystem() -> list[FilesystemPermission]:
    return [FilesystemPermission(operations=["read", "write"], paths=["/**"], mode="deny")]


def register_profile() -> None:
    """收紧默认工具面：隐藏通用文件和 shell 类工具，只保留显式传入的业务工具。"""
    register_harness_profile(
        "openai",
        HarnessProfile(
            excluded_tools=frozenset(
                {
                    "write_todos",
                    "ls",
                    "read_file",
                    "write_file",
                    "edit_file",
                    "glob",
                    "grep",
                    "execute",
                }
            ),
            general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False),
        ),
    )


def build_agent():
    model = build_model()
    permissions = deny_filesystem()
    register_profile()

    return create_deep_agent(
        model=model,
        system_prompt=SYSTEM_PROMPT,
        tools=[save_pipeline_result],
        permissions=permissions,
        subagents=[
            {
                "name": "researcher",
                "description": "收集资料、提炼趋势、标注不确定信息。",
                "system_prompt": RESEARCHER_PROMPT,
                "model": model,
                "tools": [search_web],
                "permissions": permissions,
            },
            {
                "name": "writer",
                "description": "根据研究摘要撰写中文技术博客草稿。",
                "system_prompt": WRITER_PROMPT,
                "model": model,
                "permissions": permissions,
            },
            {
                "name": "editor",
                "description": "审核文章事实边界、表达质量和结构完整性。",
                "system_prompt": EDITOR_PROMPT,
                "model": model,
                "permissions": permissions,
            },
        ],
    )


def main() -> None:
    agent = build_agent()
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "请用多角色协作流水线写一篇关于 AI Agent 发展趋势的中文技术博客。"
                        "请先研究，再写作，再审核，最后保存结构化结果。"
                    ),
                }
            ]
        }
    )
    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))


if __name__ == "__main__":
    main()

这份代码里的多角色协作不是手写 for agent in agents，而是交给 deepagents 的主 Agent 和子 Agent 机制来完成。主 Agent 负责拆解任务和调度，researcher、writer、editor 分别拥有自己的系统提示词、工具和权限边界；这也更接近真实项目里把内容生产、SEO、审核、发布拆成多个专职智能体的做法。