langchain-workflow-builder

LangChain 工作流构建器

使用链、Agent 和检索增强生成构建强大的大模型应用。

核心工作流

1. 设置 LangChain：安装并配置
2. 创建链：构建处理流水线
3. 添加记忆：启用对话上下文
4. 定义工具：扩展 Agent 能力
5. 实现 RAG：添加知识检索
6. 部署：配置生产环境

安装

npm install langchain @langchain/openai @langchain/community

基础链

简单 LLM 链

// chains/simple.ts
import { ChatOpenAI } from '@langchain/openai';
import { ChatPromptTemplate } from '@langchain/core/prompts';
import { StringOutputParser } from '@langchain/core/output_parsers';

const model = new ChatOpenAI({
  modelName: 'gpt-4-turbo-preview',
  temperature: 0.7,
});

const prompt = ChatPromptTemplate.fromMessages([
  ['system', 'You are a helpful assistant that {task}.'],
  ['human', '{input}'],
]);

const chain = prompt.pipe(model).pipe(new StringOutputParser());

// Usage
const result = await chain.invoke({
  task: 'summarizes text concisely',
  input: 'Summarize this article: ...',
});

顺序链

// chains/sequential.ts
import { RunnableSequence } from '@langchain/core/runnables';

// Chain 1: Extract key points
const extractChain = ChatPromptTemplate.fromMessages([
  ['system', 'Extract the key points from the following text.'],
  ['human', '{text}'],
]).pipe(model).pipe(new StringOutputParser());

// Chain 2: Summarize key points
const summarizeChain = ChatPromptTemplate.fromMessages([
  ['system', 'Create a brief summary from these key points.'],
  ['human', '{keyPoints}'],
]).pipe(model).pipe(new StringOutputParser());

// Combined chain
const fullChain = RunnableSequence.from([
  {
    keyPoints: extractChain,
    originalText: (input) => input.text,
  },
  {
    summary: summarizeChain,
    keyPoints: (input) => input.keyPoints,
  },
]);

const result = await fullChain.invoke({ text: 'Long article...' });
// { summary: '...', keyPoints: '...' }

分支链

// chains/branching.ts
import { RunnableBranch } from '@langchain/core/runnables';

const classifyChain = ChatPromptTemplate.fromMessages([
  ['system', 'Classify the query as: question, complaint, or feedback'],
  ['human', '{query}'],
]).pipe(model).pipe(new StringOutputParser());

const questionChain = ChatPromptTemplate.fromMessages([
  ['system', 'Answer this question helpfully.'],
  ['human', '{query}'],
]).pipe(model).pipe(new StringOutputParser());

const complaintChain = ChatPromptTemplate.fromMessages([
  ['system', 'Respond empathetically to this complaint.'],
  ['human', '{query}'],
]).pipe(model).pipe(new StringOutputParser());

const feedbackChain = ChatPromptTemplate.fromMessages([
  ['system', 'Thank the user for their feedback.'],
  ['human', '{query}'],
]).pipe(model).pipe(new StringOutputParser());

const routingChain = RunnableSequence.from([
  {
    classification: classifyChain,
    query: (input) => input.query,
  },
  RunnableBranch.from([
    [(input) => input.classification.includes('question'), questionChain],
    [(input) => input.classification.includes('complaint'), complaintChain],
    feedbackChain, // Default
  ]),
]);

记忆与对话

缓冲记忆

// memory/conversation.ts
import { BufferMemory } from 'langchain/memory';
import { ConversationChain } from 'langchain/chains';

const memory = new BufferMemory({
  returnMessages: true,
  memoryKey: 'history',
});

const chain = new ConversationChain({
  llm: model,
  memory,
  prompt: ChatPromptTemplate.fromMessages([
    ['system', 'You are a helpful assistant.'],
    new MessagesPlaceholder('history'),
    ['human', '{input}'],
  ]),
});

// Conversation maintains context
await chain.invoke({ input: 'My name is Alice' });
await chain.invoke({ input: 'What is my name?' }); // Remembers Alice

窗口记忆

// memory/window.ts
import { BufferWindowMemory } from 'langchain/memory';

const memory = new BufferWindowMemory({
  k: 5, // Keep last 5 exchanges
  returnMessages: true,
  memoryKey: 'history',
});

摘要记忆

// memory/summary.ts
import { ConversationSummaryMemory } from 'langchain/memory';

