AI智能体工程化实战：基于LangGraph构建多智能体协作系统-北京尧图网络

大家好我是专注于技术实战分享的博主。在探索AI工程化落地的过程中我们常常面临一个核心挑战如何将前沿的AI能力特别是智能体Agents有效地整合到现有的软件工程流程中这不仅仅是调用一个API那么简单它涉及到代码库Codebases的架构设计、团队Teams协作模式的变革以及一系列工程化实践的建立。本文将围绕“智能体、代码库与团队”这一主题深入探讨如何作为一名AI工程师AI Engineer系统性地构建、管理和迭代基于大语言模型LLM的智能体应用。无论你是希望将AI能力引入现有项目的开发者还是正在从零构建AI驱动产品的团队负责人本文都将提供一套从概念到部署的完整实操指南。1. 智能体Agents的核心概念与工程价值在传统软件开发中程序的行为由预先编写的、确定的逻辑控制。而基于LLM的智能体其核心在于引入了“推理”和“决策”能力。它可以根据目标、上下文和工具Tools来规划并执行一系列动作从而完成更复杂的任务。1.1 什么是AI智能体简单来说一个AI智能体是一个能够感知环境、进行思考推理、并采取行动以实现特定目标的软件实体。在LLM的语境下这个“思考”过程由大语言模型驱动。一个典型的智能体工作流包括目标理解智能体解析用户或系统给出的指令如“分析上个月的销售数据并生成报告”。任务规划智能体将复杂目标拆解为一系列可执行的子任务如1. 连接数据库2. 查询销售数据3. 进行数据分析4. 调用报告生成工具。工具调用智能体根据任务需求选择并调用预先定义好的工具Tools如执行SQL查询、调用外部API、读写文件等。观察与迭代智能体观察工具执行的结果评估是否达成子目标并决定下一步行动直至最终目标完成或无法继续。1.2 为什么需要关注代码库与团队这正是AI工程化AI Engineering的关键所在。如果只是实验性地构建一个智能体原型可能只需要一个Jupyter Notebook。但要将其转化为可维护、可扩展、可协作的生产级应用就必须考虑代码库Codebases智能体的逻辑、工具定义、提示词Prompts、记忆Memory管理、配置等如何组织如何版本控制如何与现有业务代码集成团队Teams智能体的开发涉及提示词工程师、后端开发者、前端开发者、产品经理、运维工程师等多个角色。他们如何协作职责边界如何划分如何建立评审和测试流程忽视这两点很容易导致“智能体孤岛”——一堆无法维护、无法理解、且与核心业务脱节的实验性代码最终难以产生实际业务价值。2. 环境准备与核心框架选择在开始构建之前我们需要搭建开发环境并选择合适的框架。目前社区有多种优秀的智能体框架它们抽象了智能体的核心循环让我们能更专注于业务逻辑。2.1 环境与工具栈编程语言Python 是目前AI智能体生态最丰富的语言本文示例将基于Python。Python版本建议使用 Python 3.10 或更高版本。包管理使用pip或更推荐的poetry/uv进行依赖管理。LLM服务你需要一个LLM的API访问权限。本文示例使用 OpenAI 的 GPT-4 模型但你也可以轻松替换为 Anthropic Claude、Google Gemini 或开源模型通过 Ollama、vLLM 等。版本控制Git 是必须的。2.2 主流框架简介与选择LangChain / LangGraphLangChain提供了构建链Chains和智能体的基础模块如模型封装、提示词模板、记忆、工具等。它非常灵活但需要更多配置。LangGraph建立在LangChain之上用于构建有状态的、多智能体工作流。它通过图Graph来定义智能体之间的交互和状态流转非常适合复杂场景。LlamaIndex最初专注于数据索引和检索现已扩展为强大的智能体框架尤其在处理私有数据文档、数据库方面有优势。AutoGen (by Microsoft)专注于多智能体对话和协作。你可以轻松定义不同的智能体角色如程序员、产品经理、测试员并让它们通过对话解决问题。CrewAI一个较新的框架强调角色扮演Role-playing和任务导向的多智能体协作设计上更贴近人类团队的工作模式。选择建议对于刚入门或构建相对简单的单智能体应用可以从LangChain开始。当你需要构建涉及多个智能体协作、有复杂状态管理的系统时LangGraph或CrewAI是更好的选择。本文将以LangChain和LangGraph为主要示例框架。2.3 初始化项目首先创建一个干净的项目目录并初始化虚拟环境。# 创建项目目录 mkdir ai-agent-project cd ai-agent-project # 创建虚拟环境 (以 venv 为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install langchain langchain-openai langgraph # 如果需要与网络或文档交互可以安装以下工具 pip install langchain-community requests beautifulsoup4创建基本的项目结构ai-agent-project/ ├── .gitignore ├── pyproject.toml # 如果使用 poetry ├── requirements.txt # 如果使用 pip ├── src/ │ ├── agents/ # 智能体定义 │ ├── tools/ # 工具定义 │ ├── memory/ # 记忆管理 │ ├── config/ # 配置文件 │ └── main.py # 应用入口 └── tests/ # 测试文件3. 构建你的第一个智能体从单智能体到多智能体工作流我们将从一个简单的单智能体开始逐步构建一个能进行网络搜索和总结的多智能体系统。3.1 单智能体基础工具调用假设我们要构建一个能查询天气的智能体。首先我们需要定义一个“获取天气”的工具。步骤1定义工具Tool在src/tools/weather_tool.py中import requests from typing import Optional from langchain.tools import tool from pydantic import BaseModel, Field # 定义工具的输入模型Schema class WeatherInput(BaseModel): city: str Field(descriptionThe city name to get weather for, e.g., Beijing) tool(args_schemaWeatherInput) def get_weather(city: str) - str: Get the current weather for a given city. # 注意这里使用了一个模拟API真实场景请替换为可靠的天气API如OpenWeatherMap # 并且务必处理API密钥的安全存储不要硬编码在代码中。 try: # 模拟API调用 # response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{city}) # data response.json() # return fThe weather in {city} is {data[current][condition][text]}, temperature: {data[current][temp_c]}°C # 模拟返回 return fThe weather in {city} is sunny, 25°C. (This is a mock response. Please integrate a real weather API.) except Exception as e: return fFailed to get weather for {city}: {str(e)}步骤2创建智能体Agent在src/agents/weather_agent.py中import os from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from src.tools.weather_tool import get_weather # 1. 初始化LLM (请将你的API Key设置在环境变量中) # export OPENAI_API_KEYyour-api-key-here llm ChatOpenAI(modelgpt-4o, temperature0) # 2. 定义工具列表 tools [get_weather] # 3. 定义提示词模板 prompt ChatPromptTemplate.from_messages([ (system, You are a helpful assistant that can provide weather information. Use the tools available to you.), MessagesPlaceholder(variable_namechat_history), # 预留历史消息位置 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 智能体思考过程 ]) # 4. 创建智能体 agent create_tool_calling_agent(llmllm, toolstools, promptprompt) # 5. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 使用示例 if __name__ __main__: result agent_executor.invoke({input: Whats the weather like in Shanghai today?}) print(result[output])运行这个文件你会看到智能体识别出需要调用get_weather工具并返回结果。verboseTrue会打印出详细的推理步骤。3.2 引入状态管理使用LangGraph构建多智能体工作流单智能体适合简单任务。复杂任务通常需要多个智能体协作每个智能体负责特定角色并且它们之间需要共享状态。这就是LangGraph的用武之地。场景构建一个“研究助手”工作流包含两个智能体研究员Researcher负责根据主题进行网络搜索收集信息。撰稿人Writer负责将收集到的信息整理成结构化的报告。步骤1定义状态State在src/agents/research_state.py中我们定义一个共享的状态类用于在智能体间传递信息。from typing import TypedDict, List, Annotated import operator class ResearchState(TypedDict): # 用户输入的主题 topic: str # 研究员收集到的资料列表 research_materials: List[str] # 撰稿人生成的报告 report: str # 控制流程的指令例如继续研究、开始撰写、结束 next_step: str步骤2定义工具和节点Nodes节点是工作流中的基本执行单元可以是一个函数或一个智能体。首先为研究员定义一个搜索工具模拟在src/tools/search_tool.pyfrom langchain.tools import tool tool def web_search(query: str) - str: Perform a web search about a given topic and return summarized snippets. # 模拟搜索真实场景可集成Serper API、Google Search API等 mock_results { AI Agents: AI agents are systems that can autonomously plan and execute actions using LLMs. They are key to AI Engineering., LangGraph: LangGraph is a library for building stateful, multi-actor applications with LLMs, extending LangChain., CrewAI: CrewAI is a framework for orchestrating role-playing, autonomous AI agents. } # 简单模拟返回相关结果 for key, value in mock_results.items(): if key.lower() in query.lower(): return fSearch result for {key}: {value} return fFound general information about {query}: This is a rapidly evolving field in AI engineering.然后创建研究员节点和撰稿人节点在src/agents/research_crew.pyfrom langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langgraph.graph import StateGraph, END from src.agents.research_state import ResearchState from src.tools.search_tool import web_search llm ChatOpenAI(modelgpt-4o, temperature0.7) # 撰稿人可以更有创造性 def research_node(state: ResearchState): 研究员节点执行搜索收集资料 print(f[Researcher] Researching topic: {state[topic]}) search_result web_search.invoke(state[topic]) # 将收集到的资料添加到状态中 new_materials state.get(research_materials, []) [search_result] # 更新状态并指示下一步是撰写 return {research_materials: new_materials, next_step: write} def write_node(state: ResearchState): 撰稿人节点根据资料撰写报告 print(f[Writer] Writing report based on {len(state[research_materials])} research materials.) materials \n---\n.join(state[research_materials]) prompt ChatPromptTemplate.from_messages([ (system, You are a technical writer. Create a concise, well-structured report based on the provided research materials.), (human, fResearch Topic: {state[topic]}\n\nCollected Materials:\n{materials}\n\nPlease write a report.) ]) chain prompt | llm report chain.invoke({}) # 更新报告并指示工作流结束 return {report: report.content, next_step: end}步骤3构建并编译图Graph在同一个文件或主入口中我们将节点连接起来定义工作流逻辑。# 继续在 research_crew.py 中 def should_continue(state: ResearchState) - str: 根据状态中的 next_step 决定下一个节点 next_step state.get(next_step, research) if next_step write: return write_node elif next_step end: return END else: # 默认先进行研究 return research_node # 创建图 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(research_node, research_node) workflow.add_node(write_node, write_node) # 设置入口点 workflow.set_entry_point(research_node) # 添加条件边Conditional Edge workflow.add_conditional_edges( research_node, should_continue # 这个函数决定从 research_node 出来后去哪 ) workflow.add_conditional_edges( write_node, should_continue # 这个函数决定从 write_node 出来后去哪应该是END ) # 编译图 app workflow.compile()步骤4运行工作流在src/main.py中from src.agents.research_crew import app from src.agents.research_state import ResearchState if __name__ __main__: # 初始化状态 initial_state: ResearchState { topic: AI Agents and LangGraph, research_materials: [], report: , next_step: research } print(Starting research workflow...) # 运行图 final_state app.invoke(initial_state) print(\n *50) print(FINAL REPORT:) print(*50) print(final_state[report])运行main.py你将看到研究员和撰稿人依次执行最终生成一份关于“AI Agents and LangGraph”的简短报告。这个例子展示了如何用有状态的工作流来组织多智能体协作。4. 工程化实践代码库管理与团队协作构建出可运行的智能体只是第一步。要使其成为团队资产必须考虑工程化。4.1 代码库组织最佳实践一个清晰的代码结构能极大提升可维护性。以下是一种推荐结构ai-agent-production/ ├── .env.example # 环境变量示例 ├── .gitignore ├── pyproject.toml # 依赖和项目配置 ├── README.md # 项目说明、快速开始 ├── docs/ # 项目文档 ├── tests/ # 单元测试、集成测试 │ ├── unit/ │ └── integration/ ├── src/ │ ├── __init__.py │ ├── main.py # 应用主入口/API入口 │ ├── config/ # 配置管理 │ │ ├── __init__.py │ │ ├── settings.py # Pydantic Settings 管理配置 │ │ └── prompts/ # 将提示词模板作为配置文件 │ │ ├── researcher.yaml │ │ └── writer.yaml │ ├── agents/ # 智能体定义 │ │ ├── __init__.py │ │ ├── base_agent.py # 基础智能体类 │ │ ├── researcher.py │ │ └── writer.py │ ├── tools/ # 工具定义 │ │ ├── __init__.py │ │ ├── web_tools.py │ │ ├── data_tools.py │ │ └── custom_tools.py │ ├── memory/ # 记忆后端Redis, Postgres等 │ │ ├── __init__.py │ │ └── redis_manager.py │ ├── workflows/ # LangGraph 工作流定义 │ │ ├── __init__.py │ │ └── research_workflow.py │ └── utils/ # 辅助函数 │ ├── __init__.py │ ├── logger.py │ └── validation.py └── scripts/ # 部署、数据迁移等脚本 └── deploy.sh关键点配置外置将LLM API密钥、模型名称、温度等参数通过环境变量或配置文件管理切勿硬编码。提示词即代码将复杂的提示词模板从Python代码中分离出来存为YAML或JSON文件便于版本控制和A/B测试。工具模块化每个工具功能单一便于单独测试和复用。工作流独立每个LangGraph工作流是一个独立的模块清晰定义输入输出。4.2 团队协作流程AI智能体项目是典型的跨职能项目需要建立新的协作规范。角色定义AI工程师/提示词工程师负责设计智能体工作流、优化提示词、集成工具和模型。后端工程师负责提供稳定的工具API如数据库查询、内部服务调用、部署智能体服务、保障系统性能和可靠性。前端工程师负责构建用户与智能体交互的界面如聊天界面、仪表盘。产品经理定义智能体的能力边界、用户体验和成功指标。测试工程师设计针对智能体输出稳定性、工具调用正确性的评估Evals用例。开发流程需求细化明确智能体的目标、可用工具、交互协议和评估标准。提示词开发与版本控制像管理代码一样管理提示词使用Git进行版本跟踪建立提示词评审机制。工具开发先行确保所有工具都有明确的接口、完善的错误处理和单元测试。智能体的可靠性很大程度上依赖于工具的可靠性。集成测试与评估Evals建立自动化测试流水线不仅测试代码功能更要评估智能体在多样本输入下的输出质量、安全性和稳定性。可以使用langsmith或trulens等平台。代码审查智能体逻辑、提示词、工具代码都需要经过同行审查。文档为每个智能体、工具和工作流编写清晰的文档说明其目的、输入输出、以及如何扩展。5. 常见问题与排查思路在开发和运行智能体时你可能会遇到以下典型问题问题现象可能原因排查与解决思路智能体不调用工具直接回答1. 提示词未明确要求使用工具。2. 工具描述不够清晰。3. LLM温度temperature过高导致创造性过强而忽略工具。1. 检查系统提示词加入“你必须使用提供的工具来回答问题”等指令。2. 优化工具函数的description和参数Field的description使其更精确。3. 尝试降低temperature如设为0。工具调用参数错误1. LLM未能正确解析用户意图为工具参数。2. 工具参数Schema定义太复杂或模糊。1. 在提示词中提供更清晰的示例Few-shot。2. 简化工具参数使用更明确的类型和描述。使用Pydantic进行严格验证。LangGraph工作流陷入循环状态State中的next_step逻辑有误或条件边Conditional Edge判断函数逻辑错误。1. 打印或记录每个节点执行后的状态。2. 检查should_continue或类似的路由函数确保所有可能的状态都有明确的出口指向下一个节点或END。3. 可以为图设置最大循环次数checkpointer配置。智能体响应慢1. LLM API调用延迟高。2. 工具本身是慢操作如网络请求、复杂计算。3. 智能体进行了不必要的多步推理。1. 考虑使用更快的模型如gpt-4o-mini或配置合理的超时。2. 为慢工具设置异步调用或增加缓存。3. 优化提示词引导智能体更直接地规划行动。生产环境内存/状态管理问题默认的内存可能基于内存在多实例部署下状态无法共享或会丢失。1. 为LangGraph配置持久化检查点Checkpointer如使用Redis、PostgreSQL作为后端。2. 对于聊天历史使用外部存储数据库、矢量库而非单纯的内存列表。6. 进阶主题与最佳实践6.1 评估Evals与监控“如何知道智能体工作得好不好” 这是AI工程的核心问题。你需要建立评估体系。单元测试针对工具确保每个工具函数在各种边界条件下都能正确运行和返回。集成测试针对工作流模拟端到端的用户输入验证最终输出是否符合预期。基于LLM的评估使用另一个LLM评判员来评估智能体输出的相关性、准确性、有用性和安全性。LangSmith提供了强大的工具来追踪Trace、评估和比较不同提示词或智能体版本的表现。监控与日志记录每一次智能体运行的完整轨迹Trace包括用户输入、中间步骤、工具调用、LLM请求/响应、最终输出。这对于调试和优化至关重要。6.2 安全与合规工具权限为智能体配置最小权限原则。例如一个总结文档的智能体不应该有删除数据库的权限。输入输出过滤对用户输入和智能体输出进行内容安全过滤防止注入攻击或生成有害内容。数据隐私明确哪些数据会发送给外部LLM API确保符合数据隐私法规如GDPR。对于敏感数据考虑使用本地部署的模型。人机回环Human-in-the-loop对于关键操作如发送邮件、发布内容、支付设计审批流程让人类拥有最终决定权。6.3 性能与成本优化缓存对频繁且结果不变的LLM请求或工具调用结果进行缓存。模型选择根据任务复杂度选择合适的模型。简单的分类任务可能不需要gpt-4gpt-3.5-turbo可能更经济高效。提示词优化精简提示词移除不必要的上下文可以有效降低Token消耗和延迟。异步处理对于耗时长的智能体任务采用异步处理模式通过回调或轮询告知用户结果。构建和维护AI智能体系统是一个持续迭代的过程。它要求开发者不仅要有软件工程的扎实功底还要对LLM的能力和局限有深刻理解。从组织好你的代码库开始建立清晰的团队协作规范注重测试和评估你就能稳步地将AI智能体从炫酷的概念转化为驱动业务价值的可靠引擎。

AI智能体工程化实战：基于LangGraph构建多智能体协作系统

相关推荐

高性能分布式KV存储引擎RocksDB入门与C/C++编码实战

Yuki第007个开关：骰子与猜拳结果设置的位置、验证方法与公平互动边界

Linux Mint SSH服务器安装配置全攻略：从基础安装到安全加固

更多资讯

两个月AI辅助学习实战：LLM、Agent、Token与API踩坑心得

跨芯片算子优化实战：FlagOS 如何让 Triton GEMM 反超原生库

教师AI工具选型指南：聊天、画图、视频、数字人四类工具教学应用与避坑

企业Wiki智能化改造：AI技术提升知识管理效率

macOS 27本地AI引擎fmkit：系统级Foundation Model运行时详解

SLDS 2.0分布式架构演进：物流系统性能优化与容灾实践

今日更新

PLFM_RADAR：从被动告警到主动感知的微服务健康监控系统

SPI隔离设计实战：MAX14431BAEE+与STM32F407工业级应用

数字隔离器如何根治SPI通信地环路干扰：MAX14431与TM4C129实战

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？