1. LangGraph Agent工作流全景解析最近在AI编程圈里LangGraph这个框架突然火了起来。作为一个长期关注AI工程化的开发者我花了三周时间深度体验了LangGraph的完整工作流特别是它在文本处理任务上的表现确实让人惊艳。今天就用最直白的语言带大家走通从文本分类到自动生成摘要的完整流程就算你是刚接触AI编程的新手跟着做也能快速上手。LangGraph本质上是一个面向长时间运行、有状态工作流的智能体编排框架。和常见的LangChain相比它的核心优势在于对复杂任务的状态管理和持久化支持。举个例子当处理一篇万字长文时传统方法可能需要手动拆分段落再逐个处理而LangGraph可以自动维护处理进度和中间状态即使程序中途崩溃也能从断点继续。2. 环境准备与基础配置2.1 安装与最小化验证建议使用Python 3.9环境先创建一个干净的虚拟环境python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows安装核心依赖包pip install -U langgraph pip install langchain[anthropic] # 用于调用Claude模型验证安装是否成功from langgraph.prebuilt import create_react_agent def mock_tool(query: str) - str: 测试用工具函数 return f已处理: {query} agent create_react_agent( modelanthropic:claude-3-7-sonnet-latest, tools[mock_tool], prompt你是测试助手 ) print(agent.invoke({messages: [{role: user, content: ping}]}))2.2 关键组件说明LangGraph的工作流主要由这些核心部件构成智能体(Agent)任务执行主体包含LLM核心和工具集工具(Tools)具体功能实现比如文本处理API状态(State)维护任务执行上下文边(Edges)定义工作流跳转逻辑典型文本处理工作流的组件对应关系组件类型文本分类场景摘要生成场景工具分类模型接口摘要模型接口状态待分类文本分类结果原文分段摘要最终摘要边分类→结果校验→输出分段→汇总→润色→输出3. 文本分类实战实现3.1 分类工具封装我们先实现一个基于规则模型的混合分类器from typing import Literal from snownlp import SnowNLP class TextClassifier: staticmethod def rule_based(text: str) - Literal[科技, 财经, 社会, 其他]: 基于关键词的粗分类 keywords { 科技: [AI, 算法, 模型, 神经网络], 财经: [股票, 基金, 利率, 通胀], 社会: [疫情, 教育, 就业, 医疗] } for cat, words in keywords.items(): if any(word in text for word in words): return cat return 其他 staticmethod def model_based(text: str) - float: 情感倾向评分 return SnowNLP(text).sentiments def classify_text(text: str) - dict: 分类工具函数 category TextClassifier.rule_based(text) sentiment TextClassifier.model_based(text) return { category: category, sentiment: round(sentiment, 2), length: len(text) }3.2 构建分类智能体创建专门处理分类任务的智能体from langgraph.prebuilt import create_react_agent from langchain_anthropic import ChatAnthropic classifier_agent create_react_agent( modelChatAnthropic(modelclaude-3-haiku-20240307), tools[classify_text], prompt你是一个专业文本分类助手请根据用户提供的文本 1. 调用工具获取基础分类 2. 结合文本内容补充分类理由 3. 输出格式{ category: ..., reason: ..., meta: {原始工具输出} } )3.3 分类流程执行示例测试分类效果sample_text Transformer架构在LLM领域引发革命性变化 result classifier_agent.invoke({ messages: [{ role: user, content: sample_text }] }) print(result)典型输出结构{ category: 科技, reason: 文本提及Transformer架构和LLM领域属于人工智能技术范畴, meta: { category: 科技, sentiment: 0.87, length: 28 } }4. 自动摘要生成进阶4.1 分段摘要工作流设计处理长文本摘要时需要分阶段进行文本分块超过500字自动拆分各段落独立摘要摘要聚合与精炼最终风格调整实现这个工作流需要定义状态结构from typing import TypedDict, List from langgraph.graph.message import AnyMessage class SummaryState(TypedDict): original_text: str chunks: List[str] chunk_summaries: List[str] final_summary: str feedback: AnyMessage4.2 关键工具实现import re from dify import DifyClient # 假设使用Dify的摘要API def split_text(text: str, max_len500) - List[str]: 智能文本分块 sentences re.split(r(?[。]), text) chunks [] current_chunk for sent in sentences: if len(current_chunk) len(sent) max_len: chunks.append(current_chunk) current_chunk sent else: current_chunk sent if current_chunk: chunks.append(current_chunk) return chunks def generate_chunk_summary(text: str) - str: 单段落摘要 client DifyClient(api_keyyour_key) prompt f用中文总结以下内容保留关键事实不超过100字\n{text} return client.complete(prompt) def refine_summary(summaries: List[str]) - str: 摘要聚合 joined \n.join(f- {s} for s in summaries) client DifyClient(api_keyyour_key) prompt f整合以下分段摘要为连贯的完整摘要保持专业语气\n{joined} return client.complete(prompt)4.3 完整摘要智能体配置from langgraph.graph import END, MessageGraph workflow MessageGraph() # 定义工作流节点 workflow.add_node(split, lambda state: {chunks: split_text(state[original_text])}) workflow.add_node(summarize_chunks, lambda state: { chunk_summaries: [generate_chunk_summary(chunk) for chunk in state[chunks]] }) workflow.add_node(combine, lambda state: { final_summary: refine_summary(state[chunk_summaries]) }) # 定义边关系 workflow.add_edge(split, summarize_chunks) workflow.add_edge(summarize_chunks, combine) workflow.add_edge(combine, END) # 编译为可执行工作流 summary_agent workflow.compile()5. 实战问题排查指南5.1 常见错误与解决方案错误现象可能原因解决方案工具调用超时API速率限制添加retry逻辑建议使用tenacity库中文输出乱码编码问题确保所有环节使用UTF-8摘要信息缺失分块不合理调整分块策略按语义而非固定长度分割分类结果不一致规则与模型冲突添加投票机制优先采用模型结果5.2 性能优化技巧缓存机制对相同文本内容缓存处理结果from functools import lru_cache lru_cache(maxsize100) def cached_classify(text: str): return classify_text(text)批量处理当需要处理多个文本时使用异步import asyncio async def batch_classify(texts: List[str]): return await asyncio.gather( *[classifier_agent.ainvoke({messages: [{role: user, content: t}]}) for t in texts] )动态温度系数根据文本复杂度调整LLM的temperature参数def dynamic_temperature(text: str) - float: length len(text) if length 1000: return 0.3 # 长文本使用低随机性 return 0.7 # 短文本允许更多变化6. 完整案例演示让我们用一篇科技新闻测试完整流程news_article 近日OpenAI发布了新一代语言模型GPT-4o...此处省略800字正文 # 执行分类 classification classifier_agent.invoke({ messages: [{role: user, content: news_article[:500]}] # 截取部分内容分类 }) # 生成摘要 summary_result summary_agent.invoke({ original_text: news_article }) print(f分类结果{classification[category]}) print(f文章摘要{summary_result[final_summary]})在实际项目中我发现这些经验特别有价值对于技术类文本在摘要前先提取专业术语表能显著提升质量分类时添加时效性判断如是否含近日等时间词可以提高分类准确率使用Claude-haiku模型处理摘要在效果和成本间取得很好平衡 LangGraph框架实战：从文本分类到自动摘要的AI工作流
1. LangGraph Agent工作流全景解析最近在AI编程圈里LangGraph这个框架突然火了起来。作为一个长期关注AI工程化的开发者我花了三周时间深度体验了LangGraph的完整工作流特别是它在文本处理任务上的表现确实让人惊艳。今天就用最直白的语言带大家走通从文本分类到自动生成摘要的完整流程就算你是刚接触AI编程的新手跟着做也能快速上手。LangGraph本质上是一个面向长时间运行、有状态工作流的智能体编排框架。和常见的LangChain相比它的核心优势在于对复杂任务的状态管理和持久化支持。举个例子当处理一篇万字长文时传统方法可能需要手动拆分段落再逐个处理而LangGraph可以自动维护处理进度和中间状态即使程序中途崩溃也能从断点继续。2. 环境准备与基础配置2.1 安装与最小化验证建议使用Python 3.9环境先创建一个干净的虚拟环境python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows安装核心依赖包pip install -U langgraph pip install langchain[anthropic] # 用于调用Claude模型验证安装是否成功from langgraph.prebuilt import create_react_agent def mock_tool(query: str) - str: 测试用工具函数 return f已处理: {query} agent create_react_agent( modelanthropic:claude-3-7-sonnet-latest, tools[mock_tool], prompt你是测试助手 ) print(agent.invoke({messages: [{role: user, content: ping}]}))2.2 关键组件说明LangGraph的工作流主要由这些核心部件构成智能体(Agent)任务执行主体包含LLM核心和工具集工具(Tools)具体功能实现比如文本处理API状态(State)维护任务执行上下文边(Edges)定义工作流跳转逻辑典型文本处理工作流的组件对应关系组件类型文本分类场景摘要生成场景工具分类模型接口摘要模型接口状态待分类文本分类结果原文分段摘要最终摘要边分类→结果校验→输出分段→汇总→润色→输出3. 文本分类实战实现3.1 分类工具封装我们先实现一个基于规则模型的混合分类器from typing import Literal from snownlp import SnowNLP class TextClassifier: staticmethod def rule_based(text: str) - Literal[科技, 财经, 社会, 其他]: 基于关键词的粗分类 keywords { 科技: [AI, 算法, 模型, 神经网络], 财经: [股票, 基金, 利率, 通胀], 社会: [疫情, 教育, 就业, 医疗] } for cat, words in keywords.items(): if any(word in text for word in words): return cat return 其他 staticmethod def model_based(text: str) - float: 情感倾向评分 return SnowNLP(text).sentiments def classify_text(text: str) - dict: 分类工具函数 category TextClassifier.rule_based(text) sentiment TextClassifier.model_based(text) return { category: category, sentiment: round(sentiment, 2), length: len(text) }3.2 构建分类智能体创建专门处理分类任务的智能体from langgraph.prebuilt import create_react_agent from langchain_anthropic import ChatAnthropic classifier_agent create_react_agent( modelChatAnthropic(modelclaude-3-haiku-20240307), tools[classify_text], prompt你是一个专业文本分类助手请根据用户提供的文本 1. 调用工具获取基础分类 2. 结合文本内容补充分类理由 3. 输出格式{ category: ..., reason: ..., meta: {原始工具输出} } )3.3 分类流程执行示例测试分类效果sample_text Transformer架构在LLM领域引发革命性变化 result classifier_agent.invoke({ messages: [{ role: user, content: sample_text }] }) print(result)典型输出结构{ category: 科技, reason: 文本提及Transformer架构和LLM领域属于人工智能技术范畴, meta: { category: 科技, sentiment: 0.87, length: 28 } }4. 自动摘要生成进阶4.1 分段摘要工作流设计处理长文本摘要时需要分阶段进行文本分块超过500字自动拆分各段落独立摘要摘要聚合与精炼最终风格调整实现这个工作流需要定义状态结构from typing import TypedDict, List from langgraph.graph.message import AnyMessage class SummaryState(TypedDict): original_text: str chunks: List[str] chunk_summaries: List[str] final_summary: str feedback: AnyMessage4.2 关键工具实现import re from dify import DifyClient # 假设使用Dify的摘要API def split_text(text: str, max_len500) - List[str]: 智能文本分块 sentences re.split(r(?[。]), text) chunks [] current_chunk for sent in sentences: if len(current_chunk) len(sent) max_len: chunks.append(current_chunk) current_chunk sent else: current_chunk sent if current_chunk: chunks.append(current_chunk) return chunks def generate_chunk_summary(text: str) - str: 单段落摘要 client DifyClient(api_keyyour_key) prompt f用中文总结以下内容保留关键事实不超过100字\n{text} return client.complete(prompt) def refine_summary(summaries: List[str]) - str: 摘要聚合 joined \n.join(f- {s} for s in summaries) client DifyClient(api_keyyour_key) prompt f整合以下分段摘要为连贯的完整摘要保持专业语气\n{joined} return client.complete(prompt)4.3 完整摘要智能体配置from langgraph.graph import END, MessageGraph workflow MessageGraph() # 定义工作流节点 workflow.add_node(split, lambda state: {chunks: split_text(state[original_text])}) workflow.add_node(summarize_chunks, lambda state: { chunk_summaries: [generate_chunk_summary(chunk) for chunk in state[chunks]] }) workflow.add_node(combine, lambda state: { final_summary: refine_summary(state[chunk_summaries]) }) # 定义边关系 workflow.add_edge(split, summarize_chunks) workflow.add_edge(summarize_chunks, combine) workflow.add_edge(combine, END) # 编译为可执行工作流 summary_agent workflow.compile()5. 实战问题排查指南5.1 常见错误与解决方案错误现象可能原因解决方案工具调用超时API速率限制添加retry逻辑建议使用tenacity库中文输出乱码编码问题确保所有环节使用UTF-8摘要信息缺失分块不合理调整分块策略按语义而非固定长度分割分类结果不一致规则与模型冲突添加投票机制优先采用模型结果5.2 性能优化技巧缓存机制对相同文本内容缓存处理结果from functools import lru_cache lru_cache(maxsize100) def cached_classify(text: str): return classify_text(text)批量处理当需要处理多个文本时使用异步import asyncio async def batch_classify(texts: List[str]): return await asyncio.gather( *[classifier_agent.ainvoke({messages: [{role: user, content: t}]}) for t in texts] )动态温度系数根据文本复杂度调整LLM的temperature参数def dynamic_temperature(text: str) - float: length len(text) if length 1000: return 0.3 # 长文本使用低随机性 return 0.7 # 短文本允许更多变化6. 完整案例演示让我们用一篇科技新闻测试完整流程news_article 近日OpenAI发布了新一代语言模型GPT-4o...此处省略800字正文 # 执行分类 classification classifier_agent.invoke({ messages: [{role: user, content: news_article[:500]}] # 截取部分内容分类 }) # 生成摘要 summary_result summary_agent.invoke({ original_text: news_article }) print(f分类结果{classification[category]}) print(f文章摘要{summary_result[final_summary]})在实际项目中我发现这些经验特别有价值对于技术类文本在摘要前先提取专业术语表能显著提升质量分类时添加时效性判断如是否含近日等时间词可以提高分类准确率使用Claude-haiku模型处理摘要在效果和成本间取得很好平衡