基于NirDiamant/agents-towards-production项目的LangSmith可观测性实践指南

基于NirDiamant/agents-towards-production项目的LangSmith可观测性实践指南【免费下载链接】agents-towards-productionEnd-to-end, code-first tutorials for building production-grade GenAI agents. From prototype to enterprise deployment.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-towards-production概述在AI应用开发领域构建一个能工作的原型相对容易但要确保系统在生产环境中稳定运行并具备可调试性则充满挑战。本文将通过NirDiamant/agents-towards-production项目中的实践案例展示如何利用LangSmith为基于LangGraph构建的AI系统添加全面的可观测性能力。为什么可观测性至关重要现代AI系统往往在演示阶段表现良好但在实际部署后却难以调试和优化。缺乏对决策过程的可见性开发团队会面临一系列关键问题决策透明度为什么AI会生成特定的响应性能瓶颈系统的哪些部分运行缓慢或成本高昂持续优化如何基于实际使用模式系统性地改进性能工具轨迹哪些工具调用路径最有效且成本最优可观测性就像为AI系统安装飞行记录器它能捕获每个决策点、执行时间和数据流将AI开发从猜测游戏转变为真正的工程实践。环境准备与配置基础环境要求Python 3.9 环境OpenAI API密钥用于访问语言模型LangSmith账户提供可观测性仪表板# 安装必要的开发包 !pip install -U langchain-core langchain-openai langgraph langsmith requestsAPI密钥配置import os # 配置API密钥实际使用时应替换为真实密钥 os.environ[OPENAI_API_KEY] your_openai_key os.environ[LANGCHAIN_API_KEY] your_langsmith_key os.environ[LANGCHAIN_TRACING_V2] true # 启用可观测性 os.environ[LANGCHAIN_PROJECT] langsmith-tutorial-demo # 指定追踪项目名称 # 验证配置 required_vars [OPENAI_API_KEY, LANGCHAIN_API_KEY] for var in required_vars: if not os.getenv(var): print(f警告: {var} 需要配置有效密钥)设置LANGCHAIN_TRACING_V2true会自动启用对所有操作的全面日志记录这相当于为AI系统安装了飞行记录器。构建可观测的智能体系统定义智能体状态我们首先定义一个简单的状态结构它将贯穿整个工作流程from typing import TypedDict class AgentState(TypedDict): 智能体的状态数据结构 user_question: str # 用户原始问题 needs_search: bool # 是否需要搜索 search_result: str # 搜索结果如果使用 final_answer: str # 最终响应 reasoning: str # 决策理由对可观测性非常重要这个结构化状态使LangSmith能够跟踪信息流每个字段都有特定的可观测性目的。实现搜索工具我们创建一个基于Wikipedia搜索API的工具tool def wikipedia_search(query: str) - str: 在Wikipedia上搜索关于主题的最新信息 try: # 使用Wikipedia搜索API处理通用查询 search_url https://en.wikipedia.org/w/api.php search_params { action: query, list: search, srsearch: query, format: json, srlimit: 3 # 获取前3个结果 } response requests.get(search_url, paramssearch_params, timeout10) if response.status_code 200: data response.json() search_results data.get(query, {}).get(search, []) if search_results: # 获取最相关结果并提取摘要 top_result search_results[0] page_title top_result[title] # 使用精确标题获取页面摘要 summary_url fhttps://en.wikipedia.org/api/rest_v1/page/summary/{page_title.replace( , _)} summary_response requests.get(summary_url, timeout10) if summary_response.status_code 200: summary_data summary_response.json() extract summary_data.get(extract, 无可用摘要) return f找到关于{page_title}的信息: {extract[:400]}... else: return f找到{page_title}但无法获取摘要 else: return f没有找到关于{query}的Wikipedia文章 else: return fWikipedia搜索失败状态码{response.status_code} except Exception as e: return f搜索错误: {str(e)}构建决策工作流步骤1决定是否需要搜索def decide_search_need(state: AgentState) - AgentState: 分析问题并决定是否需要搜索最新信息 user_question state[user_question] decision_prompt f 分析这个问题并判断是否需要不在你训练数据中的最新/近期信息: 问题: \{user_question}\ 考虑: - 是否询问近期事件、当前价格或突发新闻? - 是否询问经常变化的人物、公司或主题? - 你能否利用现有知识很好地回答这个问题? 如果需要当前信息请准确回复SEARCH如果可以直接回答则回复DIRECT。 然后在新的一行简要解释你的推理过程。 response llm.invoke([SystemMessage(contentdecision_prompt)]) decision_text response.content.strip() # 解析响应 lines decision_text.split(\n) decision lines[0].strip() reasoning lines[1] if len(lines) 1 else 未提供推理 # 更新状态 state[needs_search] decision SEARCH state[reasoning] f决策: {decision}. 推理: {reasoning} print(f决策: {SEARCH if state[needs_search] else DIRECT} - {reasoning}) return state步骤2执行搜索如果需要def execute_search(state: AgentState) - AgentState: 如果需要则执行搜索否则跳过此步骤 if not state[needs_search]: print(跳过搜索 - 此问题不需要) state[search_result] 未执行搜索 return state print(f执行搜索: {state[user_question]}) # 执行搜索工具 - 在LangSmith中会显示为独立步骤 search_result wikipedia_search.invoke({query: state[user_question]}) state[search_result] search_result print(f搜索完成: 返回{len(search_result)}个字符) return state步骤3生成最终响应def generate_response(state: AgentState) - AgentState: 生成最终响应结合所有可用信息 user_question state[user_question] if state[needs_search]: # 使用搜索结果的响应模板 prompt f 你被问到: \{user_question}\ 你决定搜索最新信息并获得了这些结果: {state[search_result]} 请提供一个有帮助的、准确的回答引用你找到的信息。 保持简洁专业。 else: # 直接回答的模板 prompt f 你被问到: \{user_question}\ 你确定这个问题可以使用你的现有知识回答。 请提供一个有帮助的、准确的回答。 保持简洁专业。 response llm.invoke([SystemMessage(contentprompt)]) state[final_answer] response.content print(响应生成完成) return state组装工作流图from langgraph.graph import StateGraph, END # 创建工作流 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(decide_search, decide_search_need) workflow.add_node(execute_search, execute_search) workflow.add_node(generate_response, generate_response) # 定义边 workflow.add_edge(decide_search, execute_search) workflow.add_edge(execute_search, generate_response) workflow.add_edge(generate_response, END) # 设置入口点 workflow.set_entry_point(decide_search) # 编译工作流 agent workflow.compile()运行与观测现在我们可以运行智能体并观察LangSmith中的追踪# 运行智能体 result agent.invoke({ user_question: 2023年诺贝尔物理学奖获得者是谁?, needs_search: None, search_result: None, final_answer: None, reasoning: None }) print(\n最终回答:) print(result[final_answer])在LangSmith仪表板中你将看到完整的执行轨迹包括决策节点的输入/输出搜索工具的执行详情响应生成过程每个步骤的执行时间和顺序可观测性最佳实践结构化状态设计确保状态包含足够的信息来理解系统行为清晰的节点边界将功能分解为离散的、可观测的步骤详细的推理记录捕获AI的决策过程而不仅仅是最终结果错误处理确保错误信息对调试有帮助性能指标关注关键步骤的执行时间通过遵循这些模式你可以构建不仅功能强大而且易于监控和调试的AI系统。LangSmith提供的可观测性能力使开发团队能够快速识别问题、优化性能并理解用户与系统的交互方式。【免费下载链接】agents-towards-productionEnd-to-end, code-first tutorials for building production-grade GenAI agents. From prototype to enterprise deployment.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-towards-production创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考