const memory = new ConversationSummaryMemory({
  llm: model,
  memoryKey: 'history',
});
// Summarizes conversation to save tokens

使用 Redis 的持久化记忆

// memory/redis.ts
import { BufferMemory } from 'langchain/memory';
import { RedisChatMessageHistory } from '@langchain/community/stores/message/redis';

const memory = new BufferMemory({
  chatHistory: new RedisChatMessageHistory({
    sessionId: `user:${userId}:session:${sessionId}`,
    client: redisClient,
    ttl: 3600, // 1 hour
  }),
  returnMessages: true,
  memoryKey: 'history',
});

工具与 Agent

定义自定义工具

// tools/custom.ts
import { DynamicTool, DynamicStructuredTool } from '@langchain/core/tools';
import { z } from 'zod';

// Simple tool
const searchTool = new DynamicTool({
  name: 'search',
  description: 'Search the web for information',
  func: async (query: string) => {
    const results = await searchAPI.search(query);
    return JSON.stringify(results);
  },
});

// Structured tool with schema
const calculatorTool = new DynamicStructuredTool({
  name: 'calculator',
  description: 'Perform mathematical calculations',
  schema: z.object({
    expression: z.string().describe('Mathematical expression to evaluate'),
  }),
  func: async ({ expression }) => {
    try {
      const result = eval(expression); // Use safer math parser in production
      return String(result);
    } catch {
      return 'Error: Invalid expression';
    }
  },
});

// Database query tool
const dbQueryTool = new DynamicStructuredTool({
  name: 'query_database',
  description: 'Query the database for user or order information',
  schema: z.object({
    table: z.enum(['users', 'orders', 'products']),
    filter: z.record(z.string()).optional(),
    limit: z.number().default(10),
  }),
  func: async ({ table, filter, limit }) => {
    const results = await db[table].findMany({
      where: filter,
      take: limit,
    });
    return JSON.stringify(results);
  },
});

创建 Agent

// agents/react.ts
import { createReactAgent, AgentExecutor } from 'langchain/agents';
import { pull } from 'langchain/hub';

// Get standard ReAct prompt
const prompt = await pull('hwchase17/react');

// Create agent
const agent = await createReactAgent({
  llm: model,
  tools: [searchTool, calculatorTool, dbQueryTool],
  prompt,
});

// Create executor
const executor = new AgentExecutor({
  agent,
  tools: [searchTool, calculatorTool, dbQueryTool],
  verbose: true,
  maxIterations: 5,
});

// Run agent
const result = await executor.invoke({
  input: 'What is the square root of the number of users in our database?',
});

OpenAI Functions Agent

// agents/openai-functions.ts
import { createOpenAIFunctionsAgent, AgentExecutor } from 'langchain/agents';

const prompt = ChatPromptTemplate.fromMessages([
  ['system', 'You are a helpful assistant with access to tools.'],
  new MessagesPlaceholder('chat_history'),
  ['human', '{input}'],
  new MessagesPlaceholder('agent_scratchpad'),
]);

const agent = await createOpenAIFunctionsAgent({
  llm: new ChatOpenAI({ modelName: 'gpt-4-turbo-preview' }),
  tools: [searchTool, calculatorTool],
  prompt,
});

const executor = new AgentExecutor({
  agent,
  tools: [searchTool, calculatorTool],
  memory: new BufferMemory({
    returnMessages: true,
    memoryKey: 'chat_history',
  }),
});

RAG 流水线

文档加载

// rag/loader.ts
import { DirectoryLoader } from 'langchain/document_loaders/fs/directory';
import { PDFLoader } from 'langchain/document_loaders/fs/pdf';
import { TextLoader } from 'langchain/document_loaders/fs/text';
import { JSONLoader } from 'langchain/document_loaders/fs/json';

const loader = new DirectoryLoader('./documents', {
  '.pdf': (path) => new PDFLoader(path),
  '.txt': (path) => new TextLoader(path),
  '.json': (path) => new JSONLoader(path),
});

const docs = await loader.load();

文本拆分

// rag/splitter.ts
import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter';

const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 1000,
  chunkOverlap: 200,
  separators: ['\n\n', '\n', ' ', ''],
});

const chunks = await splitter.splitDocuments(docs);

向量存储

