Mastra vs LangChain: Building an AI Agent Pipeline and Analyzing the Data

Topic
  ↓
1. RESEARCH   (Tavily 网络搜索，5 个结果，附相关性评分)
2. ANALYSIS   (提取 5 个关键发现，3 个主题，1 个核心论点)
3. WRITE      (撰写结构化的约 400 字报告)
4. CRITIC     (对草稿进行评分，提供维度级别的反馈)
5. LOOP       (如果评分低于 7 分则进行修订，通过或达到 3 次迭代则输出)

mastra-vs-langchain/
├── packages/
│   ├── mastra-pipeline/          # Mastra 实现
│   ├── langchain-pipeline/       # LangChain/LangGraph 实现
│   ├── web/                      # Next.js 16 App Router 仪表盘
│   └── shared/                   # 共享的 TypeScript 类型
├── convex/                       # 实时后端
└── package.json                  # 工作区根目录

// packages/shared/src/types.ts
export interface PipelineCallbacks {
  onPipelineStart: () => Promise<string>;
  onPipelineComplete: (id: string, data: PipelineCompleteData) => Promise<void>;
  onPipelineError: (id: string, error: string) => Promise<void>;
  step: {
    onStepStart: (stepName: string, iteration: number, input: string) => Promise<string>;
    onStepComplete: (stepId: string, data: StepCompleteData) => Promise<void>;
    onStepError: (stepId: string, error: string) => Promise<void>;
  };
}

// packages/mastra-pipeline/src/tools/search.ts
import { createTool } from "@mastra/core/tools";
import { z } from "zod";
import { tavily } from "@tavily/core";

const client = tavily({ apiKey: process.env.TAVILY_API_KEY! });

export let lastTavilyCapture: { query: string; results: any[] } = {
  query: "",
  results: [],
};

export function resetTavilyCapture() {
  lastTavilyCapture = { query: "", results: [] };
}

export const searchTool = createTool({
  id: "web-search",
  description: "在某个主题上搜索网络信息",
  inputSchema: z.object({ query: z.string() }),
  execute: async ({ query }) => {
    lastTavilyCapture = { query, results: [] };
    const results = await client.search(query, {
      maxResults: 5,
      searchDepth: "basic",
    });
    lastTavilyCapture.results = results.results;
    return { results: results.results };
  },
});

// packages/mastra-pipeline/src/agents/researcher.ts
export const researcherAgent = new Agent({
  name: "Researcher",
  id: "researcher",           // v1.41 中必需 - 容易被遗漏
  instructions: `你是一个研究代理。当给出一个主题时，使用 
  web-search 工具查找 5 个相关结果。以格式化字符串的形式返回所有原始搜索 
  结果，包括标题、URL 和内容片段。`,
  model: anthropic("claude-haiku-4-5"),
  tools: { searchTool },
});

// packages/mastra-pipeline/src/agents/writer.ts
export const writerAgent = new Agent({
  name: "Writer",
  id: "writer",
  instructions: `你是一个为技术受众写作的研究分析师。

严格要求：
- 开头句子必须陈述研究中的具体发现。
  永远不要以 "X 越来越重要" 开头。
- 每个段落只做一个论点。首先陈述它。
  用具体的证据支持它。
- 提到具体的工具、框架、公司、数字和日期。
- 结论必须做出具体的建议或预测。
  它不能重复引言。
- 目标长度：350-450 字。

禁止使用的短语：
"it is important to note", "it is worth noting",
"organizations must consider", "in conclusion", "in summary",
"as we look to the future", "rapidly evolving landscape",
任何如果替换主题后句子同样成立的句子`,
  model: anthropic("claude-haiku-4-5"),
});

