尧图网站建设 
在构建企业级AI应用时我们常常面临一个核心矛盾如何将前沿的AI模型如大语言模型稳定、可靠、可观测地集成到复杂的业务流程中尤其是在财务分析这类对准确性、可追溯性和合规性要求极高的领域。传统的“脚本API调用”模式在开发效率、监控评估和迭代优化上捉襟见肘导致项目难以规模化落地。本文将围绕Harness架构与Langfuse这两大AI工程化核心工具手把手带你构建一个企业级的“财务分析智能体”项目。这不是一个简单的Demo而是一个覆盖从智能体开发、流程编排、全面评估到生产部署全链路的实战教程。无论你是希望将AI能力引入现有系统的后端工程师还是专注于AI应用落地的算法工程师都能从中获得一套可直接复用的工程化方案。1. 项目背景与核心概念解析在深入代码之前我们必须厘清几个关键概念理解它们为何能解决企业级AI应用的痛点。1.1 什么是AI工程化AI工程化是将机器学习/人工智能模型的开发、部署、监控和维护过程系统化、标准化和自动化的实践。它旨在弥合数据科学实验与生产级软件交付之间的鸿沟。对于基于大语言模型LLM的应用工程化挑战尤为突出包括提示词Prompt管理、上下文窗口处理、多步骤推理Agent流程编排、成本控制、性能评估与迭代等。1.2 HarnessAI智能体的编排与执行引擎Harness在这里并非指持续交付工具Harness.io而是指一个新兴的、专注于AI智能体Agent工作流编排的开源框架注根据网络热词可能与DeepSeek等探索相关。我们可以将其理解为一个专为AI设计的“工作流引擎”。它的核心价值在于可视化/代码化编排允许你通过拖拽或代码定义复杂的、多步骤的AI任务流程例如“获取数据 - 分析 - 生成报告 - 发送审核”。状态管理与回溯自动维护智能体执行过程中的状态State方便调试和错误恢复。工具集成便捷地集成外部工具如计算器、数据库查询、API调用扩展智能体的能力边界。抽象底层LLM提供统一的接口调用不同的模型提供商OpenAI, Anthropic, 本地模型等降低耦合。简单说Harness让构建一个像“财务分析师”一样执行多步骤任务的智能体变得像搭积木一样清晰可控。1.3 LangfuseLLM应用的观测与评估平台Langfuse是一个开源的LLM应用观测平台。如果说Harness是“生产车间”那么Langfuse就是“质量检测与监控中心”。它的核心功能包括全链路追踪Tracing自动记录每次LLM调用的输入、输出、延迟、成本、token用量并可视化整个调用链。提示词管理Prompt Management版本化管理和评估不同的提示词模板实现A/B测试。评估与评分Evaluation支持基于规则如格式检查、模型使用另一个LLM打分或人工的反馈对AI输出进行量化评估。数据分析看板提供丰富的仪表盘分析成本、延迟、评分趋势定位问题。在财务分析场景中Langfuse能帮助我们回答关键问题智能体生成的报告准确性如何分析逻辑是否稳定每次调用的成本是多少1.4 财务分析智能体我们的实战目标我们将构建一个智能体它能够处理一份结构化的财务数据如CSV格式的利润表并完成以下任务数据解读识别关键财务指标营收、毛利率、净利润等。趋势分析计算环比、同比增长率。洞察生成用自然语言总结财务状况指出亮点与风险。报告格式化输出结构清晰的Markdown格式报告。这个智能体将使用Harness来编排“数据读取 - 指标计算 - LLM分析 - 报告生成”的流程并利用Langfuse对每一次执行进行追踪、记录和评估。2. 环境准备与项目初始化我们将使用Python作为主要开发语言。请确保你的环境满足以下要求。2.1 基础环境要求操作系统macOS / Linux / Windows (WSL2推荐)Python版本3.10 或 3.11建议使用3.11以获得最佳兼容性包管理工具pip 或 poetryLLM API密钥你需要一个OpenAI API密钥或 Anthropic、Groq 等兼容OpenAI SDK的API密钥用于调用模型。本文以OpenAI GPT-4o-mini为例。2.2 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir finance-ai-agent cd finance-ai-agent python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来创建requirements.txt文件包含以下核心依赖# 核心AI与编排框架 openai1.0.0 # 假设我们使用一个名为 ai-harness 的模拟Harness框架包进行演示 # 注当前Harness框架生态中有多个项目如 harnessagent-harness等此处为概念演示我们使用一个简化的自制类。 langfuse3.0.0 # 数据处理与工具 pandas2.0.0 numpy1.24.0 # 环境变量管理 python-dotenv1.0.0 # 可选Web框架如需提供API接口 fastapi0.104.0 uvicorn0.24.0安装依赖pip install -r requirements.txt由于目前没有一个统一的、名为“Harness”的Python包我们将模拟其核心编排思想构建一个轻量级的Harness类。在真实项目中你可以根据调研选择如Semantic Kernel,LangChain,AutoGen或新兴的harnessSDK。2.3 配置环境变量创建.env文件存储敏感信息和配置# .env OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key LANGFUSE_HOSThttps://cloud.langfuse.com # 或你的自托管地址 # 项目配置 DEFAULT_LLM_MODELgpt-4o-mini重要请勿将.env文件提交到版本控制系统。确保它在.gitignore中。2.4 初始化Langfuse客户端在项目根目录创建config.py用于初始化全局配置和客户端。# config.py import os from dotenv import load_dotenv from langfuse import Langfuse # 加载环境变量 load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY) LANGFUSE_HOST os.getenv(LANGFUSE_HOST) DEFAULT_LLM_MODEL os.getenv(DEFAULT_LLM_MODEL, gpt-4o-mini) staticmethod def get_langfuse_client(): 初始化并返回Langfuse客户端实现单例模式 if not all([Config.LANGFUSE_PUBLIC_KEY, Config.LANGFUSE_SECRET_KEY]): print(警告: Langfuse 密钥未配置追踪功能将禁用。) return None try: # 在生产环境中建议配置更详细的初始化参数 langfuse_client Langfuse( public_keyConfig.LANGFUSE_PUBLIC_KEY, secret_keyConfig.LANGFUSE_SECRET_KEY, hostConfig.LANGFUSE_HOST, ) return langfuse_client except Exception as e: print(f初始化Langfuse客户端失败: {e}) return None # 全局配置和客户端实例 config Config() langfuse_client config.get_langfuse_client()3. 核心架构与模拟Harness引擎实现我们将实现一个简化的Harness类其核心思想是将复杂的AI任务分解为多个可复用的“节点”Node并通过“边”Edge定义执行顺序和数据流。3.1 定义节点基类与上下文首先定义任务执行时的上下文State和节点的基类。# harness/core.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional, Callable import inspect class State(Dict[str, Any]): 任务执行状态上下文是一个增强型字典。 pass class Node(ABC): Harness节点抽象基类。每个节点代表一个处理单元。 def __init__(self, id: str, description: str ): self.id id self.description description abstractmethod async def execute(self, state: State) - State: 执行节点的核心逻辑。 参数: state: 当前执行状态包含上游节点的输出。 返回: 更新后的状态。 pass def __repr__(self): return fNode(id{self.id})3.2 实现具体功能节点基于我们的财务分析场景我们需要几种类型的节点工具节点Tool Node执行确定性操作如读取文件、计算指标。LLM节点LLM Node调用大语言模型进行分析和生成。条件节点Condition Node根据状态决定执行分支本例暂不展开。让我们实现一个文件读取节点和一个指标计算节点。# harness/nodes/tool_nodes.py import pandas as pd from ..core import Node, State import json class LoadCSVNode(Node): 加载CSV财务数据文件到状态中。 def __init__(self, id: str, file_path_key: str file_path, output_key: str df): super().__init__(id, description加载CSV文件为DataFrame) self.file_path_key file_path_key # 状态中文件路径的键名 self.output_key output_key # 输出DataFrame的键名 async def execute(self, state: State) - State: file_path state.get(self.file_path_key) if not file_path: raise ValueError(f状态中未找到文件路径键: {self.file_path_key}) try: df pd.read_csv(file_path) state[self.output_key] df state[f{self.output_key}_shape] df.shape print(f[{self.id}] 已加载文件: {file_path}, 形状: {df.shape}) except Exception as e: raise RuntimeError(f[{self.id}] 加载CSV文件失败: {e}) return state class CalculateMetricsNode(Node): 计算基础财务指标。 def __init__(self, id: str, df_key: str df, metrics_key: str basic_metrics): super().__init__(id, description计算财务指标营收、毛利、净利等) self.df_key df_key self.metrics_key metrics_key async def execute(self, state: State) - State: df state.get(self.df_key) if df is None or not isinstance(df, pd.DataFrame): raise ValueError(f状态中未找到有效的DataFrame键: {self.df_key}) # 假设CSV有特定列名这里需要根据实际数据调整 # 示例列period, revenue, cost_of_goods_sold, operating_expenses, net_income required_cols [period, revenue, cost_of_goods_sold, net_income] if not all(col in df.columns for col in required_cols): raise ValueError(fDataFrame缺少必要列需要: {required_cols}) metrics {} latest df.iloc[-1] # 取最新一期数据 previous df.iloc[-2] if len(df) 1 else None metrics[latest_period] latest[period] metrics[revenue] float(latest[revenue]) metrics[gross_profit] float(latest[revenue] - latest[cost_of_goods_sold]) metrics[gross_margin] float(metrics[gross_profit] / latest[revenue]) if latest[revenue] ! 0 else 0 metrics[net_income] float(latest[net_income]) metrics[net_margin] float(latest[net_income] / latest[revenue]) if latest[revenue] ! 0 else 0 # 计算环比如果数据足够 if previous is not None: metrics[revenue_qoq] float((latest[revenue] - previous[revenue]) / previous[revenue]) if previous[revenue] ! 0 else 0 metrics[net_income_qoq] float((latest[net_income] - previous[net_income]) / previous[net_income]) if previous[net_income] ! 0 else 0 state[self.metrics_key] metrics print(f[{self.id}] 计算完成基础指标: {json.dumps(metrics, indent2, defaultstr)}) return state接下来实现一个与Langfuse深度集成的LLM节点。这个节点会将其调用过程自动记录到Langfuse。# harness/nodes/llm_nodes.py import openai from openai import OpenAI from ..core import Node, State from config import config, langfuse_client import json class LangfuseLLMNode(Node): 集成Langfuse追踪的LLM调用节点。 def __init__(self, id: str, prompt_template: str, input_state_keys: Dict[str, str], # 例如 {metrics: basic_metrics, df_summary: df_summary} output_key: str llm_response, model: str None, system_prompt: str 你是一个专业的财务分析师请根据提供的数据进行客观、严谨的分析。 ): super().__init__(id, description调用LLM并生成分析报告) self.prompt_template prompt_template self.input_state_keys input_state_keys # 映射模板变量名 - 状态中的键名 self.output_key output_key self.model model or config.DEFAULT_LLM_MODEL self.system_prompt system_prompt self.client OpenAI(api_keyconfig.OPENAI_API_KEY) async def execute(self, state: State) - State: # 1. 从状态中提取数据填充提示词模板 template_data {} for var_name, state_key in self.input_state_keys.items(): if state_key not in state: raise ValueError(f状态中缺少所需键 {state_key}用于模板变量 {var_name}) template_data[var_name] state[state_key] try: user_prompt self.prompt_template.format(**template_data) except KeyError as e: raise ValueError(f提示词模板变量替换失败缺少变量: {e}) # 2. 准备Langfuse追踪如果客户端可用 trace None generation None if langfuse_client: trace langfuse_client.trace( namefFinanceAnalysis-{self.id}, input{system_prompt: self.system_prompt, user_prompt_preview: user_prompt[:200]}, metadata{node_id: self.id, model: self.model}, ) generation trace.generation( nameFinancial Report Generation, modelself.model, model_parameters{temperature: 0.2, max_tokens: 1500}, input[{role: system, content: self.system_prompt}, {role: user, content: user_prompt}], ) # 3. 调用OpenAI API try: response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: self.system_prompt}, {role: user, content: user_prompt} ], temperature0.2, max_tokens1500, ) llm_output response.choices[0].message.content # 4. 记录成功结果到Langfuse if generation: generation.end(outputllm_output) trace.update(output{report: llm_output[:500]}) # 只记录预览 print(f[{self.id}] LLM调用成功结果已记录至Langfuse Trace: {trace.id}) else: print(f[{self.id}] LLM调用成功 (Langfuse未启用)) # 5. 将结果存入状态 state[self.output_key] llm_output # 也可以存储原始响应对象以备后用 state[f{self.output_key}_raw] response except Exception as e: # 6. 记录失败信息到Langfuse if generation: generation.end(errorstr(e)) trace.update(output{error: str(e)}) print(f[{self.id}] LLM调用失败: {e}) raise RuntimeError(fLLM节点执行失败: {e}) from e finally: # 确保Langfuse客户端刷新数据 if langfuse_client: langfuse_client.flush() return state3.3 实现Harness编排引擎现在我们将这些节点连接起来构建一个简单的线性执行引擎。# harness/engine.py from typing import List, Dict, Any from .core import Node, State import asyncio class Harness: 简单的线性工作流编排引擎。 def __init__(self, name: str FinanceAgentHarness): self.name name self.nodes: List[Node] [] self.state State() def add_node(self, node: Node): 向工作流添加一个节点。 self.nodes.append(node) return self # 支持链式调用 def set_initial_state(self, **kwargs): 设置工作流的初始状态。 self.state.update(kwargs) async def run(self) - State: 顺序执行所有节点。 print(f 开始执行工作流 [{self.name}] ) current_state self.state.copy() for i, node in enumerate(self.nodes): print(f\n[{i1}/{len(self.nodes)}] 执行节点: {node.id} ({node.description})) try: current_state await node.execute(current_state) except Exception as e: print(f!!! 节点 {node.id} 执行失败工作流终止。错误: {e}) # 可以将错误状态记录到Langfuse if langfuse_client: langfuse_client.trace( namefHarnessError-{self.name}, input{failed_node: node.id, state_snapshot: str(current_state)}, output{error: str(e)}, levelERROR ) langfuse_client.flush() raise print(f\n 工作流 [{self.name}] 执行完成 ) return current_state def visualize(self): 打印工作流的简单文本可视化。 print(f工作流: {self.name}) for i, node in enumerate(self.nodes): print(f {i1}. [{node.id}] - {node.description})4. 完整实战构建并运行财务分析智能体现在我们将所有部分组合起来创建一个完整的可执行示例。4.1 准备示例财务数据创建一个示例CSV文件data/sample_finance_data.csvperiod,revenue,cost_of_goods_sold,operating_expenses,net_income 2023-Q1,1000000,600000,250000,150000 2023-Q2,1200000,700000,280000,220000 2023-Q3,1100000,650000,260000,190000 2023-Q4,1300000,750000,300000,2500004.2 定义提示词模板与工作流创建主执行脚本main.py# main.py import asyncio import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from harness.engine import Harness from harness.nodes.tool_nodes import LoadCSVNode, CalculateMetricsNode from harness.nodes.llm_nodes import LangfuseLLMNode # 1. 定义分析报告提示词模板 FINANCIAL_ANALYSIS_PROMPT 你是一名资深财务分析师。请基于以下财务数据和分析指标生成一份简洁、专业的财务分析报告。 **基础数据概览** - 数据期间: {periods} - 总记录数: {record_count} **计算出的关键指标最新一期 {latest_period}** - 营业收入: {revenue:,.2f} - 毛利润: {gross_profit:,.2f} - 毛利率: {gross_margin:.2%} - 净利润: {net_income:,.2f} - 净利率: {net_margin:.2%} {trend_analysis} **请按以下结构组织报告** 1. **核心业绩摘要**用2-3句话总结本期核心财务表现。 2. **盈利能力分析**分析毛利率和净利率的水平及其含义。 3. **增长趋势洞察**结合环比数据评论营收和利润的增长趋势。 4. **潜在风险与关注点**基于数据指出1-2个可能的风险或需要关注的方面。 5. **后续建议**给出1-2条具体的后续分析或行动建议。 报告请使用Markdown格式确保数据准确分析客观。 async def main(): # 2. 初始化Harness工作流 harness Harness(nameQuarterlyFinancialAnalysis) # 3. 添加节点定义执行流水线 harness.add_node( LoadCSVNode(idload_data, file_path_keyinput_file, output_keydf) ).add_node( CalculateMetricsNode(idcalc_metrics, df_keydf, metrics_keybasic_metrics) ).add_node( LangfuseLLMNode( idgenerate_report, prompt_templateFINANCIAL_ANALYSIS_PROMPT, input_state_keys{ periods: periods_summary, # 这些键需要从状态中获取 record_count: record_count, latest_period: latest_period, revenue: revenue, gross_profit: gross_profit, gross_margin: gross_margin, net_income: net_income, net_margin: net_margin, trend_analysis: trend_text, }, output_keyanalysis_report, system_prompt你是一个严谨、客观的财务分析师擅长从数据中发现洞察并以结构化的方式呈现。, modelgpt-4o-mini # 或 gpt-3.5-turbo ) ) # 4. 准备初始状态并补充LLM节点需要的额外数据 # 注意LLM节点需要的数据有些需要前面的节点生成有些需要我们从原始数据中提取并放入状态。 # 我们可以在运行前设置一部分另一部分通过一个“数据准备节点”或直接在运行中计算后注入。 # 这里我们采用一个简单方法在运行工作流前先手动计算一些衍生数据放入状态。 # 更优雅的方式是创建一个 PrepareForLLMNode。 initial_state { input_file: data/sample_finance_data.csv, } harness.set_initial_state(**initial_state) # 5. 可视化工作流 print(构建的工作流如下) harness.visualize() # 6. 执行工作流 try: final_state await harness.run() except Exception as e: print(f\n工作流执行因错误中断: {e}) return # 7. 打印最终结果 print(\n *50) print(财务分析报告生成完成) print(*50) if analysis_report in final_state: print(final_state[analysis_report]) else: print(未生成报告。最终状态:, final_state.keys()) if __name__ __main__: asyncio.run(main())4.3 运行智能体并查看结果在终端运行python main.py你将看到类似以下的输出具体内容因模型随机性而异构建的工作流如下 工作流: QuarterlyFinancialAnalysis 1. [load_data] - 加载CSV文件为DataFrame 2. [calc_metrics] - 计算财务指标营收、毛利、净利等 3. [generate_report] - 调用LLM并生成分析报告 开始执行工作流 [QuarterlyFinancialAnalysis] [1/3] 执行节点: load_data (加载CSV文件为DataFrame) [load_data] 已加载文件: data/sample_finance_data.csv, 形状: (4, 5) [2/3] 执行节点: calc_metrics (计算财务指标营收、毛利、净利等) [calc_metrics] 计算完成基础指标: { latest_period: 2023-Q4, revenue: 1300000.0, gross_profit: 550000.0, gross_margin: 0.4230769230769231, net_income: 250000.0, net_margin: 0.19230769230769232, revenue_qoq: 0.18181818181818182, net_income_qoq: 0.3157894736842105 } [3/3] 执行节点: generate_report (调用LLM并生成分析报告) [generate_report] LLM调用成功结果已记录至Langfuse Trace: 01JXXXXXX 工作流 [QuarterlyFinancialAnalysis] 执行完成 财务分析报告生成完成 # 财务分析报告 ## 1. 核心业绩摘要 2023年第四季度公司实现营业收入130万元净利润25万元。毛利率为42.31%净利率为19.23%整体盈利能力保持稳健。 ## 2. 盈利能力分析 本期毛利率为42.31%表明公司产品或服务具有较强的市场竞争力成本控制有效。净利率达到19.23%在扣除运营费用后仍保持了较高的利润留存率整体盈利结构健康。 ## 3. 增长趋势洞察 与第三季度相比第四季度营收环比增长约18.18%净利润环比增长约31.58%增速显著。这反映出公司在年末可能采取了有效的市场策略或成本优化措施推动了利润的更快增长。 ## 4. 潜在风险与关注点 1. **成本压力**尽管毛利率可观但需持续关注原材料或直接成本销售成本的变动其占营收比例仍超过57%。 2. **增长可持续性**本季度的强劲增长是否具有可持续性还是受季节性因素影响需要结合更多历史数据和业务背景判断。 ## 5. 后续建议 1. **深入分析成本结构**建议对销售成本cost_of_goods_sold进行细分识别可优化的具体环节。 2. **制定2024年季度预算**基于2023年的增长趋势为2024年各季度设定合理的营收与利润目标并建立监控机制。4.4 在Langfuse平台查看追踪详情登录你的Langfuse账户云服务或自托管。进入“Traces”页面你应该能看到一条名为FinanceAnalysis-generate_report的追踪记录。点击进入可以看到完整的追踪详情输入系统提示词和用户提示词预览。输出生成的完整报告。元数据节点ID、使用的模型。分析Token使用量、成本、延迟。你可以在“Prompts”部分管理你的提示词模板并进行版本对比。在“Evaluations”部分可以基于本次输出创建评估例如让另一个LLM从“专业性”、“数据准确性”维度打分。5. 企业级评估平台搭建与AI工程化实践仅仅生成报告和记录追踪是不够的。企业级应用需要系统化的评估、监控和迭代能力。下面我们利用Langfuse构建一个简单的评估流程。5.1 定义自动化评估指标我们可以为财务报告定义几个自动化评估维度格式合规性报告是否包含要求的5个部分基于规则数据保真度报告中引用的数字是否与输入数据一致基于LLM或规则专业性评分报告的语言是否专业、客观基于LLM我们在项目中创建一个新的模块evaluation/evaluator.py# evaluation/evaluator.py import re from typing import Dict, Any from langfuse import Langfuse from config import langfuse_client class ReportEvaluator: def __init__(self): self.client langfuse_client def evaluate_format(self, report: str, required_sections: list) - Dict[str, Any]: 评估报告格式是否包含所有必需章节。 score 100 feedback [] missing_sections [] for section in required_sections: # 简单检查章节标题是否存在于报告中 pattern rf^#\s*{section}|^#\s*\d\.\s*{section} if not re.search(pattern, report, re.IGNORECASE | re.MULTILINE): missing_sections.append(section) score - 20 # 每缺一个章节扣20分 if missing_sections: feedback.append(f报告缺少以下必需章节: {, .join(missing_sections)}) else: feedback.append(报告格式完整包含所有必需章节。) return { score: max(0, score), feedback: ; .join(feedback), details: {missing_sections: missing_sections} } async def evaluate_data_fidelity(self, trace_id: str, original_metrics: Dict, report: str) - Dict[str, Any]: 使用LLM评估报告中的数据是否与原始指标一致。 if not self.client: return {score: -1, feedback: Langfuse客户端未初始化无法进行评估。} # 从Langfuse获取该Trace的完整生成记录 # 注意Langfuse Python SDK目前可能不直接支持通过ID查询Trace这里为演示逻辑。 # 实际应用中你可能需要在生成时保存更多上下文或使用Langfuse API。 # 此处简化为直接调用一个新的LLM进行评估。 from openai import OpenAI from config import config client OpenAI(api_keyconfig.OPENAI_API_KEY) evaluation_prompt f 你是一个严谨的数据审计员。请对比以下两组信息 **【原始财务指标】** {original_metrics} **【生成的财务分析报告】** {report} 请判断报告中引用的核心数据如营业收入、毛利率、净利润、增长率等是否与原始指标**完全一致**。 注意报告可能对数字进行格式化如添加千分位逗号或使用近似表述如“约42%”这不算错误。 请只检查是否存在**事实性矛盾**例如原始数据是130万报告写成120万。 请按以下格式回答 一致性结论: [是/否] 得分: (0-100分100分为完全一致) 不一致详情: [如果结论为“否”请列出具体不一致的数据点否则写“无”] try: response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: evaluation_prompt}], temperature0, max_tokens500, ) result_text response.choices[0].message.content # 简单解析结果实际应用需更健壮的解析 score 100 feedback 数据一致性评估完成。 if 一致性结论: 否 in result_text: score 50 # 假设不一致则给50分 feedback 检测到报告中的数据与原始指标存在潜在不一致。 # 可以将此次评估也记录到Langfuse关联到原始Trace if self.client and trace_id: self.client.score( trace_idtrace_id, namedata_fidelity, valuescore, commentfeedback, ) return {score: score, feedback: feedback, details: result_text} except Exception as e: return {score: -1, feedback: f数据一致性评估失败: {e}} def log_evaluation_to_langfuse(self, trace_id: str, evaluation_name: str, score: float, comment: str ): 将评估分数记录到Langfuse对应的Trace上。 if self.client and trace_id: try: self.client.score( trace_idtrace_id, nameevaluation_name, valuescore, commentcomment, ) print(f[评估已记录] Trace: {trace_id}, 指标: {evaluation_name}, 分数: {score}) except Exception as e: print(f记录评估分数到Langfuse失败: {e})5.2 集成评估到主工作流修改main.py在工作流执行后自动触发评估。# 在 main.py 的 main() 函数末尾打印报告后添加 # ... 原有打印报告的代码 ... # 8. 执行自动化评估 if analysis_report in final_state and langfuse_client: print(\n *50) print(开始自动化报告评估...) print(*50) evaluator ReportEvaluator() report final_state[analysis_report] # 评估1格式合规性 required_sections [核心业绩摘要, 盈利能力分析, 增长趋势洞察, 潜在风险与关注点, 后续建议] format_result evaluator.evaluate_format(report, required_sections) print(f格式评估得分: {format_result[score]}/100 - {format_result[feedback]}) # 记录分数到Langfuse (需要trace_id) trace_id 从final_state或全局变量中获取 # 实际需要从LLM节点执行时获取并传递 # evaluator.log_evaluation_to_langfuse(trace_id, format_compliance, format_result[score]) # 评估2数据保真度 (异步) # 需要获取原始指标和trace_id # original_metrics final_state.get(basic_metrics, {}) # fidelity_result await evaluator.evaluate_data_fidelity(trace_id, original_metrics, report) # print(f数据一致性评估得分: {fidelity_result[score]}/100 - {fidelity_result[feedback]}) print(自动化评估完成。请登录Langfuse平台查看详细追踪与评分。)5.3 构建评估看板与持续迭代在Langfuse平台上你可以创建Dashboard监控“报告生成延迟”、“平均格式得分”、“数据一致性得分”等核心指标。设置警报当评估分数低于阈值如格式分80时发送通知。Prompt版本管理在Langfuse的“Prompts”中迭代你的FINANCIAL_ANALYSIS_PROMPT并对比不同版本生成报告的质量。人工反馈集成在生成的报告旁提供“拇指向上/下”按钮收集人工反馈并将其作为评估信号。通过这一套组合你就建立了一个“开发Harness - 运行 - 观测与评估Langfuse - 迭代优化”的完整AI工程化闭环。6. 常见问题与排查思路在实际部署和运行中你可能会遇到以下问题问题现象可能原因排查思路与解决方案运行时报ModuleNotFoundError: No module named harness自定义的harness模块路径未正确加入Python路径。1. 确保在项目根目录下运行脚本。2. 在main.py开头使用sys.path.append添加项目根目录或使用PYTHONPATH环境变量。3. 检查__init__.py文件是否存在于harness/和harness/nodes/目录下。Langfuse追踪未显示在控制台1. API密钥错误或未配置。2. 网络问题。3. Langfuse客户端初始化失败。1. 检查.env文件中的LANGFUSE_PUBLIC_KEY和LANGFUSE_SECRET_KEY是否正确。2. 检查config.py中get_langfuse_client方法的错误打印。3. 尝试在代码中暂时禁用Langfuse确认基础功能正常。LLM节点调用超时或报错1. OpenAI API密钥无效或余额不足。2. 网络连接问题。3. 模型名称错误或不可用。1. 在OpenAI平台验证API密钥状态和额度。2. 尝试使用curl或openai库的简单测试脚本确认API连通性。3. 确认model参数是有效的模型名如gpt-3.5-turbo。提示词模板格式化报KeyErrorinput_state_keys中定义的键在节点执行时状态中不存在。1. 仔细检查input_state_keys的映射关系。确保上游节点如CalculateMetricsNode的输出键与映射中state_key一致。2. 在LangfuseLLMNode.execute方法开始处打印state.keys()进行调试。财务指标计算逻辑错误CSV文件列名与代码中required_cols不匹配或数据格式非数值。1. 打印加载后的DataFrame的columns和dtypes进行确认。2. 在CalculateMetricsNode中添加更严格的数据类型校验和转换。评估分数未关联到Tracetrace_id未正确传递到ReportEvaluator。1. 修改LangfuseLLMNode在执行成功后将trace.id存入state如state[trace_id] trace.id。2. 在工作流最终状态中将trace_id传递给评估器。7. 最佳实践与工程化建议将此类AI智能体项目投入生产环境需要遵循以下工程化实践7.1 项目结构与代码组织清晰的模块化正如本文所示将引擎、节点、配置、评估逻辑分离。nodes/目录下可按功能进一步细分如tools/,llms/,conditions/。配置外部化所有API密钥、模型参数、文件路径、提示词模板都应通过.env或配置中心如Apollo管理严禁硬编码。依赖管理使用requirements.txt或pyproject.toml精确锁定依赖版本避免环境差异导致运行失败。7.2 提示词工程与管理版本控制将提示词模板存储在代码库或Langfuse等专用平台中并进行版本控制。每次修改都应记录原因和预期效果。变量化与验证像本文一样使用明确的变量占位符{variable}并在填充前验证状态中是否存在对应数据避免运行时错误。系统提示词设计系统提示词system_prompt是塑造AI行为的关键。它应明确角色、任务边界、输出格式和禁忌。7.3 可观测性与监控全链路追踪务必为每个LLM调用、关键工具调用记录追踪。除了输入输出还应记录延迟、Token用量和成本。定义业务指标在Langfuse中创建与业务价值直接相关的评估指标Score如“报告格式合规分”、“数据准确分”、“用户满意度分”。通过看板持续监控。设置警报对错误率、高延迟、成本突增、评估分下降设置阈值告警。7.4 错误处理与韧性节点级容错每个Node的execute方法应有完善的try-except捕获可能异常并决定是向上抛出终止流程还是记录错误后返回降级结果。重试与退避对于LLM API调用等可能因网络或速率限制失败的操作实现带指数退避的重试机制。状态快照与回滚复杂的Harness引擎应支持将执行状态持久化在失败时可以从上一个成功节点恢复而不是从头开始。7.5 安全与合规数据脱敏财务数据高度敏感。在将数据发送给外部LLM API前必须进行严格的脱敏处理如替换真实公司名、金额缩放。对于极高敏感数据考虑使用本地模型。权限控制确保只有授权用户能触发智能体工作流并能访问生成的报告和原始数据。审计日志所有智能体的触发、输入、输出、执行人、时间都应记录到不可篡改的审计日志中满足合规要求。7.6 性能与成本优化缓存策略对于相同输入可能产生相同输出的LLM调用或计算节点考虑引入缓存如Redis显著降低成本和延迟。模型选型并非所有任务都需要最强大的模型。对于数据提取、格式转换等简单任务可使用更小、更快的模型如gpt-3.5-turbo。异步执行如果工作流中节点间没有强依赖可以将其设计为异步并行执行缩短整体耗时。通过本文的实战你不仅学会了一个财务分析智能体的构建更掌握了一套基于Harness架构思想与Langfuse评估平台的AI工程化落地方法论。这套方法论可以平移到客服、营销、运维、代码生成等几乎所有AI智能体应用场景。 基于Harness与Langfuse构建企业级AI智能体：财务分析实战
在构建企业级AI应用时我们常常面临一个核心矛盾如何将前沿的AI模型如大语言模型稳定、可靠、可观测地集成到复杂的业务流程中尤其是在财务分析这类对准确性、可追溯性和合规性要求极高的领域。传统的“脚本API调用”模式在开发效率、监控评估和迭代优化上捉襟见肘导致项目难以规模化落地。本文将围绕Harness架构与Langfuse这两大AI工程化核心工具手把手带你构建一个企业级的“财务分析智能体”项目。这不是一个简单的Demo而是一个覆盖从智能体开发、流程编排、全面评估到生产部署全链路的实战教程。无论你是希望将AI能力引入现有系统的后端工程师还是专注于AI应用落地的算法工程师都能从中获得一套可直接复用的工程化方案。1. 项目背景与核心概念解析在深入代码之前我们必须厘清几个关键概念理解它们为何能解决企业级AI应用的痛点。1.1 什么是AI工程化AI工程化是将机器学习/人工智能模型的开发、部署、监控和维护过程系统化、标准化和自动化的实践。它旨在弥合数据科学实验与生产级软件交付之间的鸿沟。对于基于大语言模型LLM的应用工程化挑战尤为突出包括提示词Prompt管理、上下文窗口处理、多步骤推理Agent流程编排、成本控制、性能评估与迭代等。1.2 HarnessAI智能体的编排与执行引擎Harness在这里并非指持续交付工具Harness.io而是指一个新兴的、专注于AI智能体Agent工作流编排的开源框架注根据网络热词可能与DeepSeek等探索相关。我们可以将其理解为一个专为AI设计的“工作流引擎”。它的核心价值在于可视化/代码化编排允许你通过拖拽或代码定义复杂的、多步骤的AI任务流程例如“获取数据 - 分析 - 生成报告 - 发送审核”。状态管理与回溯自动维护智能体执行过程中的状态State方便调试和错误恢复。工具集成便捷地集成外部工具如计算器、数据库查询、API调用扩展智能体的能力边界。抽象底层LLM提供统一的接口调用不同的模型提供商OpenAI, Anthropic, 本地模型等降低耦合。简单说Harness让构建一个像“财务分析师”一样执行多步骤任务的智能体变得像搭积木一样清晰可控。1.3 LangfuseLLM应用的观测与评估平台Langfuse是一个开源的LLM应用观测平台。如果说Harness是“生产车间”那么Langfuse就是“质量检测与监控中心”。它的核心功能包括全链路追踪Tracing自动记录每次LLM调用的输入、输出、延迟、成本、token用量并可视化整个调用链。提示词管理Prompt Management版本化管理和评估不同的提示词模板实现A/B测试。评估与评分Evaluation支持基于规则如格式检查、模型使用另一个LLM打分或人工的反馈对AI输出进行量化评估。数据分析看板提供丰富的仪表盘分析成本、延迟、评分趋势定位问题。在财务分析场景中Langfuse能帮助我们回答关键问题智能体生成的报告准确性如何分析逻辑是否稳定每次调用的成本是多少1.4 财务分析智能体我们的实战目标我们将构建一个智能体它能够处理一份结构化的财务数据如CSV格式的利润表并完成以下任务数据解读识别关键财务指标营收、毛利率、净利润等。趋势分析计算环比、同比增长率。洞察生成用自然语言总结财务状况指出亮点与风险。报告格式化输出结构清晰的Markdown格式报告。这个智能体将使用Harness来编排“数据读取 - 指标计算 - LLM分析 - 报告生成”的流程并利用Langfuse对每一次执行进行追踪、记录和评估。2. 环境准备与项目初始化我们将使用Python作为主要开发语言。请确保你的环境满足以下要求。2.1 基础环境要求操作系统macOS / Linux / Windows (WSL2推荐)Python版本3.10 或 3.11建议使用3.11以获得最佳兼容性包管理工具pip 或 poetryLLM API密钥你需要一个OpenAI API密钥或 Anthropic、Groq 等兼容OpenAI SDK的API密钥用于调用模型。本文以OpenAI GPT-4o-mini为例。2.2 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir finance-ai-agent cd finance-ai-agent python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来创建requirements.txt文件包含以下核心依赖# 核心AI与编排框架 openai1.0.0 # 假设我们使用一个名为 ai-harness 的模拟Harness框架包进行演示 # 注当前Harness框架生态中有多个项目如 harnessagent-harness等此处为概念演示我们使用一个简化的自制类。 langfuse3.0.0 # 数据处理与工具 pandas2.0.0 numpy1.24.0 # 环境变量管理 python-dotenv1.0.0 # 可选Web框架如需提供API接口 fastapi0.104.0 uvicorn0.24.0安装依赖pip install -r requirements.txt由于目前没有一个统一的、名为“Harness”的Python包我们将模拟其核心编排思想构建一个轻量级的Harness类。在真实项目中你可以根据调研选择如Semantic Kernel,LangChain,AutoGen或新兴的harnessSDK。2.3 配置环境变量创建.env文件存储敏感信息和配置# .env OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key LANGFUSE_HOSThttps://cloud.langfuse.com # 或你的自托管地址 # 项目配置 DEFAULT_LLM_MODELgpt-4o-mini重要请勿将.env文件提交到版本控制系统。确保它在.gitignore中。2.4 初始化Langfuse客户端在项目根目录创建config.py用于初始化全局配置和客户端。# config.py import os from dotenv import load_dotenv from langfuse import Langfuse # 加载环境变量 load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY) LANGFUSE_HOST os.getenv(LANGFUSE_HOST) DEFAULT_LLM_MODEL os.getenv(DEFAULT_LLM_MODEL, gpt-4o-mini) staticmethod def get_langfuse_client(): 初始化并返回Langfuse客户端实现单例模式 if not all([Config.LANGFUSE_PUBLIC_KEY, Config.LANGFUSE_SECRET_KEY]): print(警告: Langfuse 密钥未配置追踪功能将禁用。) return None try: # 在生产环境中建议配置更详细的初始化参数 langfuse_client Langfuse( public_keyConfig.LANGFUSE_PUBLIC_KEY, secret_keyConfig.LANGFUSE_SECRET_KEY, hostConfig.LANGFUSE_HOST, ) return langfuse_client except Exception as e: print(f初始化Langfuse客户端失败: {e}) return None # 全局配置和客户端实例 config Config() langfuse_client config.get_langfuse_client()3. 核心架构与模拟Harness引擎实现我们将实现一个简化的Harness类其核心思想是将复杂的AI任务分解为多个可复用的“节点”Node并通过“边”Edge定义执行顺序和数据流。3.1 定义节点基类与上下文首先定义任务执行时的上下文State和节点的基类。# harness/core.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional, Callable import inspect class State(Dict[str, Any]): 任务执行状态上下文是一个增强型字典。 pass class Node(ABC): Harness节点抽象基类。每个节点代表一个处理单元。 def __init__(self, id: str, description: str ): self.id id self.description description abstractmethod async def execute(self, state: State) - State: 执行节点的核心逻辑。 参数: state: 当前执行状态包含上游节点的输出。 返回: 更新后的状态。 pass def __repr__(self): return fNode(id{self.id})3.2 实现具体功能节点基于我们的财务分析场景我们需要几种类型的节点工具节点Tool Node执行确定性操作如读取文件、计算指标。LLM节点LLM Node调用大语言模型进行分析和生成。条件节点Condition Node根据状态决定执行分支本例暂不展开。让我们实现一个文件读取节点和一个指标计算节点。# harness/nodes/tool_nodes.py import pandas as pd from ..core import Node, State import json class LoadCSVNode(Node): 加载CSV财务数据文件到状态中。 def __init__(self, id: str, file_path_key: str file_path, output_key: str df): super().__init__(id, description加载CSV文件为DataFrame) self.file_path_key file_path_key # 状态中文件路径的键名 self.output_key output_key # 输出DataFrame的键名 async def execute(self, state: State) - State: file_path state.get(self.file_path_key) if not file_path: raise ValueError(f状态中未找到文件路径键: {self.file_path_key}) try: df pd.read_csv(file_path) state[self.output_key] df state[f{self.output_key}_shape] df.shape print(f[{self.id}] 已加载文件: {file_path}, 形状: {df.shape}) except Exception as e: raise RuntimeError(f[{self.id}] 加载CSV文件失败: {e}) return state class CalculateMetricsNode(Node): 计算基础财务指标。 def __init__(self, id: str, df_key: str df, metrics_key: str basic_metrics): super().__init__(id, description计算财务指标营收、毛利、净利等) self.df_key df_key self.metrics_key metrics_key async def execute(self, state: State) - State: df state.get(self.df_key) if df is None or not isinstance(df, pd.DataFrame): raise ValueError(f状态中未找到有效的DataFrame键: {self.df_key}) # 假设CSV有特定列名这里需要根据实际数据调整 # 示例列period, revenue, cost_of_goods_sold, operating_expenses, net_income required_cols [period, revenue, cost_of_goods_sold, net_income] if not all(col in df.columns for col in required_cols): raise ValueError(fDataFrame缺少必要列需要: {required_cols}) metrics {} latest df.iloc[-1] # 取最新一期数据 previous df.iloc[-2] if len(df) 1 else None metrics[latest_period] latest[period] metrics[revenue] float(latest[revenue]) metrics[gross_profit] float(latest[revenue] - latest[cost_of_goods_sold]) metrics[gross_margin] float(metrics[gross_profit] / latest[revenue]) if latest[revenue] ! 0 else 0 metrics[net_income] float(latest[net_income]) metrics[net_margin] float(latest[net_income] / latest[revenue]) if latest[revenue] ! 0 else 0 # 计算环比如果数据足够 if previous is not None: metrics[revenue_qoq] float((latest[revenue] - previous[revenue]) / previous[revenue]) if previous[revenue] ! 0 else 0 metrics[net_income_qoq] float((latest[net_income] - previous[net_income]) / previous[net_income]) if previous[net_income] ! 0 else 0 state[self.metrics_key] metrics print(f[{self.id}] 计算完成基础指标: {json.dumps(metrics, indent2, defaultstr)}) return state接下来实现一个与Langfuse深度集成的LLM节点。这个节点会将其调用过程自动记录到Langfuse。# harness/nodes/llm_nodes.py import openai from openai import OpenAI from ..core import Node, State from config import config, langfuse_client import json class LangfuseLLMNode(Node): 集成Langfuse追踪的LLM调用节点。 def __init__(self, id: str, prompt_template: str, input_state_keys: Dict[str, str], # 例如 {metrics: basic_metrics, df_summary: df_summary} output_key: str llm_response, model: str None, system_prompt: str 你是一个专业的财务分析师请根据提供的数据进行客观、严谨的分析。 ): super().__init__(id, description调用LLM并生成分析报告) self.prompt_template prompt_template self.input_state_keys input_state_keys # 映射模板变量名 - 状态中的键名 self.output_key output_key self.model model or config.DEFAULT_LLM_MODEL self.system_prompt system_prompt self.client OpenAI(api_keyconfig.OPENAI_API_KEY) async def execute(self, state: State) - State: # 1. 从状态中提取数据填充提示词模板 template_data {} for var_name, state_key in self.input_state_keys.items(): if state_key not in state: raise ValueError(f状态中缺少所需键 {state_key}用于模板变量 {var_name}) template_data[var_name] state[state_key] try: user_prompt self.prompt_template.format(**template_data) except KeyError as e: raise ValueError(f提示词模板变量替换失败缺少变量: {e}) # 2. 准备Langfuse追踪如果客户端可用 trace None generation None if langfuse_client: trace langfuse_client.trace( namefFinanceAnalysis-{self.id}, input{system_prompt: self.system_prompt, user_prompt_preview: user_prompt[:200]}, metadata{node_id: self.id, model: self.model}, ) generation trace.generation( nameFinancial Report Generation, modelself.model, model_parameters{temperature: 0.2, max_tokens: 1500}, input[{role: system, content: self.system_prompt}, {role: user, content: user_prompt}], ) # 3. 调用OpenAI API try: response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: self.system_prompt}, {role: user, content: user_prompt} ], temperature0.2, max_tokens1500, ) llm_output response.choices[0].message.content # 4. 记录成功结果到Langfuse if generation: generation.end(outputllm_output) trace.update(output{report: llm_output[:500]}) # 只记录预览 print(f[{self.id}] LLM调用成功结果已记录至Langfuse Trace: {trace.id}) else: print(f[{self.id}] LLM调用成功 (Langfuse未启用)) # 5. 将结果存入状态 state[self.output_key] llm_output # 也可以存储原始响应对象以备后用 state[f{self.output_key}_raw] response except Exception as e: # 6. 记录失败信息到Langfuse if generation: generation.end(errorstr(e)) trace.update(output{error: str(e)}) print(f[{self.id}] LLM调用失败: {e}) raise RuntimeError(fLLM节点执行失败: {e}) from e finally: # 确保Langfuse客户端刷新数据 if langfuse_client: langfuse_client.flush() return state3.3 实现Harness编排引擎现在我们将这些节点连接起来构建一个简单的线性执行引擎。# harness/engine.py from typing import List, Dict, Any from .core import Node, State import asyncio class Harness: 简单的线性工作流编排引擎。 def __init__(self, name: str FinanceAgentHarness): self.name name self.nodes: List[Node] [] self.state State() def add_node(self, node: Node): 向工作流添加一个节点。 self.nodes.append(node) return self # 支持链式调用 def set_initial_state(self, **kwargs): 设置工作流的初始状态。 self.state.update(kwargs) async def run(self) - State: 顺序执行所有节点。 print(f 开始执行工作流 [{self.name}] ) current_state self.state.copy() for i, node in enumerate(self.nodes): print(f\n[{i1}/{len(self.nodes)}] 执行节点: {node.id} ({node.description})) try: current_state await node.execute(current_state) except Exception as e: print(f!!! 节点 {node.id} 执行失败工作流终止。错误: {e}) # 可以将错误状态记录到Langfuse if langfuse_client: langfuse_client.trace( namefHarnessError-{self.name}, input{failed_node: node.id, state_snapshot: str(current_state)}, output{error: str(e)}, levelERROR ) langfuse_client.flush() raise print(f\n 工作流 [{self.name}] 执行完成 ) return current_state def visualize(self): 打印工作流的简单文本可视化。 print(f工作流: {self.name}) for i, node in enumerate(self.nodes): print(f {i1}. [{node.id}] - {node.description})4. 完整实战构建并运行财务分析智能体现在我们将所有部分组合起来创建一个完整的可执行示例。4.1 准备示例财务数据创建一个示例CSV文件data/sample_finance_data.csvperiod,revenue,cost_of_goods_sold,operating_expenses,net_income 2023-Q1,1000000,600000,250000,150000 2023-Q2,1200000,700000,280000,220000 2023-Q3,1100000,650000,260000,190000 2023-Q4,1300000,750000,300000,2500004.2 定义提示词模板与工作流创建主执行脚本main.py# main.py import asyncio import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from harness.engine import Harness from harness.nodes.tool_nodes import LoadCSVNode, CalculateMetricsNode from harness.nodes.llm_nodes import LangfuseLLMNode # 1. 定义分析报告提示词模板 FINANCIAL_ANALYSIS_PROMPT 你是一名资深财务分析师。请基于以下财务数据和分析指标生成一份简洁、专业的财务分析报告。 **基础数据概览** - 数据期间: {periods} - 总记录数: {record_count} **计算出的关键指标最新一期 {latest_period}** - 营业收入: {revenue:,.2f} - 毛利润: {gross_profit:,.2f} - 毛利率: {gross_margin:.2%} - 净利润: {net_income:,.2f} - 净利率: {net_margin:.2%} {trend_analysis} **请按以下结构组织报告** 1. **核心业绩摘要**用2-3句话总结本期核心财务表现。 2. **盈利能力分析**分析毛利率和净利率的水平及其含义。 3. **增长趋势洞察**结合环比数据评论营收和利润的增长趋势。 4. **潜在风险与关注点**基于数据指出1-2个可能的风险或需要关注的方面。 5. **后续建议**给出1-2条具体的后续分析或行动建议。 报告请使用Markdown格式确保数据准确分析客观。 async def main(): # 2. 初始化Harness工作流 harness Harness(nameQuarterlyFinancialAnalysis) # 3. 添加节点定义执行流水线 harness.add_node( LoadCSVNode(idload_data, file_path_keyinput_file, output_keydf) ).add_node( CalculateMetricsNode(idcalc_metrics, df_keydf, metrics_keybasic_metrics) ).add_node( LangfuseLLMNode( idgenerate_report, prompt_templateFINANCIAL_ANALYSIS_PROMPT, input_state_keys{ periods: periods_summary, # 这些键需要从状态中获取 record_count: record_count, latest_period: latest_period, revenue: revenue, gross_profit: gross_profit, gross_margin: gross_margin, net_income: net_income, net_margin: net_margin, trend_analysis: trend_text, }, output_keyanalysis_report, system_prompt你是一个严谨、客观的财务分析师擅长从数据中发现洞察并以结构化的方式呈现。, modelgpt-4o-mini # 或 gpt-3.5-turbo ) ) # 4. 准备初始状态并补充LLM节点需要的额外数据 # 注意LLM节点需要的数据有些需要前面的节点生成有些需要我们从原始数据中提取并放入状态。 # 我们可以在运行前设置一部分另一部分通过一个“数据准备节点”或直接在运行中计算后注入。 # 这里我们采用一个简单方法在运行工作流前先手动计算一些衍生数据放入状态。 # 更优雅的方式是创建一个 PrepareForLLMNode。 initial_state { input_file: data/sample_finance_data.csv, } harness.set_initial_state(**initial_state) # 5. 可视化工作流 print(构建的工作流如下) harness.visualize() # 6. 执行工作流 try: final_state await harness.run() except Exception as e: print(f\n工作流执行因错误中断: {e}) return # 7. 打印最终结果 print(\n *50) print(财务分析报告生成完成) print(*50) if analysis_report in final_state: print(final_state[analysis_report]) else: print(未生成报告。最终状态:, final_state.keys()) if __name__ __main__: asyncio.run(main())4.3 运行智能体并查看结果在终端运行python main.py你将看到类似以下的输出具体内容因模型随机性而异构建的工作流如下 工作流: QuarterlyFinancialAnalysis 1. [load_data] - 加载CSV文件为DataFrame 2. [calc_metrics] - 计算财务指标营收、毛利、净利等 3. [generate_report] - 调用LLM并生成分析报告 开始执行工作流 [QuarterlyFinancialAnalysis] [1/3] 执行节点: load_data (加载CSV文件为DataFrame) [load_data] 已加载文件: data/sample_finance_data.csv, 形状: (4, 5) [2/3] 执行节点: calc_metrics (计算财务指标营收、毛利、净利等) [calc_metrics] 计算完成基础指标: { latest_period: 2023-Q4, revenue: 1300000.0, gross_profit: 550000.0, gross_margin: 0.4230769230769231, net_income: 250000.0, net_margin: 0.19230769230769232, revenue_qoq: 0.18181818181818182, net_income_qoq: 0.3157894736842105 } [3/3] 执行节点: generate_report (调用LLM并生成分析报告) [generate_report] LLM调用成功结果已记录至Langfuse Trace: 01JXXXXXX 工作流 [QuarterlyFinancialAnalysis] 执行完成 财务分析报告生成完成 # 财务分析报告 ## 1. 核心业绩摘要 2023年第四季度公司实现营业收入130万元净利润25万元。毛利率为42.31%净利率为19.23%整体盈利能力保持稳健。 ## 2. 盈利能力分析 本期毛利率为42.31%表明公司产品或服务具有较强的市场竞争力成本控制有效。净利率达到19.23%在扣除运营费用后仍保持了较高的利润留存率整体盈利结构健康。 ## 3. 增长趋势洞察 与第三季度相比第四季度营收环比增长约18.18%净利润环比增长约31.58%增速显著。这反映出公司在年末可能采取了有效的市场策略或成本优化措施推动了利润的更快增长。 ## 4. 潜在风险与关注点 1. **成本压力**尽管毛利率可观但需持续关注原材料或直接成本销售成本的变动其占营收比例仍超过57%。 2. **增长可持续性**本季度的强劲增长是否具有可持续性还是受季节性因素影响需要结合更多历史数据和业务背景判断。 ## 5. 后续建议 1. **深入分析成本结构**建议对销售成本cost_of_goods_sold进行细分识别可优化的具体环节。 2. **制定2024年季度预算**基于2023年的增长趋势为2024年各季度设定合理的营收与利润目标并建立监控机制。4.4 在Langfuse平台查看追踪详情登录你的Langfuse账户云服务或自托管。进入“Traces”页面你应该能看到一条名为FinanceAnalysis-generate_report的追踪记录。点击进入可以看到完整的追踪详情输入系统提示词和用户提示词预览。输出生成的完整报告。元数据节点ID、使用的模型。分析Token使用量、成本、延迟。你可以在“Prompts”部分管理你的提示词模板并进行版本对比。在“Evaluations”部分可以基于本次输出创建评估例如让另一个LLM从“专业性”、“数据准确性”维度打分。5. 企业级评估平台搭建与AI工程化实践仅仅生成报告和记录追踪是不够的。企业级应用需要系统化的评估、监控和迭代能力。下面我们利用Langfuse构建一个简单的评估流程。5.1 定义自动化评估指标我们可以为财务报告定义几个自动化评估维度格式合规性报告是否包含要求的5个部分基于规则数据保真度报告中引用的数字是否与输入数据一致基于LLM或规则专业性评分报告的语言是否专业、客观基于LLM我们在项目中创建一个新的模块evaluation/evaluator.py# evaluation/evaluator.py import re from typing import Dict, Any from langfuse import Langfuse from config import langfuse_client class ReportEvaluator: def __init__(self): self.client langfuse_client def evaluate_format(self, report: str, required_sections: list) - Dict[str, Any]: 评估报告格式是否包含所有必需章节。 score 100 feedback [] missing_sections [] for section in required_sections: # 简单检查章节标题是否存在于报告中 pattern rf^#\s*{section}|^#\s*\d\.\s*{section} if not re.search(pattern, report, re.IGNORECASE | re.MULTILINE): missing_sections.append(section) score - 20 # 每缺一个章节扣20分 if missing_sections: feedback.append(f报告缺少以下必需章节: {, .join(missing_sections)}) else: feedback.append(报告格式完整包含所有必需章节。) return { score: max(0, score), feedback: ; .join(feedback), details: {missing_sections: missing_sections} } async def evaluate_data_fidelity(self, trace_id: str, original_metrics: Dict, report: str) - Dict[str, Any]: 使用LLM评估报告中的数据是否与原始指标一致。 if not self.client: return {score: -1, feedback: Langfuse客户端未初始化无法进行评估。} # 从Langfuse获取该Trace的完整生成记录 # 注意Langfuse Python SDK目前可能不直接支持通过ID查询Trace这里为演示逻辑。 # 实际应用中你可能需要在生成时保存更多上下文或使用Langfuse API。 # 此处简化为直接调用一个新的LLM进行评估。 from openai import OpenAI from config import config client OpenAI(api_keyconfig.OPENAI_API_KEY) evaluation_prompt f 你是一个严谨的数据审计员。请对比以下两组信息 **【原始财务指标】** {original_metrics} **【生成的财务分析报告】** {report} 请判断报告中引用的核心数据如营业收入、毛利率、净利润、增长率等是否与原始指标**完全一致**。 注意报告可能对数字进行格式化如添加千分位逗号或使用近似表述如“约42%”这不算错误。 请只检查是否存在**事实性矛盾**例如原始数据是130万报告写成120万。 请按以下格式回答 一致性结论: [是/否] 得分: (0-100分100分为完全一致) 不一致详情: [如果结论为“否”请列出具体不一致的数据点否则写“无”] try: response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: evaluation_prompt}], temperature0, max_tokens500, ) result_text response.choices[0].message.content # 简单解析结果实际应用需更健壮的解析 score 100 feedback 数据一致性评估完成。 if 一致性结论: 否 in result_text: score 50 # 假设不一致则给50分 feedback 检测到报告中的数据与原始指标存在潜在不一致。 # 可以将此次评估也记录到Langfuse关联到原始Trace if self.client and trace_id: self.client.score( trace_idtrace_id, namedata_fidelity, valuescore, commentfeedback, ) return {score: score, feedback: feedback, details: result_text} except Exception as e: return {score: -1, feedback: f数据一致性评估失败: {e}} def log_evaluation_to_langfuse(self, trace_id: str, evaluation_name: str, score: float, comment: str ): 将评估分数记录到Langfuse对应的Trace上。 if self.client and trace_id: try: self.client.score( trace_idtrace_id, nameevaluation_name, valuescore, commentcomment, ) print(f[评估已记录] Trace: {trace_id}, 指标: {evaluation_name}, 分数: {score}) except Exception as e: print(f记录评估分数到Langfuse失败: {e})5.2 集成评估到主工作流修改main.py在工作流执行后自动触发评估。# 在 main.py 的 main() 函数末尾打印报告后添加 # ... 原有打印报告的代码 ... # 8. 执行自动化评估 if analysis_report in final_state and langfuse_client: print(\n *50) print(开始自动化报告评估...) print(*50) evaluator ReportEvaluator() report final_state[analysis_report] # 评估1格式合规性 required_sections [核心业绩摘要, 盈利能力分析, 增长趋势洞察, 潜在风险与关注点, 后续建议] format_result evaluator.evaluate_format(report, required_sections) print(f格式评估得分: {format_result[score]}/100 - {format_result[feedback]}) # 记录分数到Langfuse (需要trace_id) trace_id 从final_state或全局变量中获取 # 实际需要从LLM节点执行时获取并传递 # evaluator.log_evaluation_to_langfuse(trace_id, format_compliance, format_result[score]) # 评估2数据保真度 (异步) # 需要获取原始指标和trace_id # original_metrics final_state.get(basic_metrics, {}) # fidelity_result await evaluator.evaluate_data_fidelity(trace_id, original_metrics, report) # print(f数据一致性评估得分: {fidelity_result[score]}/100 - {fidelity_result[feedback]}) print(自动化评估完成。请登录Langfuse平台查看详细追踪与评分。)5.3 构建评估看板与持续迭代在Langfuse平台上你可以创建Dashboard监控“报告生成延迟”、“平均格式得分”、“数据一致性得分”等核心指标。设置警报当评估分数低于阈值如格式分80时发送通知。Prompt版本管理在Langfuse的“Prompts”中迭代你的FINANCIAL_ANALYSIS_PROMPT并对比不同版本生成报告的质量。人工反馈集成在生成的报告旁提供“拇指向上/下”按钮收集人工反馈并将其作为评估信号。通过这一套组合你就建立了一个“开发Harness - 运行 - 观测与评估Langfuse - 迭代优化”的完整AI工程化闭环。6. 常见问题与排查思路在实际部署和运行中你可能会遇到以下问题问题现象可能原因排查思路与解决方案运行时报ModuleNotFoundError: No module named harness自定义的harness模块路径未正确加入Python路径。1. 确保在项目根目录下运行脚本。2. 在main.py开头使用sys.path.append添加项目根目录或使用PYTHONPATH环境变量。3. 检查__init__.py文件是否存在于harness/和harness/nodes/目录下。Langfuse追踪未显示在控制台1. API密钥错误或未配置。2. 网络问题。3. Langfuse客户端初始化失败。1. 检查.env文件中的LANGFUSE_PUBLIC_KEY和LANGFUSE_SECRET_KEY是否正确。2. 检查config.py中get_langfuse_client方法的错误打印。3. 尝试在代码中暂时禁用Langfuse确认基础功能正常。LLM节点调用超时或报错1. OpenAI API密钥无效或余额不足。2. 网络连接问题。3. 模型名称错误或不可用。1. 在OpenAI平台验证API密钥状态和额度。2. 尝试使用curl或openai库的简单测试脚本确认API连通性。3. 确认model参数是有效的模型名如gpt-3.5-turbo。提示词模板格式化报KeyErrorinput_state_keys中定义的键在节点执行时状态中不存在。1. 仔细检查input_state_keys的映射关系。确保上游节点如CalculateMetricsNode的输出键与映射中state_key一致。2. 在LangfuseLLMNode.execute方法开始处打印state.keys()进行调试。财务指标计算逻辑错误CSV文件列名与代码中required_cols不匹配或数据格式非数值。1. 打印加载后的DataFrame的columns和dtypes进行确认。2. 在CalculateMetricsNode中添加更严格的数据类型校验和转换。评估分数未关联到Tracetrace_id未正确传递到ReportEvaluator。1. 修改LangfuseLLMNode在执行成功后将trace.id存入state如state[trace_id] trace.id。2. 在工作流最终状态中将trace_id传递给评估器。7. 最佳实践与工程化建议将此类AI智能体项目投入生产环境需要遵循以下工程化实践7.1 项目结构与代码组织清晰的模块化正如本文所示将引擎、节点、配置、评估逻辑分离。nodes/目录下可按功能进一步细分如tools/,llms/,conditions/。配置外部化所有API密钥、模型参数、文件路径、提示词模板都应通过.env或配置中心如Apollo管理严禁硬编码。依赖管理使用requirements.txt或pyproject.toml精确锁定依赖版本避免环境差异导致运行失败。7.2 提示词工程与管理版本控制将提示词模板存储在代码库或Langfuse等专用平台中并进行版本控制。每次修改都应记录原因和预期效果。变量化与验证像本文一样使用明确的变量占位符{variable}并在填充前验证状态中是否存在对应数据避免运行时错误。系统提示词设计系统提示词system_prompt是塑造AI行为的关键。它应明确角色、任务边界、输出格式和禁忌。7.3 可观测性与监控全链路追踪务必为每个LLM调用、关键工具调用记录追踪。除了输入输出还应记录延迟、Token用量和成本。定义业务指标在Langfuse中创建与业务价值直接相关的评估指标Score如“报告格式合规分”、“数据准确分”、“用户满意度分”。通过看板持续监控。设置警报对错误率、高延迟、成本突增、评估分下降设置阈值告警。7.4 错误处理与韧性节点级容错每个Node的execute方法应有完善的try-except捕获可能异常并决定是向上抛出终止流程还是记录错误后返回降级结果。重试与退避对于LLM API调用等可能因网络或速率限制失败的操作实现带指数退避的重试机制。状态快照与回滚复杂的Harness引擎应支持将执行状态持久化在失败时可以从上一个成功节点恢复而不是从头开始。7.5 安全与合规数据脱敏财务数据高度敏感。在将数据发送给外部LLM API前必须进行严格的脱敏处理如替换真实公司名、金额缩放。对于极高敏感数据考虑使用本地模型。权限控制确保只有授权用户能触发智能体工作流并能访问生成的报告和原始数据。审计日志所有智能体的触发、输入、输出、执行人、时间都应记录到不可篡改的审计日志中满足合规要求。7.6 性能与成本优化缓存策略对于相同输入可能产生相同输出的LLM调用或计算节点考虑引入缓存如Redis显著降低成本和延迟。模型选型并非所有任务都需要最强大的模型。对于数据提取、格式转换等简单任务可使用更小、更快的模型如gpt-3.5-turbo。异步执行如果工作流中节点间没有强依赖可以将其设计为异步并行执行缩短整体耗时。通过本文的实战你不仅学会了一个财务分析智能体的构建更掌握了一套基于Harness架构思想与Langfuse评估平台的AI工程化落地方法论。这套方法论可以平移到客服、营销、运维、代码生成等几乎所有AI智能体应用场景。