// rag/vectorstore.ts
import { OpenAIEmbeddings } from '@langchain/openai';
import { PineconeStore } from '@langchain/pinecone';
import { Pinecone } from '@pinecone-database/pinecone';

const embeddings = new OpenAIEmbeddings();

const pinecone = new Pinecone();
const index = pinecone.Index(process.env.PINECONE_INDEX!);

// Create vector store from documents
const vectorStore = await PineconeStore.fromDocuments(chunks, embeddings, {
  pineconeIndex: index,
  namespace: 'documents',
});

// Or connect to existing
const existingStore = await PineconeStore.fromExistingIndex(embeddings, {
  pineconeIndex: index,
  namespace: 'documents',
});

RAG 链

// rag/chain.ts
import { createRetrievalChain } from 'langchain/chains/retrieval';
import { createStuffDocumentsChain } from 'langchain/chains/combine_documents';

const retriever = vectorStore.asRetriever({
  k: 5, // Return top 5 results
  searchType: 'similarity',
});

const prompt = ChatPromptTemplate.fromMessages([
  [
    'system',
    `Answer questions based on the following context. If the answer is not in the context, say so.

Context: {context}`,
  ],
  ['human', '{input}'],
]);

const combineDocsChain = await createStuffDocumentsChain({
  llm: model,
  prompt,
});

const ragChain = await createRetrievalChain({
  retriever,
  combineDocsChain,
});

const result = await ragChain.invoke({
  input: 'What is our refund policy?',
});
// { answer: '...', context: [Document, Document, ...] }

对话式 RAG

// rag/conversational.ts
import { createHistoryAwareRetriever } from 'langchain/chains/history_aware_retriever';

const contextualizePrompt = ChatPromptTemplate.fromMessages([
  ['system', 'Given chat history and question, reformulate as standalone question.'],
  new MessagesPlaceholder('chat_history'),
  ['human', '{input}'],
]);

const historyAwareRetriever = await createHistoryAwareRetriever({
  llm: model,
  retriever,
  rephrasePrompt: contextualizePrompt,
});

const qaPrompt = ChatPromptTemplate.fromMessages([
  ['system', 'Answer based on context:\n\n{context}'],
  new MessagesPlaceholder('chat_history'),
  ['human', '{input}'],
]);

const questionAnswerChain = await createStuffDocumentsChain({
  llm: model,
  prompt: qaPrompt,
});

const conversationalRagChain = await createRetrievalChain({
  retriever: historyAwareRetriever,
  combineDocsChain: questionAnswerChain,
});

流式处理

// streaming/stream.ts
const stream = await chain.stream({ input: 'Tell me a story' });

for await (const chunk of stream) {
  process.stdout.write(chunk);
}

// With callbacks
const result = await chain.invoke(
  { input: 'Tell me a story' },
  {
    callbacks: [
      {
        handleLLMNewToken(token: string) {
          process.stdout.write(token);
        },
      },
    ],
  }
);

生产环境配置

// config/langchain.ts
import { ChatOpenAI } from '@langchain/openai';

export const model = new ChatOpenAI({
  modelName: process.env.OPENAI_MODEL || 'gpt-4-turbo-preview',
  temperature: 0.7,
  maxTokens: 4096,
  timeout: 60000,
  maxRetries: 3,
  cache: true,
});

// With rate limiting
import Bottleneck from 'bottleneck';

const limiter = new Bottleneck({
  minTime: 100, // 10 requests per second
  maxConcurrent: 5,
});

export async function invokeWithRateLimit(chain: any, input: any) {
  return limiter.schedule(() => chain.invoke(input));
}

最佳实践

1. 使用 LCEL：通过 LangChain 表达式语言实现组合
2. 策略性地添加记忆：平衡上下文与 token 数量
3. 编写清晰的工具描述：帮助 Agent 正确选择
4. 妥善拆分文档：优化检索效果
5. 使用流式处理：改善长回复的用户体验
6. 缓存嵌入向量：降低 API 成本
7. 监控 token 用量：跟踪并优化
8. 妥善处理错误：重试并使用备用方案

输出检查清单

每个 LangChain 实现都应包含：

• 配置带重试机制的模型
• 使用变量的提示词模板
• 使用 LCEL 组合链
• 用于对话上下文的记忆
• 描述清晰的工具
• 设置迭代次数限制的 Agent
• 使用恰当分块的 RAG
• 提供流式处理以改善用户体验
• 错误处理
• 用于生产环境的速率限制