const writeCriticStep = createStep({
  id: "write-critic",
  inputSchema: z.object({
    topic: z.string(),
    research: z.string(),
    analysis: z.string(),
    keyFindings: z.array(z.string()),
    mainThemes: z.array(z.string()),
    centralArgument: z.string(),
    draft: z.string().optional(),       // 在第一次迭代后填充
    score: z.number().optional(),       // 在第一次迭代后填充
    feedback: z.string().optional(),    // 在第一次迭代后填充
    iterations: z.number().optional(),
  }),
  outputSchema: z.object({
    // ... 所有输入字段加上 draft, score, feedback, iterations
  }),
  execute: async ({ inputData }) => {
    const iteration = (inputData.iterations ?? 0) + 1;

    // 写作阶段
    let writerPrompt = `Topic: "\${inputData.topic}"\n\nResearch:\n\${inputData.research}\n\nAnalysis:\n\${inputData.analysis}`;
    if (inputData.feedback && inputData.draft) {
      // 在修改过程中，作者可以看到其之前的尝试和具体的反馈
      writerPrompt += `\n\nPrevious draft:\n\${inputData.draft}\n\nFeedback:\n\${inputData.feedback}`;
    }

    const writeStepId = await callbacks.step.onStepStart("write", iteration, writerPrompt.slice(0, 500));
    const writerResult = await writerAgent.generate(writerPrompt);
    const draft = writerResult.text;
    await callbacks.step.onStepComplete(writeStepId, { output: draft, /* token data */ });

    // 批评阶段：在写作阶段之后立即运行，使用相同的草稿
    const criticPrompt = `RESEARCH:\n\${inputData.research}\n\nANALYSIS:\n\${inputData.analysis}\n\nDRAFT:\n\${draft}`;
    const criticStepId = await callbacks.step.onStepStart("critic", iteration, draft.slice(0, 500));
    const criticResult = await criticAgent.generate(criticPrompt);
    const parsed = extractJson(criticResult.text);
    const score = parsed?.score ?? 4;
    const feedback = parsed?.feedback ?? "Score parsing failed";
    await callbacks.step.onStepComplete(criticStepId, { output: criticResult.text, criticScore: score });

    return { ...inputData, draft, score, feedback, iterations: iteration };
  },
});

const workflow = createWorkflow({
  id: `research-pipeline-\${Date.now()}`,  // 时间戳可防止并发运行时的冲突
  inputSchema: z.object({ topic: z.string() }),
})
  .then(researchStep)
  .then(analysisStep)
  .dowhile(
    writeCriticStep,
    async ({ inputData }) => inputData.score < 7 && inputData.iterations < 3
  )
  .commit();

const inputTokens =
  (result as any).usage?.promptTokens ??
  (result as any).usage?.inputTokens ??
  0;
const outputTokens =
  (result as any).usage?.completionTokens ??
  (result as any).usage?.outputTokens ??
  0;

// packages/langchain-pipeline/src/graph/state.ts
export const PipelineState = Annotation.Root({
  topic: Annotation<string>(),
  research: Annotation<string>(),
  analysis: Annotation<string>(),
  draft: Annotation<string>(),
  score: Annotation<number>(),
  feedback: Annotation<string>(),
  iterations: Annotation<number>(),
  finalReport: Annotation<string>(),
  criticDimensions: Annotation<object>(),
});

// packages/langchain-pipeline/src/graph/nodes.ts
export function createNodes(
  callbacks: PipelineCallbacks,
  acc: { inputTokens: number; outputTokens: number }
) {
  const tavilyClient = tavily({ apiKey: process.env.TAVILY_API_KEY! });

  async function researchNode(state: PipelineStateType): Promise<Partial<PipelineStateType>> {
    const stepId = await callbacks.step.onStepStart("research", 1, state.topic);
    const results = await tavilyClient.search(state.topic, { maxResults: 5, searchDepth: "basic" });
    const research = results.results
      .map((r, i) => `[\({i + 1}] \){r.title}\nURL: \({r.url}\nContent: \){r.content}`)
      .join("\n\n");
    await callbacks.step.onStepComplete(stepId, {
      output: research,
      promptSent: state.topic,
      timeMs: elapsed,
      inputTokens: 0,      // research step uses Tavily, not an LLM
      outputTokens: 0,
      model: "tavily-search",
      tavilyQuery: state.topic,
      tavilyResults: JSON.stringify(results.results),
    });
    return { research };
  }

  // analysisNode, writeNode, criticNode follow the same pattern

  return { researchNode, analysisNode, writeNode, criticNode };
}

const inputTokens = response.usage_metadata?.input_tokens ?? 0;
const outputTokens = response.usage_metadata?.output_tokens ?? 0;
acc.inputTokens += inputTokens;
acc.outputTokens += outputTokens;

