如果你正在尝试将AI能力应用到企业财务分析这类严肃业务中大概率会遇到这样的困境模型生成的财务报告看似专业但数据来源不明、逻辑链条断裂你无法向业务部门解释“这个结论是怎么来的”或者你费尽心思搭建的智能体在生产环境跑了几次就“失忆”或“胡言乱语”却找不到问题出在哪个环节。这背后不是模型能力的问题而是AI工程化的缺失。我们习惯了在Jupyter Notebook里快速验证一个想法但要将它变成企业里稳定、可评估、可迭代的生产系统中间隔着一道巨大的鸿沟。今天要讨论的“Harness架构”与“Langfuse”正是填平这道鸿沟的关键工具组合。它们一个负责智能体的标准化生产与部署Harness一个负责全链路追踪与评估Langfuse。本文不会空谈概念我们将通过一个企业级财务分析智能体的完整项目实战带你一步步实现从本地原型到可评估、可运维的生产化落地。你会看到如何用代码构建一个能理解财报、进行比率分析、生成洞察报告的智能体并在此过程中深刻理解Harness的“工程化”思想和Langfuse的“可观测性”价值。1. 这篇文章真正要解决的问题从AI玩具到生产工具为什么很多AI项目停留在演示阶段核心在于缺乏标准化流水线和效果度量体系。想象一下传统软件开发我们有Git做版本控制Jenkins做CI/CDPrometheus做监控日志系统做排错。这套体系确保了软件的可重复构建、自动化测试和稳定运行。而当前的AI应用开发尤其是基于大语言模型的智能体开发很大程度上还处在“手工作坊”时代提示词写在笔记本里上下文管理靠代码硬编码每次调优靠人工对比输出上线后效果波动无从追溯。这就是Harness和Langfuse要解决的问题Harness它提供了一套用于构建、编排和部署AI智能体Agent的标准化框架。你可以把它理解为AI智能体的“Kubernetes”或“Spring Framework”。它定义了智能体Agent、技能Skill、工作流Workflow等核心概念让开发变得模块化、可配置。Langfuse它是一个开源的LLM应用可观测性平台。它自动追踪每次AI调用的输入Prompt、输出Completion、耗时、成本并能通过手动或自动评分进行效果评估。它是你AI应用的“黑匣子”和“评分器”。本实战项目将模拟一个经典场景自动财务分析智能体。给定一家上市公司的股票代码或名称智能体将自动完成以下任务从公开数据源如Yahoo Finance获取财务数据。计算关键财务比率如毛利率、净利率、资产负债率等。结合行业基准和历史趋势生成一份结构化的分析报告。整个过程的每一步都被Harness清晰地编排并被Langfuse完整地追踪和评估。通过这个项目你将掌握的不再是零散的API调用而是一套让AI应用变得可靠、可解释、可优化的工程化方法。2. 核心概念与架构解析在动手之前必须厘清几个关键概念否则代码只是空中楼阁。2.1 Harness 核心概念像组装乐高一样构建智能体Harness将复杂的智能体系统分解为几个层次分明的组件Skill技能最基础的执行单元。一个Skill完成一件特定、原子性的任务。例如“获取股票财务数据”、“计算流动比率”、“生成自然语言摘要”。每个Skill通常包含清晰的输入、输出定义和执行逻辑可以是调用LLM也可以是纯计算。Agent智能体由一个或多个Skill组装而成能够完成一个更复杂的、有明确目标的任务。例如“财务分析智能体”就是由“数据获取Skill”、“比率计算Skill”、“报告生成Skill”组合而成。Agent负责管理Skill之间的执行顺序和数据流转。Workflow工作流用于编排多个Agent或更复杂任务的执行流程。它可以包含条件分支、循环等逻辑。在财务分析场景中一个工作流可以先后调用“数据获取Agent”和“报告生成Agent”。Harness 指代整个框架或运行时环境。它提供了Skill和Agent的注册、发现、执行引擎以及与其他系统如Langfuse集成的能力。通俗理解Skill是“拧螺丝”、“装轮子”的工人Agent是能“组装一辆自行车”的班组Workflow是“先组装车架再喷漆最后质检”的工厂流水线Harness就是整座“工厂”的管理系统。2.2 Langfuse 核心价值为AI应用装上“仪表盘”Langfuse关注的是应用运行时的“可观测性”Trace追踪一次完整的用户交互或任务执行过程。例如用户询问“分析一下AAPL的财务状况”从接收到请求到返回最终报告这整个过程构成一个Trace。Observation观测点Trace中的每一个步骤比如一次LLM调用、一次工具调用、一次数据查询都是一个Observation。Langfuse会自动记录它们的输入、输出、开始时间、结束时间、消耗的Token数、成本等。Score评分对Trace或Observation的质量进行量化评价。可以是人工打分比如业务专家审阅报告后给分也可以是自动评分比如用另一个模型检查报告是否包含关键指标。Dataset Evaluation数据集与评估你可以将标准的测试用例如“分析AAPL”、“分析MSFT”定义为数据集然后定期或批量运行你的智能体用Langfuse自动收集结果并评分形成效果评估报告。关键洞察没有Langfuse你的智能体就是一个“盲盒”出了问题只能靠猜。有了Langfuse每一次调用都透明可见每一次迭代都有数据支撑。2.3 项目整体架构图用户请求 | v [ Harness 网关 / API ] | v [ 财务分析 Agent (Harness) ] |-----------------------| | | v v [数据获取 Skill] [比率计算 Skill] [报告生成 Skill] | | | v v v 调用Yahoo Finance API 执行Python计算函数 调用LLM API (如GPT-4) | | | | | | |-----------------------|-----------------------| | v 生成最终财务报告 | v [ Langfuse SDK 集成 ] | v 全程自动记录 Trace, Observation, Token用量成本... | v 可选人工或自动评分 (Score)这个架构确保了功能解耦Harness和过程透明Langfuse。3. 环境准备与项目初始化我们使用Python作为开发语言。请确保你的环境满足以下要求。3.1 基础环境要求操作系统 macOS / Linux / Windows (WSL2推荐)Python版本 3.9 或以上包管理工具 pip 或 conda3.2 关键依赖安装创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir finance_ai_agent cd finance_ai_agent # 创建并激活虚拟环境 (以venv为例) python -m venv venv # Windows venv\Scripts\activate # macOS/Linux source venv/bin/activate # 安装核心框架和工具 pip install harness-core # Harness核心框架 pip install langfuse # Langfuse Python SDK pip install openai # 用于调用GPT模型 pip install yfinance # 用于获取雅虎财经数据 pip install pandas numpy # 数据处理 pip install python-dotenv # 管理环境变量重要提示harness-core是Harness框架的核心Python库。网络热词中提到的deepseek harness可能指特定版本或发行版在本文中我们以通用的harness-core概念和模式进行讲解其工程化思想是相通的。3.3 获取并配置API密钥本项目需要OpenAI和Langfuse的API密钥。OpenAI API Key: 访问 OpenAI平台 创建。Langfuse API Key: 你有两种选择云服务 注册 Langfuse Cloud 。自托管 按照官方文档部署Langfuse服务涉及Docker, PostgreSQL等。 获取你的LANGFUSE_SECRET_KEY,LANGFUSE_PUBLIC_KEY和LANGFUSE_HOST。在项目根目录创建.env文件填入你的密钥# .env 文件 OPENAI_API_KEYsk-your-openai-api-key-here # Langfuse 配置 (如果使用云服务HOST通常是 https://cloud.langfuse.com) LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key LANGFUSE_HOSThttps://cloud.langfuse.com安全警告 务必在.gitignore文件中加入.env切勿将包含密钥的文件提交到版本控制系统。4. 构建财务分析智能体的核心Skills我们将遵循Harness的架构先构建三个核心Skill。4.1 Skill 1: 财务数据获取 (FetchFinancialDataSkill)这个Skill负责从Yahoo Finance获取原始的财务数据。# skills/fetch_financial_data.py import yfinance as yf from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import Dict, Any class FinancialDataInput(SkillInput): 获取财务数据Skill的输入模型 symbol: str Field(..., description上市公司股票代码例如AAPL, MSFT) class FinancialDataOutput(SkillOutput): 获取财务数据Skill的输出模型 income_stmt: Dict[str, Any] Field(default_factorydict, description利润表数据) balance_sheet: Dict[str, Any] Field(default_factorydict, description资产负债表数据) cash_flow: Dict[str, Any] Field(default_factorydict, description现金流量表数据) info: Dict[str, Any] Field(default_factorydict, description公司基本信息) skill( namefetch_financial_data, description根据股票代码从雅虎财经获取利润表、资产负债表和现金流量表数据。, input_modelFinancialDataInput, output_modelFinancialDataOutput ) async def fetch_financial_data_skill(input_data: FinancialDataInput) - FinancialDataOutput: 财务数据获取技能 ticker yf.Ticker(input_data.symbol) # 获取最近一个财年的数据 income_stmt ticker.income_stmt balance_sheet ticker.balance_sheet cash_flow_stmt ticker.cashflow info ticker.info # 将DataFrame转换为字典以便序列化传输 output_data FinancialDataOutput( income_stmtincome_stmt.to_dict() if income_stmt is not None else {}, balance_sheetbalance_sheet.to_dict() if balance_sheet is not None else {}, cash_flowcash_flow_stmt.to_dict() if cash_flow_stmt is not None else {}, infoinfo ) return output_data关键点解析skill装饰器将普通函数注册为Harness框架可识别的Skill。SkillInput和SkillOutput继承自BaseModel明确定义了技能的“接口契约”这是工程化的关键确保了类型安全和自描述性。函数内部使用yfinance库获取数据并进行简单的格式转换。4.2 Skill 2: 财务比率计算 (CalculateRatiosSkill)这个Skill接收上一个Skill的财务数据计算关键财务比率。# skills/calculate_ratios.py from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import Dict, Any, List class RatioCalculationInput(SkillInput): 计算财务比率Skill的输入模型 income_stmt: Dict[str, Any] Field(..., description利润表数据) balance_sheet: Dict[str, Any] Field(..., description资产负债表数据) cash_flow: Dict[str, Any] Field(default_factorydict, description现金流量表数据) class FinancialRatio(BaseModel): 单个财务比率的模型 name: str value: float description: str formula: str class RatioCalculationOutput(SkillOutput): 计算财务比率Skill的输出模型 ratios: List[FinancialRatio] Field(default_factorylist, description计算出的财务比率列表) warnings: List[str] Field(default_factorylist, description计算过程中的警告信息) def safe_get(data_dict: Dict[str, Any], key: str, default: float 0.0) - float: 安全地从嵌套字典中获取数值避免KeyError。 # 简化实现实际项目可能需要更复杂的逻辑处理DataFrame转换后的结构 if isinstance(data_dict, dict): # 假设数据格式为 {‘2023-12-31‘: 100000, ‘2022-12-31‘: 90000} # 我们取最近一年的数据 for date_key, value in data_dict.items(): if isinstance(value, (int, float)): return float(value) return default skill( namecalculate_financial_ratios, description基于利润表和资产负债表计算关键财务比率如盈利能力、偿债能力、运营效率比率。, input_modelRatioCalculationInput, output_modelRatioCalculationOutput ) async def calculate_ratios_skill(input_data: RatioCalculationInput) - RatioCalculationOutput: 财务比率计算技能 income_stmt input_data.income_stmt balance_sheet input_data.balance_sheet ratios [] warnings [] # 1. 盈利能力比率 revenue safe_get(income_stmt, Total Revenue) gross_profit safe_get(income_stmt, Gross Profit) net_income safe_get(income_stmt, Net Income) total_assets safe_get(balance_sheet, Total Assets) if revenue 0: gross_margin (gross_profit / revenue) * 100 ratios.append(FinancialRatio( nameGross Margin, valueround(gross_margin, 2), description毛利率反映核心业务的盈利能力。, formulaGross Profit / Total Revenue )) net_margin (net_income / revenue) * 100 ratios.append(FinancialRatio( nameNet Margin, valueround(net_margin, 2), description净利率反映最终盈利能力。, formulaNet Income / Total Revenue )) else: warnings.append(营收数据为0或缺失无法计算利润率。) # 2. 偿债能力比率 total_liabilities safe_get(balance_sheet, Total Liabilities Net Minority Interest) total_equity safe_get(balance_sheet, Total Equity Gross Minority Interest) if total_assets 0: debt_to_assets (total_liabilities / total_assets) * 100 ratios.append(FinancialRatio( nameDebt to Assets Ratio, valueround(debt_to_assets, 2), description资产负债率反映资产中有多少由债务构成。, formulaTotal Liabilities / Total Assets )) # 3. 运营效率比率 (示例资产周转率) if total_assets 0 and revenue 0: asset_turnover revenue / total_assets ratios.append(FinancialRatio( nameAsset Turnover, valueround(asset_turnover, 2), description资产周转率反映资产创造收入的效率。, formulaTotal Revenue / Total Assets )) return RatioCalculationOutput(ratiosratios, warningswarnings)关键点解析定义了FinancialRatio这个数据模型使输出高度结构化。safe_get函数处理了从原始数据字典中安全提取数值的逻辑这是工程健壮性的体现。计算了典型的财务比率并附上了描述和公式增强了结果的可解释性。4.3 Skill 3: 生成分析报告 (GenerateReportSkill)这个Skill将前两个Skill的结果整合调用LLM生成一份易于理解的财务分析报告。# skills/generate_report.py import os from openai import OpenAI from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import List, Dict, Any from dotenv import load_dotenv load_dotenv() # 加载环境变量 class ReportGenerationInput(SkillInput): 生成报告Skill的输入模型 symbol: str Field(..., description股票代码) company_info: Dict[str, Any] Field(default_factorydict, description公司基本信息) ratios: List[Dict[str, Any]] Field(default_factorylist, description财务比率列表) raw_data_snapshot: Dict[str, Any] Field(default_factorydict, description原始数据摘要) class ReportGenerationOutput(SkillOutput): 生成报告Skill的输出模型 analysis_report: str Field(..., description生成的财务分析报告文本) model_used: str Field(defaultgpt-4, description生成报告所使用的LLM模型) skill( namegenerate_financial_analysis_report, description基于财务数据和计算出的比率生成一份结构化的、易于理解的财务分析报告。, input_modelReportGenerationInput, output_modelReportGenerationOutput ) async def generate_report_skill(input_data: ReportGenerationInput) - ReportGenerationOutput: 财务分析报告生成技能 client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 构建给LLM的提示词 prompt f 你是一名资深的财务分析师。请根据以下信息为股票代码为 {input_data.symbol} 的公司生成一份专业的财务分析报告。 公司基本信息 {input_data.company_info} 计算出的关键财务比率 {input_data.ratios} 原始数据摘要 {input_data.raw_data_snapshot} 报告要求 1. 以“【财务分析报告】{input_data.symbol}”为标题。 2. 报告结构应包含摘要、盈利能力分析、偿债能力分析、运营效率分析、总体评价与建议。 3. 分析要结合比率数值给出定性判断例如优秀、良好、一般、需警惕。 4. 语言专业、简洁、客观。 5. 在报告末尾注明“本报告基于公开财务数据生成不构成投资建议。” try: response client.chat.completions.create( modelgpt-4-turbo-preview, # 可根据需要调整模型 messages[ {role: system, content: 你是一名专业的财务分析师。}, {role: user, content: prompt} ], temperature0.2, # 低温度值使输出更稳定、更聚焦 max_tokens1500 ) report_content response.choices[0].message.content return ReportGenerationOutput( analysis_reportreport_content, model_usedgpt-4-turbo-preview ) except Exception as e: # 异常处理返回一个包含错误信息的报告 error_report f【财务分析报告生成失败】{input_data.symbol} 抱歉在生成分析报告时遇到错误{str(e)} 已获取的数据和比例如下可供手动分析 公司信息{input_data.company_info} 财务比率{input_data.ratios} return ReportGenerationOutput( analysis_reporterror_report, model_usederror )关键点解析集成了OpenAI API使用Chat Completion模式。精心构造了提示词Prompt明确规定了报告的结构、内容和风格。这是保证LLM输出质量稳定的关键。加入了异常处理即使LLM调用失败也能返回有意义的输出保证了整个Agent工作流的鲁棒性。5. 组装Agent与集成Langfuse追踪现在我们将三个Skill组装成一个完整的财务分析Agent并集成Langfuse进行全链路追踪。5.1 创建主Agent并集成Langfuse# agent/financial_analysis_agent.py import asyncio import os from typing import Dict, Any from langfuse import Langfuse from langfuse.decorators import observe, langfuse_context # 注意检查你的langfuse版本是否支持decorators from dotenv import load_dotenv from harness.core.agent import Agent from skills.fetch_financial_data import fetch_financial_data_skill, FinancialDataOutput from skills.calculate_ratios import calculate_ratios_skill, RatioCalculationInput from skills.generate_report import generate_report_skill, ReportGenerationInput load_dotenv() # 初始化Langfuse客户端 langfuse Langfuse( secret_keyos.getenv(LANGFUSE_SECRET_KEY), public_keyos.getenv(LANGFUSE_PUBLIC_KEY), hostos.getenv(LANGFUSE_HOST) ) class FinancialAnalysisAgent(Agent): 财务分析智能体 def __init__(self): super().__init__(namefinancial_analysis_agent) # 注册该Agent所拥有的Skills (虽然Skill已全局注册这里可以声明依赖) self.required_skills [fetch_financial_data, calculate_financial_ratios, generate_financial_analysis_report] observe(namefinancial_analysis_agent_execution) # Langfuse装饰器自动追踪此函数 async def execute(self, symbol: str) - Dict[str, Any]: 执行财务分析流程 # 设置Langfuse Trace的元数据 langfuse_context.update_current_trace( namefFinance Analysis for {symbol}, user_iddemo_user, # 在实际应用中这里可以传入真实用户ID metadata{symbol: symbol, agent_version: 1.0} ) final_result {} try: # 步骤1: 获取财务数据 data_input {symbol: symbol} fetch_result: FinancialDataOutput await fetch_financial_data_skill(data_input) final_result[raw_data] { info: fetch_result.info, income_stmt_keys: list(fetch_result.income_stmt.keys())[:3] if fetch_result.income_stmt else [], balance_sheet_keys: list(fetch_result.balance_sheet.keys())[:3] if fetch_result.balance_sheet else [] } # 步骤2: 计算财务比率 ratio_input RatioCalculationInput( income_stmtfetch_result.income_stmt, balance_sheetfetch_result.balance_sheet, cash_flowfetch_result.cash_flow ) ratio_result await calculate_ratios_skill(ratio_input.dict()) final_result[ratios] [ratio.dict() for ratio in ratio_result.ratios] final_result[warnings] ratio_result.warnings # 步骤3: 生成分析报告 # 准备报告生成的输入数据 report_input ReportGenerationInput( symbolsymbol, company_info{k: fetch_result.info.get(k, N/A) for k in [longName, sector, industry]}, ratiosfinal_result[ratios], raw_data_snapshotfinal_result[raw_data] ) report_result await generate_report_skill(report_input.dict()) final_result[analysis_report] report_result.analysis_report final_result[report_model] report_result.model_used # 记录成功结果到Langfuse langfuse_context.score_current_trace( nameagent_execution_success, value1, commentAgent executed successfully and generated a report. ) except Exception as e: # 记录失败信息到Langfuse langfuse_context.score_current_trace( nameagent_execution_failure, value0, commentfAgent execution failed: {str(e)} ) final_result[error] str(e) raise e # 可以选择重新抛出异常或处理异常 finally: # 确保Langfuse数据发送出去 langfuse.flush() return final_result # 为了方便直接运行提供一个异步入口函数 async def analyze_company(symbol: str AAPL): 分析指定公司的财务状况 agent FinancialAnalysisAgent() print(f开始分析 {symbol}...) result await agent.execute(symbol) print(\n *50) print(f【{symbol} 财务分析完成】) print(*50) print(\n生成的报告) print(result.get(analysis_report, No report generated.)) print(\n计算出的比率) for ratio in result.get(ratios, []): print(f - {ratio[name]}: {ratio[value]} ({ratio[description]})) if result.get(warnings): print(\n警告) for warn in result[warnings]: print(f - {warn}) return result if __name__ __main__: # 示例分析苹果公司 asyncio.run(analyze_company(AAPL))关键点解析Agent类 继承了Harness的Agent基类execute方法定义了智能体的核心工作流。Langfuse集成observe装饰器自动将execute方法包装为一个Trace。langfuse_context.update_current_trace用于设置Trace的元数据便于后续筛选和查看。langfuse_context.score_current_trace用于对本次Trace的执行结果进行评分成功1失败0。在实际项目中评分逻辑可以更复杂比如基于报告质量。langfuse.flush()确保追踪数据被发送到Langfuse服务器。工作流清晰execute方法顺序调用了三个Skill并传递数据。这体现了Harness将复杂任务分解为可复用Skill的思想。6. 运行、验证与在Langfuse中查看结果6.1 运行智能体在项目根目录下直接运行Agent主文件python agent/financial_analysis_agent.py如果一切配置正确你将看到类似以下的输出开始分析 AAPL... 【AAPL 财务分析完成】 生成的报告 【财务分析报告】AAPL 摘要苹果公司(AAPL)在截至2023年9月的财年中展现了强大的盈利能力和财务稳健性... 盈利能力分析毛利率达到XX%处于科技硬件行业领先水平... 偿债能力分析资产负债率为XX%财务结构健康... 运营效率分析... 总体评价与建议... 本报告基于公开财务数据生成不构成投资建议。 计算出的比率 - Gross Margin: 45.2 (毛利率反映核心业务的盈利能力。) - Net Margin: 25.8 (净利率反映最终盈利能力。) - Debt to Assets Ratio: 35.1 (资产负债率反映资产中有多少由债务构成。) - Asset Turnover: 0.92 (资产周转率反映资产创造收入的效率。)6.2 在Langfuse平台验证追踪数据登录你的Langfuse Cloud控制台或自部署实例。进入“Traces”页面。你应该能看到一条名为”Finance Analysis for AAPL“的Trace记录。点击进入这条Trace你将看到完整的执行流水线根节点 (Root):financial_analysis_agent_execution包含了整个执行的耗时和元数据。子节点 (Observations):三次LLM调用如果Skill内部有调用或者报告生成Skill的调用会被自动捕获。你的三个Skill的函数调用因为被observe或类似机制装饰。Langfuse SDK会自动记录函数的输入输出。查看“Scores”标签页可以看到我们代码中设置的agent_execution_success评分。你可以在“Analytics”页面查看所有Trace的耗时分布、成本统计如果配置了模型单价和评分趋势。这才是AI工程化的核心价值每一次执行都有据可查你可以精确知道是哪个环节慢了、贵了、或者出错了。7. 常见问题与排查思路在实践过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行时报ModuleNotFoundError: No module named ‘harness‘harness-core包未正确安装或虚拟环境未激活。1. 检查当前Python环境python --version。2. 使用 pip listgrep harness 查看是否安装。Langfuse Trace在控制台看不到1. API密钥或Host配置错误。2. 网络问题。3. 代码中未调用langfuse.flush()。1. 检查.env文件配置。2. 在代码中添加print(os.getenv(‘LANGFUSE_SECRET_KEY‘))验证。3. 查看Langfuse控制台是否有错误日志。1. 核对并修正.env文件。2. 确保代码执行路径中调用了flush()。3. 检查防火墙或网络代理设置。yfinance无法获取数据或返回空1. 股票代码错误。2. 网络连接问题。3. Yahoo Finance API限制或变更。1. 在浏览器中访问 Yahoo Finance 验证代码。2. 尝试其他代码如”MSFT“。3. 查看yfinance库的报错信息。1. 使用正确的股票代码。2. 添加重试机制或使用备用数据源。3. 考虑使用更稳定的金融数据API如Alpha Vantage需注册API Key。OpenAI API调用失败或超时1. API Key无效或余额不足。2. 网络问题。3. 请求速率超限。1. 在OpenAI平台检查API Key状态和用量。2. 尝试简单的curl命令测试连通性。3. 查看OpenAI返回的错误信息。1. 更换有效API Key并充值。2. 增加超时设置timeout30。3. 实现指数退避重试逻辑。生成的报告质量不高或格式不对提示词Prompt设计不佳。1. 在Langfuse中查看发送给LLM的完整Prompt。2. 分析LLM的原始输出。1. 迭代优化Prompt增加更具体的指令和示例。2. 使用Langfuse的评分功能收集反馈并迭代Prompt。Harness Skill无法被Agent找到Skill未在Harness框架中正确注册。1. 确保Skill文件被导入。2. 检查skill装饰器的参数是否正确。1. 在Agent执行前确保所有Skill模块已被Python解释器加载通过import。2. 查阅Harness文档确认是否需要显式注册到某个Registry。8. 企业级最佳实践与进阶建议将上述Demo提升到企业级生产环境还需要考虑以下方面8.1 配置管理与安全使用配置中心 不要将API密钥等敏感信息硬编码或放在.env文件中提交。应使用AWS Secrets Manager、HashiCorp Vault或Azure Key Vault等专业秘密管理服务。环境隔离 为开发、测试、生产环境配置不同的Langfuse项目和OpenAI API Key。8.2 性能与可靠性异步与并发 Harness天然支持异步。确保你的Skill是async函数并在Agent中合理使用asyncio.gather来并发执行无依赖的任务以提升性能。缓存 对获取的财务数据等不变或低频变的数据进行缓存如使用Redis避免重复调用外部API节省成本和延迟。重试与降级 为网络调用如yfinance, OpenAI添加重试机制如tenacity库。为关键Skill设计降级方案例如当报告生成失败时返回结构化的数据表格。8.3 可观测性与评估深化自定义评分 不要只满足于“成功/失败”评分。利用Langfuse的SDK在报告生成后可以调用另一个LLM或规则系统对报告的相关性、完整性、准确性进行自动评分并记录。创建数据集 在Langfuse中创建一个“财务分析测试集”包含20-30个不同行业、不同规模的上市公司股票代码。定期如每天运行你的Agent自动化评估其表现监控效果波动。链路追踪 利用Langfuse的Trace功能深入分析每个Skill的耗时和Token消耗为性能优化和成本控制提供数据支持。8.4 部署与运维容器化 使用Docker将你的智能体应用容器化确保环境一致性。服务化 使用FastAPI或Flask将你的Agent包装成RESTful API服务供其他系统调用。CI/CD 将代码仓库与CI/CD管道如GitHub Actions, GitLab CI集成实现自动化测试和部署。可以在CI中运行针对测试集的评估只有评分达标才允许部署到生产环境。通过以上步骤你构建的不仅仅是一个能分析财务的脚本而是一个具备标准化构建、全链路可观测、持续可评估能力的生产级AI应用。这才是Harness和Langfuse带来的真正范式转变让AI应用的开发和管理变得像传统软件工程一样有序和可靠。 基于Harness与Langfuse的企业级财务分析智能体实战：从原型到生产
如果你正在尝试将AI能力应用到企业财务分析这类严肃业务中大概率会遇到这样的困境模型生成的财务报告看似专业但数据来源不明、逻辑链条断裂你无法向业务部门解释“这个结论是怎么来的”或者你费尽心思搭建的智能体在生产环境跑了几次就“失忆”或“胡言乱语”却找不到问题出在哪个环节。这背后不是模型能力的问题而是AI工程化的缺失。我们习惯了在Jupyter Notebook里快速验证一个想法但要将它变成企业里稳定、可评估、可迭代的生产系统中间隔着一道巨大的鸿沟。今天要讨论的“Harness架构”与“Langfuse”正是填平这道鸿沟的关键工具组合。它们一个负责智能体的标准化生产与部署Harness一个负责全链路追踪与评估Langfuse。本文不会空谈概念我们将通过一个企业级财务分析智能体的完整项目实战带你一步步实现从本地原型到可评估、可运维的生产化落地。你会看到如何用代码构建一个能理解财报、进行比率分析、生成洞察报告的智能体并在此过程中深刻理解Harness的“工程化”思想和Langfuse的“可观测性”价值。1. 这篇文章真正要解决的问题从AI玩具到生产工具为什么很多AI项目停留在演示阶段核心在于缺乏标准化流水线和效果度量体系。想象一下传统软件开发我们有Git做版本控制Jenkins做CI/CDPrometheus做监控日志系统做排错。这套体系确保了软件的可重复构建、自动化测试和稳定运行。而当前的AI应用开发尤其是基于大语言模型的智能体开发很大程度上还处在“手工作坊”时代提示词写在笔记本里上下文管理靠代码硬编码每次调优靠人工对比输出上线后效果波动无从追溯。这就是Harness和Langfuse要解决的问题Harness它提供了一套用于构建、编排和部署AI智能体Agent的标准化框架。你可以把它理解为AI智能体的“Kubernetes”或“Spring Framework”。它定义了智能体Agent、技能Skill、工作流Workflow等核心概念让开发变得模块化、可配置。Langfuse它是一个开源的LLM应用可观测性平台。它自动追踪每次AI调用的输入Prompt、输出Completion、耗时、成本并能通过手动或自动评分进行效果评估。它是你AI应用的“黑匣子”和“评分器”。本实战项目将模拟一个经典场景自动财务分析智能体。给定一家上市公司的股票代码或名称智能体将自动完成以下任务从公开数据源如Yahoo Finance获取财务数据。计算关键财务比率如毛利率、净利率、资产负债率等。结合行业基准和历史趋势生成一份结构化的分析报告。整个过程的每一步都被Harness清晰地编排并被Langfuse完整地追踪和评估。通过这个项目你将掌握的不再是零散的API调用而是一套让AI应用变得可靠、可解释、可优化的工程化方法。2. 核心概念与架构解析在动手之前必须厘清几个关键概念否则代码只是空中楼阁。2.1 Harness 核心概念像组装乐高一样构建智能体Harness将复杂的智能体系统分解为几个层次分明的组件Skill技能最基础的执行单元。一个Skill完成一件特定、原子性的任务。例如“获取股票财务数据”、“计算流动比率”、“生成自然语言摘要”。每个Skill通常包含清晰的输入、输出定义和执行逻辑可以是调用LLM也可以是纯计算。Agent智能体由一个或多个Skill组装而成能够完成一个更复杂的、有明确目标的任务。例如“财务分析智能体”就是由“数据获取Skill”、“比率计算Skill”、“报告生成Skill”组合而成。Agent负责管理Skill之间的执行顺序和数据流转。Workflow工作流用于编排多个Agent或更复杂任务的执行流程。它可以包含条件分支、循环等逻辑。在财务分析场景中一个工作流可以先后调用“数据获取Agent”和“报告生成Agent”。Harness 指代整个框架或运行时环境。它提供了Skill和Agent的注册、发现、执行引擎以及与其他系统如Langfuse集成的能力。通俗理解Skill是“拧螺丝”、“装轮子”的工人Agent是能“组装一辆自行车”的班组Workflow是“先组装车架再喷漆最后质检”的工厂流水线Harness就是整座“工厂”的管理系统。2.2 Langfuse 核心价值为AI应用装上“仪表盘”Langfuse关注的是应用运行时的“可观测性”Trace追踪一次完整的用户交互或任务执行过程。例如用户询问“分析一下AAPL的财务状况”从接收到请求到返回最终报告这整个过程构成一个Trace。Observation观测点Trace中的每一个步骤比如一次LLM调用、一次工具调用、一次数据查询都是一个Observation。Langfuse会自动记录它们的输入、输出、开始时间、结束时间、消耗的Token数、成本等。Score评分对Trace或Observation的质量进行量化评价。可以是人工打分比如业务专家审阅报告后给分也可以是自动评分比如用另一个模型检查报告是否包含关键指标。Dataset Evaluation数据集与评估你可以将标准的测试用例如“分析AAPL”、“分析MSFT”定义为数据集然后定期或批量运行你的智能体用Langfuse自动收集结果并评分形成效果评估报告。关键洞察没有Langfuse你的智能体就是一个“盲盒”出了问题只能靠猜。有了Langfuse每一次调用都透明可见每一次迭代都有数据支撑。2.3 项目整体架构图用户请求 | v [ Harness 网关 / API ] | v [ 财务分析 Agent (Harness) ] |-----------------------| | | v v [数据获取 Skill] [比率计算 Skill] [报告生成 Skill] | | | v v v 调用Yahoo Finance API 执行Python计算函数 调用LLM API (如GPT-4) | | | | | | |-----------------------|-----------------------| | v 生成最终财务报告 | v [ Langfuse SDK 集成 ] | v 全程自动记录 Trace, Observation, Token用量成本... | v 可选人工或自动评分 (Score)这个架构确保了功能解耦Harness和过程透明Langfuse。3. 环境准备与项目初始化我们使用Python作为开发语言。请确保你的环境满足以下要求。3.1 基础环境要求操作系统 macOS / Linux / Windows (WSL2推荐)Python版本 3.9 或以上包管理工具 pip 或 conda3.2 关键依赖安装创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir finance_ai_agent cd finance_ai_agent # 创建并激活虚拟环境 (以venv为例) python -m venv venv # Windows venv\Scripts\activate # macOS/Linux source venv/bin/activate # 安装核心框架和工具 pip install harness-core # Harness核心框架 pip install langfuse # Langfuse Python SDK pip install openai # 用于调用GPT模型 pip install yfinance # 用于获取雅虎财经数据 pip install pandas numpy # 数据处理 pip install python-dotenv # 管理环境变量重要提示harness-core是Harness框架的核心Python库。网络热词中提到的deepseek harness可能指特定版本或发行版在本文中我们以通用的harness-core概念和模式进行讲解其工程化思想是相通的。3.3 获取并配置API密钥本项目需要OpenAI和Langfuse的API密钥。OpenAI API Key: 访问 OpenAI平台 创建。Langfuse API Key: 你有两种选择云服务 注册 Langfuse Cloud 。自托管 按照官方文档部署Langfuse服务涉及Docker, PostgreSQL等。 获取你的LANGFUSE_SECRET_KEY,LANGFUSE_PUBLIC_KEY和LANGFUSE_HOST。在项目根目录创建.env文件填入你的密钥# .env 文件 OPENAI_API_KEYsk-your-openai-api-key-here # Langfuse 配置 (如果使用云服务HOST通常是 https://cloud.langfuse.com) LANGFUSE_SECRET_KEYsk-lf-your-langfuse-secret-key LANGFUSE_PUBLIC_KEYpk-lf-your-langfuse-public-key LANGFUSE_HOSThttps://cloud.langfuse.com安全警告 务必在.gitignore文件中加入.env切勿将包含密钥的文件提交到版本控制系统。4. 构建财务分析智能体的核心Skills我们将遵循Harness的架构先构建三个核心Skill。4.1 Skill 1: 财务数据获取 (FetchFinancialDataSkill)这个Skill负责从Yahoo Finance获取原始的财务数据。# skills/fetch_financial_data.py import yfinance as yf from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import Dict, Any class FinancialDataInput(SkillInput): 获取财务数据Skill的输入模型 symbol: str Field(..., description上市公司股票代码例如AAPL, MSFT) class FinancialDataOutput(SkillOutput): 获取财务数据Skill的输出模型 income_stmt: Dict[str, Any] Field(default_factorydict, description利润表数据) balance_sheet: Dict[str, Any] Field(default_factorydict, description资产负债表数据) cash_flow: Dict[str, Any] Field(default_factorydict, description现金流量表数据) info: Dict[str, Any] Field(default_factorydict, description公司基本信息) skill( namefetch_financial_data, description根据股票代码从雅虎财经获取利润表、资产负债表和现金流量表数据。, input_modelFinancialDataInput, output_modelFinancialDataOutput ) async def fetch_financial_data_skill(input_data: FinancialDataInput) - FinancialDataOutput: 财务数据获取技能 ticker yf.Ticker(input_data.symbol) # 获取最近一个财年的数据 income_stmt ticker.income_stmt balance_sheet ticker.balance_sheet cash_flow_stmt ticker.cashflow info ticker.info # 将DataFrame转换为字典以便序列化传输 output_data FinancialDataOutput( income_stmtincome_stmt.to_dict() if income_stmt is not None else {}, balance_sheetbalance_sheet.to_dict() if balance_sheet is not None else {}, cash_flowcash_flow_stmt.to_dict() if cash_flow_stmt is not None else {}, infoinfo ) return output_data关键点解析skill装饰器将普通函数注册为Harness框架可识别的Skill。SkillInput和SkillOutput继承自BaseModel明确定义了技能的“接口契约”这是工程化的关键确保了类型安全和自描述性。函数内部使用yfinance库获取数据并进行简单的格式转换。4.2 Skill 2: 财务比率计算 (CalculateRatiosSkill)这个Skill接收上一个Skill的财务数据计算关键财务比率。# skills/calculate_ratios.py from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import Dict, Any, List class RatioCalculationInput(SkillInput): 计算财务比率Skill的输入模型 income_stmt: Dict[str, Any] Field(..., description利润表数据) balance_sheet: Dict[str, Any] Field(..., description资产负债表数据) cash_flow: Dict[str, Any] Field(default_factorydict, description现金流量表数据) class FinancialRatio(BaseModel): 单个财务比率的模型 name: str value: float description: str formula: str class RatioCalculationOutput(SkillOutput): 计算财务比率Skill的输出模型 ratios: List[FinancialRatio] Field(default_factorylist, description计算出的财务比率列表) warnings: List[str] Field(default_factorylist, description计算过程中的警告信息) def safe_get(data_dict: Dict[str, Any], key: str, default: float 0.0) - float: 安全地从嵌套字典中获取数值避免KeyError。 # 简化实现实际项目可能需要更复杂的逻辑处理DataFrame转换后的结构 if isinstance(data_dict, dict): # 假设数据格式为 {‘2023-12-31‘: 100000, ‘2022-12-31‘: 90000} # 我们取最近一年的数据 for date_key, value in data_dict.items(): if isinstance(value, (int, float)): return float(value) return default skill( namecalculate_financial_ratios, description基于利润表和资产负债表计算关键财务比率如盈利能力、偿债能力、运营效率比率。, input_modelRatioCalculationInput, output_modelRatioCalculationOutput ) async def calculate_ratios_skill(input_data: RatioCalculationInput) - RatioCalculationOutput: 财务比率计算技能 income_stmt input_data.income_stmt balance_sheet input_data.balance_sheet ratios [] warnings [] # 1. 盈利能力比率 revenue safe_get(income_stmt, Total Revenue) gross_profit safe_get(income_stmt, Gross Profit) net_income safe_get(income_stmt, Net Income) total_assets safe_get(balance_sheet, Total Assets) if revenue 0: gross_margin (gross_profit / revenue) * 100 ratios.append(FinancialRatio( nameGross Margin, valueround(gross_margin, 2), description毛利率反映核心业务的盈利能力。, formulaGross Profit / Total Revenue )) net_margin (net_income / revenue) * 100 ratios.append(FinancialRatio( nameNet Margin, valueround(net_margin, 2), description净利率反映最终盈利能力。, formulaNet Income / Total Revenue )) else: warnings.append(营收数据为0或缺失无法计算利润率。) # 2. 偿债能力比率 total_liabilities safe_get(balance_sheet, Total Liabilities Net Minority Interest) total_equity safe_get(balance_sheet, Total Equity Gross Minority Interest) if total_assets 0: debt_to_assets (total_liabilities / total_assets) * 100 ratios.append(FinancialRatio( nameDebt to Assets Ratio, valueround(debt_to_assets, 2), description资产负债率反映资产中有多少由债务构成。, formulaTotal Liabilities / Total Assets )) # 3. 运营效率比率 (示例资产周转率) if total_assets 0 and revenue 0: asset_turnover revenue / total_assets ratios.append(FinancialRatio( nameAsset Turnover, valueround(asset_turnover, 2), description资产周转率反映资产创造收入的效率。, formulaTotal Revenue / Total Assets )) return RatioCalculationOutput(ratiosratios, warningswarnings)关键点解析定义了FinancialRatio这个数据模型使输出高度结构化。safe_get函数处理了从原始数据字典中安全提取数值的逻辑这是工程健壮性的体现。计算了典型的财务比率并附上了描述和公式增强了结果的可解释性。4.3 Skill 3: 生成分析报告 (GenerateReportSkill)这个Skill将前两个Skill的结果整合调用LLM生成一份易于理解的财务分析报告。# skills/generate_report.py import os from openai import OpenAI from harness.core.skill import skill from harness.core.models import SkillInput, SkillOutput from pydantic import BaseModel, Field from typing import List, Dict, Any from dotenv import load_dotenv load_dotenv() # 加载环境变量 class ReportGenerationInput(SkillInput): 生成报告Skill的输入模型 symbol: str Field(..., description股票代码) company_info: Dict[str, Any] Field(default_factorydict, description公司基本信息) ratios: List[Dict[str, Any]] Field(default_factorylist, description财务比率列表) raw_data_snapshot: Dict[str, Any] Field(default_factorydict, description原始数据摘要) class ReportGenerationOutput(SkillOutput): 生成报告Skill的输出模型 analysis_report: str Field(..., description生成的财务分析报告文本) model_used: str Field(defaultgpt-4, description生成报告所使用的LLM模型) skill( namegenerate_financial_analysis_report, description基于财务数据和计算出的比率生成一份结构化的、易于理解的财务分析报告。, input_modelReportGenerationInput, output_modelReportGenerationOutput ) async def generate_report_skill(input_data: ReportGenerationInput) - ReportGenerationOutput: 财务分析报告生成技能 client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 构建给LLM的提示词 prompt f 你是一名资深的财务分析师。请根据以下信息为股票代码为 {input_data.symbol} 的公司生成一份专业的财务分析报告。 公司基本信息 {input_data.company_info} 计算出的关键财务比率 {input_data.ratios} 原始数据摘要 {input_data.raw_data_snapshot} 报告要求 1. 以“【财务分析报告】{input_data.symbol}”为标题。 2. 报告结构应包含摘要、盈利能力分析、偿债能力分析、运营效率分析、总体评价与建议。 3. 分析要结合比率数值给出定性判断例如优秀、良好、一般、需警惕。 4. 语言专业、简洁、客观。 5. 在报告末尾注明“本报告基于公开财务数据生成不构成投资建议。” try: response client.chat.completions.create( modelgpt-4-turbo-preview, # 可根据需要调整模型 messages[ {role: system, content: 你是一名专业的财务分析师。}, {role: user, content: prompt} ], temperature0.2, # 低温度值使输出更稳定、更聚焦 max_tokens1500 ) report_content response.choices[0].message.content return ReportGenerationOutput( analysis_reportreport_content, model_usedgpt-4-turbo-preview ) except Exception as e: # 异常处理返回一个包含错误信息的报告 error_report f【财务分析报告生成失败】{input_data.symbol} 抱歉在生成分析报告时遇到错误{str(e)} 已获取的数据和比例如下可供手动分析 公司信息{input_data.company_info} 财务比率{input_data.ratios} return ReportGenerationOutput( analysis_reporterror_report, model_usederror )关键点解析集成了OpenAI API使用Chat Completion模式。精心构造了提示词Prompt明确规定了报告的结构、内容和风格。这是保证LLM输出质量稳定的关键。加入了异常处理即使LLM调用失败也能返回有意义的输出保证了整个Agent工作流的鲁棒性。5. 组装Agent与集成Langfuse追踪现在我们将三个Skill组装成一个完整的财务分析Agent并集成Langfuse进行全链路追踪。5.1 创建主Agent并集成Langfuse# agent/financial_analysis_agent.py import asyncio import os from typing import Dict, Any from langfuse import Langfuse from langfuse.decorators import observe, langfuse_context # 注意检查你的langfuse版本是否支持decorators from dotenv import load_dotenv from harness.core.agent import Agent from skills.fetch_financial_data import fetch_financial_data_skill, FinancialDataOutput from skills.calculate_ratios import calculate_ratios_skill, RatioCalculationInput from skills.generate_report import generate_report_skill, ReportGenerationInput load_dotenv() # 初始化Langfuse客户端 langfuse Langfuse( secret_keyos.getenv(LANGFUSE_SECRET_KEY), public_keyos.getenv(LANGFUSE_PUBLIC_KEY), hostos.getenv(LANGFUSE_HOST) ) class FinancialAnalysisAgent(Agent): 财务分析智能体 def __init__(self): super().__init__(namefinancial_analysis_agent) # 注册该Agent所拥有的Skills (虽然Skill已全局注册这里可以声明依赖) self.required_skills [fetch_financial_data, calculate_financial_ratios, generate_financial_analysis_report] observe(namefinancial_analysis_agent_execution) # Langfuse装饰器自动追踪此函数 async def execute(self, symbol: str) - Dict[str, Any]: 执行财务分析流程 # 设置Langfuse Trace的元数据 langfuse_context.update_current_trace( namefFinance Analysis for {symbol}, user_iddemo_user, # 在实际应用中这里可以传入真实用户ID metadata{symbol: symbol, agent_version: 1.0} ) final_result {} try: # 步骤1: 获取财务数据 data_input {symbol: symbol} fetch_result: FinancialDataOutput await fetch_financial_data_skill(data_input) final_result[raw_data] { info: fetch_result.info, income_stmt_keys: list(fetch_result.income_stmt.keys())[:3] if fetch_result.income_stmt else [], balance_sheet_keys: list(fetch_result.balance_sheet.keys())[:3] if fetch_result.balance_sheet else [] } # 步骤2: 计算财务比率 ratio_input RatioCalculationInput( income_stmtfetch_result.income_stmt, balance_sheetfetch_result.balance_sheet, cash_flowfetch_result.cash_flow ) ratio_result await calculate_ratios_skill(ratio_input.dict()) final_result[ratios] [ratio.dict() for ratio in ratio_result.ratios] final_result[warnings] ratio_result.warnings # 步骤3: 生成分析报告 # 准备报告生成的输入数据 report_input ReportGenerationInput( symbolsymbol, company_info{k: fetch_result.info.get(k, N/A) for k in [longName, sector, industry]}, ratiosfinal_result[ratios], raw_data_snapshotfinal_result[raw_data] ) report_result await generate_report_skill(report_input.dict()) final_result[analysis_report] report_result.analysis_report final_result[report_model] report_result.model_used # 记录成功结果到Langfuse langfuse_context.score_current_trace( nameagent_execution_success, value1, commentAgent executed successfully and generated a report. ) except Exception as e: # 记录失败信息到Langfuse langfuse_context.score_current_trace( nameagent_execution_failure, value0, commentfAgent execution failed: {str(e)} ) final_result[error] str(e) raise e # 可以选择重新抛出异常或处理异常 finally: # 确保Langfuse数据发送出去 langfuse.flush() return final_result # 为了方便直接运行提供一个异步入口函数 async def analyze_company(symbol: str AAPL): 分析指定公司的财务状况 agent FinancialAnalysisAgent() print(f开始分析 {symbol}...) result await agent.execute(symbol) print(\n *50) print(f【{symbol} 财务分析完成】) print(*50) print(\n生成的报告) print(result.get(analysis_report, No report generated.)) print(\n计算出的比率) for ratio in result.get(ratios, []): print(f - {ratio[name]}: {ratio[value]} ({ratio[description]})) if result.get(warnings): print(\n警告) for warn in result[warnings]: print(f - {warn}) return result if __name__ __main__: # 示例分析苹果公司 asyncio.run(analyze_company(AAPL))关键点解析Agent类 继承了Harness的Agent基类execute方法定义了智能体的核心工作流。Langfuse集成observe装饰器自动将execute方法包装为一个Trace。langfuse_context.update_current_trace用于设置Trace的元数据便于后续筛选和查看。langfuse_context.score_current_trace用于对本次Trace的执行结果进行评分成功1失败0。在实际项目中评分逻辑可以更复杂比如基于报告质量。langfuse.flush()确保追踪数据被发送到Langfuse服务器。工作流清晰execute方法顺序调用了三个Skill并传递数据。这体现了Harness将复杂任务分解为可复用Skill的思想。6. 运行、验证与在Langfuse中查看结果6.1 运行智能体在项目根目录下直接运行Agent主文件python agent/financial_analysis_agent.py如果一切配置正确你将看到类似以下的输出开始分析 AAPL... 【AAPL 财务分析完成】 生成的报告 【财务分析报告】AAPL 摘要苹果公司(AAPL)在截至2023年9月的财年中展现了强大的盈利能力和财务稳健性... 盈利能力分析毛利率达到XX%处于科技硬件行业领先水平... 偿债能力分析资产负债率为XX%财务结构健康... 运营效率分析... 总体评价与建议... 本报告基于公开财务数据生成不构成投资建议。 计算出的比率 - Gross Margin: 45.2 (毛利率反映核心业务的盈利能力。) - Net Margin: 25.8 (净利率反映最终盈利能力。) - Debt to Assets Ratio: 35.1 (资产负债率反映资产中有多少由债务构成。) - Asset Turnover: 0.92 (资产周转率反映资产创造收入的效率。)6.2 在Langfuse平台验证追踪数据登录你的Langfuse Cloud控制台或自部署实例。进入“Traces”页面。你应该能看到一条名为”Finance Analysis for AAPL“的Trace记录。点击进入这条Trace你将看到完整的执行流水线根节点 (Root):financial_analysis_agent_execution包含了整个执行的耗时和元数据。子节点 (Observations):三次LLM调用如果Skill内部有调用或者报告生成Skill的调用会被自动捕获。你的三个Skill的函数调用因为被observe或类似机制装饰。Langfuse SDK会自动记录函数的输入输出。查看“Scores”标签页可以看到我们代码中设置的agent_execution_success评分。你可以在“Analytics”页面查看所有Trace的耗时分布、成本统计如果配置了模型单价和评分趋势。这才是AI工程化的核心价值每一次执行都有据可查你可以精确知道是哪个环节慢了、贵了、或者出错了。7. 常见问题与排查思路在实践过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行时报ModuleNotFoundError: No module named ‘harness‘harness-core包未正确安装或虚拟环境未激活。1. 检查当前Python环境python --version。2. 使用 pip listgrep harness 查看是否安装。Langfuse Trace在控制台看不到1. API密钥或Host配置错误。2. 网络问题。3. 代码中未调用langfuse.flush()。1. 检查.env文件配置。2. 在代码中添加print(os.getenv(‘LANGFUSE_SECRET_KEY‘))验证。3. 查看Langfuse控制台是否有错误日志。1. 核对并修正.env文件。2. 确保代码执行路径中调用了flush()。3. 检查防火墙或网络代理设置。yfinance无法获取数据或返回空1. 股票代码错误。2. 网络连接问题。3. Yahoo Finance API限制或变更。1. 在浏览器中访问 Yahoo Finance 验证代码。2. 尝试其他代码如”MSFT“。3. 查看yfinance库的报错信息。1. 使用正确的股票代码。2. 添加重试机制或使用备用数据源。3. 考虑使用更稳定的金融数据API如Alpha Vantage需注册API Key。OpenAI API调用失败或超时1. API Key无效或余额不足。2. 网络问题。3. 请求速率超限。1. 在OpenAI平台检查API Key状态和用量。2. 尝试简单的curl命令测试连通性。3. 查看OpenAI返回的错误信息。1. 更换有效API Key并充值。2. 增加超时设置timeout30。3. 实现指数退避重试逻辑。生成的报告质量不高或格式不对提示词Prompt设计不佳。1. 在Langfuse中查看发送给LLM的完整Prompt。2. 分析LLM的原始输出。1. 迭代优化Prompt增加更具体的指令和示例。2. 使用Langfuse的评分功能收集反馈并迭代Prompt。Harness Skill无法被Agent找到Skill未在Harness框架中正确注册。1. 确保Skill文件被导入。2. 检查skill装饰器的参数是否正确。1. 在Agent执行前确保所有Skill模块已被Python解释器加载通过import。2. 查阅Harness文档确认是否需要显式注册到某个Registry。8. 企业级最佳实践与进阶建议将上述Demo提升到企业级生产环境还需要考虑以下方面8.1 配置管理与安全使用配置中心 不要将API密钥等敏感信息硬编码或放在.env文件中提交。应使用AWS Secrets Manager、HashiCorp Vault或Azure Key Vault等专业秘密管理服务。环境隔离 为开发、测试、生产环境配置不同的Langfuse项目和OpenAI API Key。8.2 性能与可靠性异步与并发 Harness天然支持异步。确保你的Skill是async函数并在Agent中合理使用asyncio.gather来并发执行无依赖的任务以提升性能。缓存 对获取的财务数据等不变或低频变的数据进行缓存如使用Redis避免重复调用外部API节省成本和延迟。重试与降级 为网络调用如yfinance, OpenAI添加重试机制如tenacity库。为关键Skill设计降级方案例如当报告生成失败时返回结构化的数据表格。8.3 可观测性与评估深化自定义评分 不要只满足于“成功/失败”评分。利用Langfuse的SDK在报告生成后可以调用另一个LLM或规则系统对报告的相关性、完整性、准确性进行自动评分并记录。创建数据集 在Langfuse中创建一个“财务分析测试集”包含20-30个不同行业、不同规模的上市公司股票代码。定期如每天运行你的Agent自动化评估其表现监控效果波动。链路追踪 利用Langfuse的Trace功能深入分析每个Skill的耗时和Token消耗为性能优化和成本控制提供数据支持。8.4 部署与运维容器化 使用Docker将你的智能体应用容器化确保环境一致性。服务化 使用FastAPI或Flask将你的Agent包装成RESTful API服务供其他系统调用。CI/CD 将代码仓库与CI/CD管道如GitHub Actions, GitLab CI集成实现自动化测试和部署。可以在CI中运行针对测试集的评估只有评分达标才允许部署到生产环境。通过以上步骤你构建的不仅仅是一个能分析财务的脚本而是一个具备标准化构建、全链路可观测、持续可评估能力的生产级AI应用。这才是Harness和Langfuse带来的真正范式转变让AI应用的开发和管理变得像传统软件工程一样有序和可靠。