使用 CrewAI 实现 Chain-of-Thought（思维链）与 Chain-of-Action（行动链）的完整步骤说明（中文版）

本文由chatGPT 生成，提示词: Please use crewAI to implement chain of thoughts, and show how to implement every chain of action step by step.

英文输出后，再由chatGPT生成中文

下面我将一步一步展示如何使用 CrewAI 实现 Chain-of-Thought（CoT，思维链），并清晰地展示每一个行动步骤是如何被规划、执行和记录的。

内容包括：

1. 概念概览：CrewAI 提供了什么，以及 CoT 如何映射到 CrewAI
2. 一个可运行的实现模式（安装 → 项目结构 → agents.yaml → Python 代码）
3. Prompt 模板、Planner（规划器）使用方式，以及如何通过 Callback 实时输出每一步思考
4. 扩展建议（Tree-of-Thought、验证机制、成本控制等）

1）概念速览：Chain-of-Thought 在 CrewAI 中是如何实现的

• CrewAI 通过 Agent（智能体）→ Crew（团队）→ Flow / Process（流程） 来组织多智能体系统
• 每一次 Agent 的运行，本质上都包含：
  • 思考 / 规划（Thought / Plan）
  • 工具调用（Action）
  • 结果输出（Output）
• CrewAI 把这些作为一等公民来设计（规划、遥测、可观测性）

与 CoT 直接相关的两个核心机制

① Planning（规划）

• 在 Crew 中设置 planning=True
• CrewAI 会自动调用一个 AgentPlanner
• Planner 会先生成一个分步骤执行计划（编号列表）
• 该计划会被注入到后续 Task 中

② Telemetry / Observability（遥测 / 可观测性）

• CrewAI 会记录：
  • Planner 输出
  • Agent 的思考过程
  • 工具调用与返回结果
  • 记忆（Memory）的变化
• 你可以通过 Callback 或 UI 实时查看完整“思维链 + 行动链”

2）我们要构建什么（整体目标）

我们将构建一个最小但完整的 CrewAI 项目，实现：

1. 接收一个任务输入（例如：总结隐私政策并给出风险与缓解措施）
2. Planner 先生成一个清晰的步骤计划
3. 每个步骤交给 Analyst Agent 执行：
   • 输出该步骤的 THOUGHT（思考）
   • 如有需要，调用工具（ACTION）

1. 实时输出或记录：

1. • Planner 的计划
   • 每一步的 Thought / Action / Output
   • 工具调用结果

1. 最终生成综合输出

示例执行轨迹：

• Planner：
  1. 阅读文档
  2. 提取关键条款
  3. 识别风险
  4. 给出缓解建议
• Step 1：
  1. Thought：我将扫描标题定位关键条款
  2. Action：document_search()
  3. Output：提取的条款列表

3）环境安装与初始化

# 1）安装 uv（CrewAI 官方推荐）

curl -LsSf https://astral.sh/uv/install.sh | sh

# 2）创建虚拟环境并安装 crewai

python -m venv .venv

source .venv/bin/activate

pip install crewai

# 3）可选：克隆官方示例仓库

git clone https://github.com/crewAIInc/crewAI-examples.git

4）创建 CrewAI 项目骨架

crewai create crew cot-demo

cd cot-demo

该命令会生成一个标准的 CrewAI 项目结构，包括：

• config/agents.yaml
• src/main.py
• 工具、回调等扩展目录

5）Agent 与 Planner 配置（agents.yaml）

下面的配置重点在于：

• 启用 Planner
• 强制 Agent 输出 结构化的思考与行动

agents:

  - id: planner

    role: planner

    model: gpt-4o-mini

    prompt_template: |

      你是一个任务规划智能体。

      请根据输入任务，生成一个简洁、编号的执行计划。

      输出格式：

      PLAN:

      1) 步骤一

      2) 步骤二

      ...



  - id: analyst

    role: analyst

    model: gpt-4o-mini

    prompt_template: |

      你是分析型智能体。

      对于分配给你的每一个步骤，请严格使用以下结构输出：



      THOUGHT:

      - 用简短要点描述你的推理过程



      ACTION:

      - 如果需要调用工具，使用一行：

        TOOL:search_docs(query="...")

      - 如果不需要，写 ACTION: none



      OUTPUT:

      - 本步骤的简要结果