export const pipeline = new StateGraph(PipelineState)
  .addNode("researcher", researchNode)   // 不使用 "research"：与 state.research 冲突
  .addNode("analyzer", analysisNode)     // 不使用 "analysis"：与 state.analysis 冲突
  .addNode("write", writeNode)
  .addNode("critic", criticNode)
  .addEdge(START, "researcher")
  .addEdge("researcher", "analyzer")
  .addEdge("analyzer", "write")
  .addEdge("write", "critic")
  .addConditionalEdges("critic", shouldRevise, {
    revise: "write",
    end: END,
  })
  .compile();

function shouldRevise(state: PipelineStateType): string {
  if (state.score >= 7 || state.iterations >= 3) return "end";
  return "revise";
}

async function retryOnFetch<T>(fn: () => Promise<T>, retries = 3): Promise<T> {
  for (let i = 0; i <= retries; i++) {
    try {
      return await fn();
    } catch (e: any) {
      const shouldRetry =
        e?.message?.includes("fetch") ||
        e?.message === "fetch failed" ||
        e?.message?.includes("SSL") ||
        e?.message?.includes("ECONNRESET") ||
        e?.message?.includes("other side closed") ||
        e?.cause?.code === "ECONNRESET";
      if (i < retries && shouldRetry) {
        await new Promise((r) => setTimeout(r, 1000 * (i + 1)));
        continue;
      }
      throw e;
    }
  }
  throw new Error("unreachable");
}

你是一个批判性的编辑。根据准确性、清晰度和深度对草稿进行 1-10 的评分。返回 { score, feedback }。

const CRITIC_INSTRUCTIONS = `你是一位资深的研究编辑。
找出AI生成报告失败的具体方式。

步骤1：声明审核
对每个声明进行分类：[GROUNDED] [INFERRED] [UNSUPPORTED] [HALLUCINATED]

步骤2：具体性审核
列出那些通用的句子、使用了被禁止短语或没有可证伪声明的句子。被禁止的短语包括："it is important to note"（重要的是要注意）、"organizations must consider"（组织必须考虑）、"rapidly evolving"（迅速发展）、"as we look to the future"（当我们展望未来）。

步骤3：洞察审核
结论是否添加了引言中没有的内容？

步骤3.5：反事实检查
指出一个读者在阅读此内容后会持有的具体信念，而这是他们仅从主题标题中不会持有的。如果你无法识别出一个，洞察评分不能超过6分。

步骤4：对每个维度进行评分

来源忠实度（40%权重）：
5-6：声明准确，但追溯到一般主题知识，而不是这些具体结果
7：大多数声明可追溯，至少有一个来源被命名
8：所有主要声明都有依据，两个或更多被命名的来源带有具体细节
9-10：每个声明都追溯到一个被命名的来源，至少使用了一个统计数据

具体性（30%权重）：
5-6：有一些具体的声明，但段落之间的分析较为通用
7：大部分是具体的，仍有少量填充内容
8：每个段落都可证伪，全文都有命名实体
9-10：如果你更换主题，没有句子能幸存下来

function extractJson(text: string): any {
  try { return JSON.parse(text.trim()); } catch {}

  const matches = text.match(/\{[\s\S]*\}/g);
  if (matches) {
    for (let i = matches.length - 1; i >= 0; i--) {
      try { return JSON.parse(matches[i]); } catch {}
    }
  }

  const fenced = text.match(/```(?:json)?\s*([\s\S]*?)```/);
  if (fenced) {
    try { return JSON.parse(fenced[1].trim()); } catch {}
  }

  return null;
}

async function writeNode(state: PipelineStateType): Promise<Partial<PipelineStateType>> {
  const prompt = `You are a research analyst writing for a technical audience.

RESEARCH (raw search results -- cite specific sources by name):
${state.research}

ANALYSIS:
${state.analysis}
\${state.feedback ? `\nCRITIC FEEDBACK FROM PREVIOUS DRAFT:\n\${state.feedback}` : ""}

\${WRITER_INSTRUCTIONS}

