在上一篇文章中我们探讨了大语言模型LLM工程的基础包括模型选择、提示工程和基础应用架构。然而要构建真正强大、可靠且可投入生产的AI应用我们还需要深入工程化的核心领域。本文将聚焦于LLM工程的高级实践涵盖智能体Agent开发、复杂工作流编排、生产环境部署与监控以及应对AI幻觉等关键挑战。无论你是希望将原型转化为产品的开发者还是寻求构建复杂自主系统的工程师本文都将提供一套从理论到实战的完整指南。1. 智能体Agent开发从概念到实现智能体是LLM工程中最激动人心的方向之一。它让模型从一个被动的问答工具转变为一个能够感知环境、规划步骤、使用工具并执行任务的自主系统。1.1 智能体的核心架构与组件一个典型的智能体系统由以下几个核心组件构成规划模块Planner负责将复杂任务分解为可执行的子任务序列。这通常通过思维链Chain-of-Thought或更高级的思维树Tree of Thoughts等技术实现。记忆模块Memory存储智能体与环境的交互历史、学到的知识以及任务上下文。分为短期记忆对话历史和长期记忆向量数据库等。工具使用模块Tool Use赋予智能体调用外部API、查询数据库、执行代码等能力。这是智能体与真实世界交互的关键。行动执行模块Action根据规划选择并执行具体的工具调用。观察与反思模块Observation Reflection评估行动结果判断任务是否完成或在失败时进行反思并调整计划。1.2 使用 LangChain 和 LangGraph 构建智能体当前LangChain和LangGraph是构建智能体和工作流最流行的框架之一。LangChain提供了丰富的组件而LangGraph擅长描述有状态、多分支的工作流。下面我们构建一个能够联网搜索并总结信息的智能体。环境准备确保已安装必要库。建议使用虚拟环境。pip install langchain langchain-openai langchain-community langgraph beautifulsoup4第一步定义工具我们为智能体定义一个搜索工具。# 文件tools/search_tool.py import requests from langchain.tools import tool from bs4 import BeautifulSoup tool def search_web(query: str) - str: 使用DuckDuckGo的HTML接口进行简单网页搜索并提取摘要。仅供示例生产环境请使用官方API。 try: # 注意这是一个简化的示例。实际应使用可靠的搜索API如SerperAPI、Google Custom Search。 url fhttps://html.duckduckgo.com/html/?q{requests.utils.quote(query)} headers {User-Agent: Mozilla/5.0} response requests.get(url, headersheaders, timeout10) soup BeautifulSoup(response.text, html.parser) # 提取第一个结果片段 result_snippets soup.find_all(a, class_result__snippet) if result_snippets: return result_snippets[0].get_text(stripTrue)[:500] # 限制长度 else: return 未找到相关信息。 except Exception as e: return f搜索过程中发生错误{e}第二步创建智能体状态与节点使用LangGraph定义智能体的状态图和节点。# 文件agent_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from tools.search_tool import search_web # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 消息历史 task: str # 原始任务 plan: List[str] # 任务计划 current_step: int # 当前步骤索引 final_answer: str # 最终答案 # 2. 初始化模型和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用合适的模型 tools [search_web] llm_with_tools llm.bind_tools(tools) # 3. 定义各个节点函数 def planner_node(state: AgentState) - AgentState: 规划节点将复杂任务分解为步骤。 if state.get(plan): return state # 如果已有计划则跳过 planner_prompt f 你是一个任务规划专家。请将以下用户任务分解为2-4个清晰的、可执行的步骤。 每个步骤应该是一个简单的动作例如“搜索关于XX的信息”或“总结找到的内容”。 任务{state[task]} 请只输出步骤列表每行一个步骤不要有其他文字。 plan_message [SystemMessage(content你是一个规划助手。), HumanMessage(contentplanner_prompt)] plan_response llm.invoke(plan_message) plan [step.strip() for step in plan_response.content.split(\n) if step.strip()] return {**state, plan: plan, current_step: 0} def actor_node(state: AgentState) - AgentState: 执行节点根据当前步骤决定调用工具或直接回答。 current_step_index state[current_step] if current_step_index len(state[plan]): # 所有步骤已完成跳转到总结 return {**state, messages: state[messages] [AIMessage(content所有步骤执行完毕准备总结。)]} current_step state[plan][current_step_index] # 构建给LLM的提示包含历史、当前步骤和工具描述 context f 当前任务{state[task]} 当前步骤 ({current_step_index 1}/{len(state[plan])}): {current_step} 历史对话 {chr(10).join([msg.content for msg in state[messages][-3:]])} # 最近3条消息 human_msg HumanMessage(contentcontext) # 调用绑定了工具的LLM ai_msg llm_with_tools.invoke([human_msg]) state[messages].append(ai_msg) # 检查AI消息是否包含工具调用 if hasattr(ai_msg, tool_calls) and ai_msg.tool_calls: # 在实际应用中这里会解析并执行工具调用然后将结果作为Observation添加回消息历史 # 为简化示例我们假设调用了搜索工具 tool_call ai_msg.tool_calls[0] if tool_call[name] search_web: query tool_call[args][query] result search_web.invoke(query) # 将工具执行结果作为观察消息加入历史 observation_msg AIMessage(contentf[工具调用结果] {result}) state[messages].append(observation_msg) # 更新步骤索引 new_step_index current_step_index 1 return {**state, current_step: new_step_index} def summarizer_node(state: AgentState) - AgentState: 总结节点汇总所有执行结果生成最终答案。 # 提取所有消息中的关键信息简化处理 conversation_context \n.join([msg.content for msg in state[messages] if isinstance(msg, (HumanMessage, AIMessage))]) summary_prompt f 基于以下任务执行过程中的所有对话和工具返回结果生成一个完整、简洁、准确的最终答案来回答用户最初的任务。 最初任务{state[task]} 执行过程记录 {conversation_context} 最终答案 summary_message [HumanMessage(contentsummary_prompt)] summary_response llm.invoke(summary_message) final_answer summary_response.content return {**state, final_answer: final_answer} def router_node(state: AgentState) - str: 路由节点决定下一个节点是继续执行还是总结。 if state[current_step] len(state.get(plan, [])): return actor # 继续执行步骤 else: return summarizer # 去总结 # 4. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(actor, actor_node) workflow.add_node(summarizer, summarizer_node) # 设置边和条件路由 workflow.set_entry_point(planner) workflow.add_edge(planner, actor) workflow.add_conditional_edges( actor, router_node, # 下一个节点由router_node的返回值决定 { actor: actor, # 返回actor则循环执行actor节点 summarizer: summarizer } ) workflow.add_edge(summarizer, END) # 编译图 app workflow.compile()第三步运行智能体# 文件run_agent.py from agent_graph import app from langchain_core.messages import HumanMessage # 初始化状态 initial_state { messages: [HumanMessage(content请帮我了解2024年人工智能在医疗领域的最新突破有哪些)], task: 了解2024年人工智能在医疗领域的最新突破, plan: [], current_step: 0, final_answer: } # 运行图 final_state app.invoke(initial_state) print(*50) print(最终答案) print(final_state[final_answer]) print(*50)这个示例展示了智能体的基本循环规划 - 执行可能使用工具- 观察 - 路由 - 直到完成 - 总结。在实际项目中你需要处理更复杂的工具调用、错误处理、记忆管理等。2. 构建复杂工作流LangGraph 与 MCP 模型上下文协议对于更复杂、多分支、需持久化状态的企业级应用LangGraph结合MCP是强大的解决方案。2.1 LangGraph 核心概念状态State一个贯穿工作流始终的共享数据结构通常使用TypedDict定义。节点Node执行具体业务逻辑的函数接收和返回状态。边Edge定义节点之间的流转关系可以是固定的也可以是条件性的。图Graph由节点和边组成描述了完整的业务流程。2.2 集成 MCP 以扩展工具能力MCP是一种新兴的协议旨在标准化LLM与外部工具/数据源的交互方式。它允许你将数据库、API、文件系统等资源作为“工具”暴露给LLM而无需为每个资源编写特定的适配器代码。示例使用 MCP 服务器连接 PostgreSQL 数据库首先你需要一个 MCP 服务器。假设我们使用一个社区提供的 PostgreSQL MCP 服务器。启动 MCP 服务器通常通过命令行或Docker# 示例具体命令取决于MCP服务器实现 docker run -p 8080:8080 -e DB_URLpostgresql://user:passhost/db mcp-postgres-server在 LangChain/LangGraph 中连接 MCP 服务器# 文件mcp_integration.py (概念性代码) # 注意LangChain对MCP的官方集成可能在快速演进中以下展示概念 from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_mcp import MCPServer # 假设的库 # 1. 连接到MCP服务器 mcp_server MCPServer(urlhttp://localhost:8080) # 2. 获取MCP服务器暴露的所有工具 mcp_tools mcp_server.get_tools() # 3. 将这些工具提供给智能体 llm ChatOpenAI(modelgpt-4, temperature0) agent create_openai_tools_agent(llm, mcp_tools, prompt) agent_executor AgentExecutor(agentagent, toolsmcp_tools, verboseTrue) # 4. 现在智能体就可以使用类似“查询上季度销售额”、“插入一条用户反馈”等自然语言指令来操作数据库了。通过 MCP智能体可以动态发现并使用大量预定义的工具极大增强了其与复杂企业系统交互的能力。3. 生产环境部署与监控将LLM应用部署到生产环境面临着与标准软件不同的挑战如高延迟、高成本、非确定性输出等。3.1 部署模式与架构API 服务化使用 FastAPI、Flask 或专为ML服务的框架如 Ray Serve、BentoML将模型封装为REST或gRPC API。# 使用 FastAPI 的简单示例 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from your_agent_module import app as agent_graph # 导入之前编译的LangGraph应用 app FastAPI(titleLLM Agent API) class QueryRequest(BaseModel): question: str app.post(/ask) async def ask_agent(request: QueryRequest): try: initial_state { messages: [HumanMessage(contentrequest.question)], task: request.question, plan: [], current_step: 0, final_answer: } result agent_graph.invoke(initial_state) return {answer: result[final_answer]} except Exception as e: raise HTTPException(status_code500, detailstr(e))异步与流式响应对于长耗时任务使用异步端点async/await和服务器发送事件Server-Sent Events, SSE返回流式结果提升用户体验。容器化与编排使用 Docker 容器化应用并通过 Kubernetes 进行编排实现弹性伸缩、滚动更新和高可用。# Dockerfile 示例 FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]3.2 性能、成本与监控性能优化缓存对频繁且结果确定的提示词如系统指令、模板和相似查询的结果进行缓存可以使用Redis或Memcached。批处理对于非实时任务将多个请求合并为一个批处理调用模型API可以显著降低平均延迟和成本。模型蒸馏与量化考虑使用更小的模型如经过蒸馏的模型或对模型进行量化如使用bitsandbytes库以减少内存占用和推理时间。成本控制使用分层模型简单任务使用廉价、快速的小模型如gpt-3.5-turbo复杂任务再路由到强大但昂贵的大模型如gpt-4。设置预算与限流在API网关层面设置调用频率限制和月度预算告警。监控Token使用量详细记录每次调用的输入/输出Token数分析成本分布。可观测性监控日志结构化记录每个请求的输入、输出、使用的模型、Token数、耗时、工具调用详情和任何错误。指标监控QPS、延迟P50, P95, P99、错误率、Token消耗速率。追踪使用 OpenTelemetry 等工具对单个用户请求在智能体工作流中的完整路径进行追踪便于调试复杂问题。LLM特定监控输出质量通过规则如是否包含敏感词或小型评估模型对输出进行打分。幻觉检测对关键事实陈述可调用验证工具如二次搜索进行交叉验证。4. 应对核心挑战幻觉、安全与评估4.1 缓解幻觉策略幻觉是LLM生成看似合理但不符合事实或输入内容的信息。以下是一些工程化缓解手段检索增强生成RAG这是最有效的抗幻觉手段之一。确保模型生成答案时严格基于你提供的检索到的上下文。# 关键在提示词中强制模型引用上下文 rag_prompt 请严格根据以下提供的上下文信息来回答问题。如果上下文不包含回答问题所需的信息请直接说“根据提供的信息我无法回答此问题”。 上下文 {context} 问题{question} 答案 自我验证与一致性检查反向提问让模型根据自己生成的答案提出几个验证性问题然后检查原始上下文是否能回答这些问题。多轮采样让模型多次生成答案然后通过投票或选择最一致的部分作为最终输出。后处理与过滤使用命名实体识别NER提取答案中的关键实体人物、地点、组织并与可信知识库进行验证。对数字、日期、统计类信息使用正则表达式提取并与上下文核对。4.2 安全与合规输入/输出过滤输入清洗过滤用户输入中的恶意指令、敏感个人信息PII。输出审查对模型输出进行内容安全审查防止生成暴力、仇恨、违法或偏见性内容。可以使用内容安全API或本地分类器。权限与访问控制确保智能体只能访问其被授权使用的工具和数据源。在工具调用层实施严格的权限校验。数据隐私避免在提示词中泄露用户隐私或公司敏感数据。对用于微调或嵌入模型的数据进行脱敏处理。4.3 系统化评估构建评估体系是迭代改进LLM应用的关键。评估维度忠实度答案是否与提供的事实来源一致对抗幻觉相关性答案是否直接回答了问题完整性答案是否涵盖了问题的所有方面安全性答案是否无害、无偏见、合规代码执行如果涉及代码生成代码是否可运行且正确评估方法基于规则的评估检查输出是否包含特定关键词、是否符合格式。基于模型的评估使用另一个LLM如GPT-4作为裁判根据评分标准对答案进行打分。这被称为“LLM-as-a-Judge”。人工评估对于关键场景建立人工评估流程收集高质量反馈数据。构建评估流水线使用像Ragas、DeepEval、LangSmith这样的框架自动化地运行评估数据集生成评估报告追踪模型性能变化。5. 进阶主题与最佳实践5.1 智能体记忆设计向量存储长期记忆将重要的对话摘要、学到的用户偏好、任务结果存入向量数据库如Chroma,Weaviate,Pinecone供未来检索。摘要式记忆在对话轮次过长时让模型自动生成当前对话的摘要用摘要替代部分旧历史以节省上下文窗口。记忆索引为记忆建立多级索引如按时间、主题、实体提高检索效率。5.2 工具抽象与版本管理将工具定义为清晰的接口与具体实现解耦。为工具提供版本号当工具API变更时可以平滑升级智能体而不影响现有功能。建立工具注册中心方便智能体动态发现和调用。5.3 测试策略单元测试测试单个工具函数、提示词模板、解析逻辑。集成测试测试智能体在模拟环境中的完整工作流。端到端测试使用代表性用户问题测试整个系统并断言关键输出。模糊测试与对抗测试输入各种边缘案例、无意义或恶意提示检验系统的鲁棒性和安全性。5.4 持续学习与迭代收集生产数据在遵守隐私政策的前提下匿名化收集用户与智能体的交互日志。识别失败模式定期分析日志找出常见的错误类型如工具调用失败、幻觉、用户不满意。创建改进闭环根据失败模式针对性改进提示词、工具设计、工作流或模型选择并将改进部署到A/B测试环境中验证效果。大语言模型工程是一个融合了软件工程、机器学习、人机交互和数据管理的交叉领域。从构建一个简单的提示链到部署一个健壮、安全、可扩展的智能体系统每一步都需要严谨的工程化思维。本文涵盖了从智能体开发、工作流编排到生产部署和评估的核心路径。真正的精通来自于实践建议你从一个明确的小项目开始比如一个能帮你管理日程的智能体或一个基于公司文档的问答机器人在实践中不断遇到和解决本文提到的各种挑战逐步构建起你自己的LLM工程知识体系。记住保持对新技术如MCP的关注并始终将系统的可靠性、安全性和用户体验放在首位。