877-LangChain框架Use-Cases - LangGraph Studio多智能体系统分析-深圳市維司達科技有限公司

案例目标

本案例展示了如何使用LangChain和LangGraph Studio构建一个多智能体工作流系统，用于研究特定人物、职业背景、所属公司，并生成后续问题或面试提示。该系统通过多个专门的智能体协同工作，实现从信息收集、分析到问题生成的完整流程，并通过LangGraph Studio提供可视化调试和扩展功能。

核心目标：

构建多智能体协同工作流，实现信息收集与分析的自动化
集成LangGraph Studio，提供可视化调试界面
实现人员、公司和项目信息的全面研究
基于研究结果生成高质量的面试问题

技术栈与核心依赖

核心框架

LangChain LangGraph LangGraph Studio LangSmith

AI模型

Anthropic Claude OpenAI GPT

数据与工具

Tavily搜索 Docker Python 3.11

辅助库

Pydantic typing-extensions tavily-python

环境配置

# 安装核心依赖 %pip install -qU langchain-opentutorial langsmith langchain-anthropic langgraph tavily-python # 环境变量配置 import os os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key" os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langchain-api-key" os.environ["TAVILY_API_KEY"] = "your-tavily-api-key"

LangGraph Studio环境要求：

Docker Desktop（用于运行LangGraph Studio）
langgraph_studio目录结构，包含langgraph.json配置文件
requirements.txt文件定义项目依赖
.env文件存储环境变量

案例实现

1. 数据模型定义

# 人员信息模型 class Person(BaseModel): """要研究的人员""" name: str = Field(description="人员姓名") company: str = Field(description="所属公司") background: str = Field(description="职业背景") # 状态管理模型 class InputState(BaseModel): person: Person = Field(description="要研究的人员") class OverallState(BaseModel): person: Person = Field(description="要研究的人员") company_notes: Optional[str] = Field(description="公司研究笔记") person_notes: Optional[str] = Field(description="人员研究笔记") project_notes: Optional[str] = Field(description="项目研究笔记") questions: Optional[List[str]] = Field(description="生成的问题") reflection: Optional[ReflectionOutput] = Field(description="反思结果") class OutputState(BaseModel): questions: List[str] = Field(description="生成的问题")

2. 查询生成模块

# 人员研究查询生成 def generate_queries(state: OverallState) -> OverallState: """为人员研究生成搜索查询""" prompt = QUERY_WRITER_PROMPT.format( person_name=state.person.name, person_background=state.person.background, person_company=state.person.company ) queries = llm.invoke(prompt) return {"queries": queries} # 公司研究查询生成 def generate_queries_for_company(state: OverallState) -> OverallState: """为公司研究生成搜索查询""" prompt = SEARCH_COMPANY_PROMPT.format( company_name=state.person.company, person_name=state.person.name ) queries = llm.invoke(prompt) return {"company_queries": queries}

3. 信息研究模块

# 公司研究 def research_company(state: OverallState) -> OverallState: """研究公司信息""" queries = json.loads(state.company_queries) search_results = tavily_search.run(queries) prompt = COMPANY_INFO_PROMPT.format( company_name=state.person.company, person_name=state.person.name, search_results=search_results ) company_notes = llm.invoke(prompt) return {"company_notes": company_notes} # 人员研究 def research_person(state: OverallState) -> OverallState: """研究人员信息""" queries = json.loads(state.queries) search_results = tavily_search.run(queries) prompt = INFO_PROMPT.format( person_name=state.person.name, person_background=state.person.background, search_results=search_results ) person_notes = llm.invoke(prompt) return {"person_notes": person_notes}

4. 项目研究与信息整合

# 项目查询提取 def extract_project_queries(state: OverallState) -> OverallState: """从人员笔记中提取项目相关查询""" prompt = PROJECT_EXTRACTOR_PROMPT.format( person_notes=state.person_notes ) project_queries = llm.invoke(prompt) return {"project_queries": project_queries} # 项目研究 def research_projects(state: OverallState) -> OverallState: """研究项目信息""" queries = json.loads(state.project_queries) search_results = tavily_search.run(queries) prompt = PROJECT_INFO_PROMPT.format( person_name=state.person.name, search_results=search_results ) project_notes = llm.invoke(prompt) return {"project_notes": project_notes} # 信息整合 def combine_notes(state: OverallState) -> OverallState: """合并所有笔记""" combined = f""" # 人员笔记 {state.person_notes} # 公司笔记 {state.company_notes} # 项目笔记 {state.project_notes} """ return {"combined_notes": combined}

5. 问题生成与反思

# 问题生成 def generate_questions(state: OverallState) -> OverallState: """基于合并笔记生成面试问题""" prompt = QUESTION_WRITER_PROMPT.format( person_name=state.person.name, person_background=state.person.background, combined_notes=state.combined_notes ) questions = llm.invoke(prompt) return {"questions": questions} # 反思模块 def reflection(state: OverallState) -> OverallState: """评估信息质量并决定是否需要更多研究""" prompt = REFLECTION_PROMPT.format( person_name=state.person.name, person_background=state.person.background, person_notes=state.person_notes, company_notes=state.company_notes, project_notes=state.project_notes, questions=state.questions ) reflection_output = llm.with_structured_output(ReflectionOutput).invoke(prompt) return {"reflection": reflection_output}