Return ONLY the report text.`;

  const response = await retryOnFetch(() => llm.invoke(prompt));
  return { draft: response.content as string, iterations: (state.iterations ?? 0) + 1 };
}

steps: defineTable({
  runId: v.id("runs"),
  pipelineResultId: v.id("pipelineResults"),
  framework: v.union(v.literal("mastra"), v.literal("langchain")),
  stepName: v.union(
    v.literal("research"), v.literal("analysis"),
    v.literal("write"), v.literal("critic")
  ),
  iterationNumber: v.number(),
  status: v.union(v.literal("running"), v.literal("complete"), v.literal("error")),
  promptSent: v.optional(v.string()),
  output: v.optional(v.string()),
  timeMs: v.optional(v.number()),
  inputTokens: v.optional(v.number()),
  outputTokens: v.optional(v.number()),
  model: v.optional(v.string()),
  tavilyQuery: v.optional(v.string()),
  tavilyResults: v.optional(v.string()),
  criticScore: v.optional(v.number()),
  criticFeedback: v.optional(v.string()),
  criticDimensions: v.optional(v.object({
    fidelity: v.number(),
    specificity: v.number(),
    insight: v.number(),
    fidelityReasoning: v.string(),
    specificityReasoning: v.string(),
    insightReasoning: v.string(),
  })),
}).index("by_pipeline_result", ["pipelineResultId"]),

const activeTasks = new Map<string, Promise<void>>();

export async function POST(req: NextRequest) {
  const { topic, category } = await req.json();

  // 同步创建 Convex 记录（这些操作非常快）
  const runId = await retryMutation(() =>
    fetchMutation(api.runs.createRun, { topic, category, status: "running" })
  );
  const mastraResultId = await retryMutation(() =>
    fetchMutation(api.pipelineResults.createPipelineResult, {
      runId, framework: "mastra", status: "running", iterations: 0,
    })
  );
  const langchainResultId = await retryMutation(() =>
    fetchMutation(api.pipelineResults.createPipelineResult, {
      runId, framework: "langchain", status: "running", iterations: 0,
    })
  );

  // 不等待它们，启动两个流水线
  const task = Promise.allSettled([
    withRetry(() => runMastraPipeline(topic, buildCallbacks(runId, mastraResultId, "mastra"))),
    withRetry(() => runLangChainPipeline(topic, buildCallbacks(runId, langchainResultId, "langchain"))),
  ]).then(async () => {
    await retryMutation(() =>
      fetchMutation(api.runs.updateRunStatus, { runId, status: "complete" })
    );
    activeTasks.delete(runId as string);
  });

  // 在 Map 中保存引用，防止 Node.js 垃圾回收该 promise
  activeTasks.set(runId as string, task);
  return NextResponse.json({ runId });   // 立即返回
}

import { waitUntil } from "@vercel/functions";

waitUntil(task);
return NextResponse.json({ runId });

const mastraSteps = useQuery(
  api.steps.getStepsForPipelineResult,
  mastraResult ? { pipelineResultId: mastraResult._id } : "skip"
);

const stepMap = new Map<string, Step>();
[...steps]
  .sort((a, b) => (a._creationTime ?? 0) - (b._creationTime ?? 0))
  .forEach((s) => {
    const key = `\${s.stepName}-\${s.iterationNumber}`;
    const existing = stepMap.get(key);
    if (!existing) { stepMap.set(key, s); return; }
    if (s.status === "complete") { stepMap.set(key, s); return; }
    if (existing.status !== "complete") { stepMap.set(key, s); }
  });

function LiveLogPanel({ logs }: { logs?: LogEntry[] }) {
  const endRef = useRef<HTMLDivElement>(null);

  useEffect(() => {
    endRef.current?.scrollIntoView({ behavior: "smooth" });
  }, [logs?.length]);

  return (
    <div className="max-h-52 overflow-y-auto font-mono text-xs">
      {logs?.map((entry, i) => (
        <div key={i} className="flex gap-2">
          <span className="text-[#484f58]">[{fmtTs(entry.timestamp)}]</span>
          <span className={`font-bold w-14 ${tagColor(entry.tag)}`}>{entry.tag}</span>
          <span className="text-[#c9d1d9]">{entry.message}</span>
        </div>
      ))}
      <div ref={endRef} />
    </div>
  );
}

git clone https://github.com/sholajegede/mastra-vs-langchain.git
cd mastra-vs-langchain
npm install
cp .env.example .env
# 添加 ANTHROPIC_API_KEY 和 TAVILY_API_KEY
npx convex dev   # 终端 1
npm run web      # 终端 2