这个结构非常关键，它让你可以程序化解析思维链与行动链

6）在 Python 中创建 Crew 并启用规划

from crewai import Crew, Agent, Task, Process, Runner

planner = Agent.from_config("planner")

analyst = Agent.from_config("analyst")

tasks = [

    Task(id="step1", agent="analyst", description="{plan_step_1}"),

    Task(id="step2", agent="analyst", description="{plan_step_2}"),

]

crew = Crew(

    agents=[planner, analyst],

    tasks=tasks,

    process=Process.sequential,

    planning=True

)

runner = Runner(crew=crew

inputs = {

    "topic": "总结某隐私政策并列出主要风险与缓解措施",

    "document_url": "file://..."

}

runner.kickoff(inputs=inputs)

当 planning=True 时：

1. Planner 会先运行
2. 生成的步骤会自动填充到 {plan_step_n}
3. Agent 按步骤顺序执行

7）解析并输出 THOUGHT / ACTION / OUTPUT（Callback）

from crewai.callbacks import BaseCallbackHandler

class ThoughtStreamCallback(BaseCallbackHandler):

    def on_agent_response(self, event):

        text = event.response

        thought, action, output = "", "", ""

        if "THOUGHT:" in text:

            try:

                thought = text.split("THOUGHT:")[1].split("ACTION:")[0].strip()

                action = text.split("ACTION:")[1].split("OUTPUT:")[0].strip()

                output = text.split("OUTPUT:")[1].strip()

            except:

                thought = text

        print(f"[{event.agent_id}] 思考:\n{thought}\n")

        print(f"[{event.agent_id}] 行动:\n{action}\n")

        print(f"[{event.agent_id}] 输出:\n{output}\n")

在 Runner 中注册：

runner = Runner(

    crew=crew,

    callbacks=[ThoughtStreamCallback()]

)

8）工具调用（Action）的实现示例

from crewai.tools import Too

def search_docs(query, top_k=3):

    return ["示例结果1", "示例结果2"]

search_tool = Tool(

    name="search_docs",

    func=search_docs

)

Agent 输出：

ACTION:

TOOL:search_docs(query="数据泄露", top_k=3)

CrewAI 会自动执行该工具并记录返回结果。

9）Token 级实时思维链输出（进阶）

你可以通过 Streaming Callback 实现：

• Token 级别的 Thought 输出
• 实时 UI（Streamlit / WebSocket）
• 类似“AI 正在思考”的可视化体验

实现方式：

• 监听 on_stream_token
• 根据 THOUGHT: / ACTION: 标志切换缓冲区

10）一次完整运行的执行流程

1. runner.kickoff()
2. Planner 生成计划
3. 每个 Task：
   • Agent 思考（THOUGHT）
   • 工具调用（ACTION）
   • 输出结果（OUTPUT）

1. 最终 Synthesizer 汇总
2. 全流程被记录、可回放、可审计

11）实用建议（非常重要）

• 强制结构化输出（JSON 最佳）
• Planner 使用 few-shot 提示，避免步骤不稳定
• 对工具调用做参数校验（安全）
• 注意 CoT 会增加 token 成本，仅在必要时启用

12）可扩展模式

• Tree-of-Thought：多个 Planner 并行生成方案，投票或评估
• Verifier Agent：对每一步结果进行事实校验
• 可观测 UI：回放整个思维链与行动链

13）快速检查清单（Checklist）

• 安装 crewai
• Crew 启用 planning=True
• Agent 强制结构化 CoT 输出
• 工具映射与校验
• Callback 记录思维与行动
• 可选：验证 Agent

14）下一步我可以帮你做什么？

如果你愿意，我可以直接帮你：

• ✅ 生成一个 可直接运行的完整仓库
• ✅ 写一个 Streamlit 思维链可视化 UI
• ✅ 帮你把 CoT 改成 JSON Schema + 校验
• ✅ 设计 企业级多 Agent 架构

只需要告诉我你想要哪一个 👍