6. LangGraph工作流构建

# 构建工作流图 builder = StateGraph(OverallState, input=InputState, output=OutputState) # 添加节点 builder.add_node("generate_queries", generate_queries) builder.add_node("research_person", research_person) builder.add_node("generate_queries_for_company", generate_queries_for_company) builder.add_node("research_company", research_company) builder.add_node("extract_project_queries", extract_project_queries) builder.add_node("research_projects", research_projects) builder.add_node("combine_notes", combine_notes) builder.add_node("generate_questions", generate_questions) builder.add_node("reflection", reflection) # 添加边 builder.add_edge(START, "generate_queries") builder.add_edge("generate_queries", "research_person") builder.add_edge("research_person", "generate_queries_for_company") builder.add_edge("generate_queries_for_company", "research_company") builder.add_edge("research_company", "extract_project_queries") builder.add_edge("extract_project_queries", "research_projects") builder.add_edge("research_projects", "combine_notes") builder.add_edge("combine_notes", "generate_questions") builder.add_edge("generate_questions", "reflection") # 添加条件边 builder.add_conditional_edges( "reflection", route_from_reflection, { "generate_more_queries": "generate_queries", "end": END } ) # 编译图 graph = builder.compile()

7. LangGraph Studio集成

# langgraph.json配置文件 { "dependencies": ["."], "graphs": { "multi_agent": "./langgraph_studio/output_script.py:graph" } } # 代码提取脚本 def extract_code_cells(notebook_path, output_path): """从Jupyter notebook提取代码单元格并转换为可执行脚本""" with open(notebook_path, 'r', encoding='utf-8') as f: notebook = json.load(f) # 处理代码单元格 for cell in notebook['cells']: if cell['cell_type'] == 'code': # 转换pip安装命令 # 处理可视化代码 # 去重并保存 # 保存转换后的代码 with open(output_path, 'w', encoding='utf-8') as f: for cell in unique_cells: f.write(cell + '\n\n')

案例效果

该多智能体系统实现了以下效果：

全面信息收集：通过多个专门智能体，分别收集人员、公司和项目相关信息
智能问题生成：基于收集的信息，生成针对性和专业性强的面试问题
可视化调试：通过LangGraph Studio提供直观的工作流可视化界面，便于调试和优化
自适应研究：通过反思机制，系统能够评估信息质量并决定是否需要进一步研究

生成的问题示例：

您能描述一下您从Tiffany & Co.的数据分析转向AI工程的转变过程，以及您之前的经验如何影响您当前在LLM和RAG系统方面的工作吗？
在开发"Ticki tacka"项目控制系统时，您遇到了哪些具体的NLP挑战，您是如何解决这些挑战的？
您能介绍一下"Ticki tacka"系统的核心AI组件以及您在开发过程中的角色吗？
您在实施RAG系统方面有哪些经验，在企业应用中遇到了哪些挑战？
您如何平衡AI解决方案中的模型性能和实际业务需求？

案例实现思路

本案例的实现基于以下思路：

模块化设计：将复杂任务分解为多个专门模块，每个模块负责特定功能
状态管理：使用Pydantic模型定义工作流状态，确保数据结构的一致性和类型安全
智能体协同：通过LangGraph构建智能体工作流，实现信息收集、处理和生成的自动化
反馈循环：引入反思机制，评估工作流输出质量并决定是否需要迭代优化
可视化调试：集成LangGraph Studio，提供直观的工作流可视化界面，便于调试和优化

关键技术点：

使用LangGraph的StateGraph构建复杂工作流
通过条件边实现基于状态的流程控制
利用结构化输出确保LLM响应的一致性
通过LangGraph Studio实现工作流可视化调试

扩展建议

功能扩展

增加更多研究维度，如行业趋势、竞争对手分析
实现多语言支持，扩展非英语信息收集能力
添加信息可信度评估机制
实现问题分类和难度评级

技术优化

引入缓存机制，减少重复搜索请求
实现并行处理，提高研究效率
添加用户反馈机制，持续优化问题质量
集成更多数据源，提高信息全面性

总结

本案例展示了如何使用LangChain和LangGraph Studio构建一个复杂的多智能体工作流系统。通过模块化设计、智能体协同和可视化调试，该系统能够自动收集、分析信息并生成高质量的面试问题。LangGraph Studio的集成大大简化了复杂工作流的开发和调试过程，为构建高级AI应用提供了强大支持。

该系统的核心价值在于将复杂任务分解为多个专门模块，通过智能体协同工作实现自动化处理，同时通过可视化工具提供直观的调试和优化界面。这种设计模式可以广泛应用于各种需要多步骤、多维度信息处理的AI应用场景。