30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度最近在尝试将大模型能力集成到实际业务系统时很多开发者都会遇到一个共同的难题如何让大模型稳定、可靠、可控地执行复杂任务直接调用API虽然简单但面对多步骤推理、工具调用、状态管理和错误处理时代码会迅速变得臃肿且难以维护。这正是Harness 工程要解决的核心问题。它并非一个具体的开源框架而是一套构建可靠AI智能体Agent的工程化方法论和最佳实践集合。本文将为你系统拆解Harness工程的核心思想并通过一个完整的“金融大模型问答机器人”项目带你从零开始实践如何运用Harness理念结合LangChain、FastAPI、RAG等技术构建一个高可用的AI应用。无论你是刚接触Agent概念的初学者还是希望提升工程化能力的中高级开发者都能从中获得一套可复用的架构方案和实战代码。1. Harness工程AI智能体的“缰绳”与“鞍具”在深入代码之前我们必须先理解Harness原意为“马具”在AI语境下的含义。你可以把它想象成驾驭一匹强大但不可预知的赛马大模型所需要的全套装备——缰绳控制方向、鞍具提供稳定支点和脚蹬方便骑手发力。1.1 什么是Harness工程Harness工程是一套用于设计、构建和管理AI智能体Agent的软件工程实践。其核心目标是将大模型不可预测的生成能力约束和引导到可预测、可测试、可运维的业务流程中。它与传统软件工程的关键区别在于处理对象传统工程处理的是确定性的逻辑if-else, loops而Harness工程处理的是非确定性的、基于概率的大模型输出。因此它更关注流程编排将复杂任务分解为可管理的步骤Plan、Execute、Evaluate、Refine。状态管理在多个执行步骤和工具调用间持久化和管理上下文。错误处理与回退当模型产生幻觉、调用工具失败或超出约束时有明确的恢复机制。可观测性对智能体的决策过程、工具使用、耗时和成本进行全面的日志记录和监控。1.2 Harness vs. Agent vs. 框架这几个概念容易混淆我们来清晰界定一下Agent智能体一个能够感知环境、进行决策并执行动作如调用工具、生成代码以完成目标的系统。它是我们要构建的“智能实体”。Harness工程是构建和运行Agent的方法论和基础设施。它定义了Agent的能力分层、模块边界、通信协议和生命周期管理。你可以把它看作开发Agent的“脚手架”和“运维平台”。LangChain等框架提供了实现Agent所需的基础组件如LLM调用、提示模板、记忆、工具链。Harness工程会利用这些框架但更侧重于在其之上构建稳健的应用层架构。简单说你用LangChain框架提供的积木按照Harness工程蓝图和方法的指导搭建出一个可靠的Agent智能体应用。1.3 为什么需要Harness工程—— 解决Agent的“四大痛点”脆弱性一次错误的模型输出或工具调用失败可能导致整个流程崩溃。不可控性模型可能偏离预定任务产生无关内容或执行危险操作。难调试黑盒模型决策过程不透明问题定位困难。难集成如何将Agent优雅地嵌入现有业务系统管理其输入输出和状态Harness工程通过引入分层设计、标准化接口、状态机和监督流程来系统性解决这些问题。2. 项目实战金融大模型问答机器人我们将构建一个服务于金融客服或分析场景的问答机器人。它不仅能基于通用知识回答更能精准处理公司财报、金融术语、实时股价查询等专业需求。项目核心需求处理用户关于金融概念、产品的咨询。支持上传PDF格式的公司年报、研报并基于文档内容进行问答RAG。支持查询实时股票价格需要调用外部API。对话需有上下文记忆。系统需具备高可用性关键步骤有日志和错误处理。技术栈选型LLMQwen-7B-Chat (本地部署) 或 Qwen-plus (API)。兼顾效果与成本。应用框架FastAPI。轻量、异步友好便于提供HTTP接口。Agent框架LangChain。生态成熟组件丰富。向量数据库Chroma。轻量易于集成。RAGLangChain Chroma。实现文档的索引与检索。微调与优化LoRA, SFT, 量化用于后期模型效果优化。工程化核心自实现的Harness层用于管理Agent工作流。3. 环境准备与项目结构3.1 环境与版本说明建议使用Python 3.9。以下依赖版本是一个稳定的起点请根据实际情况调整。# 核心依赖 langchain0.1.0 langchain-community0.0.10 chromadb0.4.22 fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 httpx0.25.1 # 大模型相关 (以通义千问为例) # 使用DashScope API dashscope1.14.0 # 或使用兼容OpenAI API的本地部署 # openai1.3.0 (如果本地部署的Qwen服务兼容OpenAI API格式) # 工具包 pypdf3.17.4 # PDF解析 python-dotenv1.0.0 # 环境变量管理3.2 项目目录结构一个清晰的目录结构是Harness工程的良好开端。它体现了关注点分离的原则。financial_agent_harness/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── harness/ # **Harness核心层** │ │ ├── __init__.py │ │ ├── agent_harness.py # Agent流程编排与状态管理 │ │ ├── state.py # 状态数据模型 │ │ └── supervisor.py # 监督与回退逻辑 │ ├── agents/ # 智能体定义层 │ │ ├── __init__.py │ │ ├── base_agent.py # 基础Agent抽象类 │ │ ├── qa_agent.py # 问答Agent │ │ └── tool_agent.py # 工具调用Agent │ ├── tools/ # 工具层 │ │ ├── __init__.py │ │ ├── stock_tool.py # 股票查询工具 │ │ └── rag_tool.py # RAG检索工具 │ ├── chains/ # 链式流程层可选复杂流程用 │ ├── models/ # 数据模型层 (Pydantic) │ ├── services/ # 业务服务层 (如RAG服务) │ │ ├── __init__.py │ │ └── rag_service.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logger.py ├── data/ # 存放上传的PDF文档 ├── storage/ # 向量数据库持久化目录 ├── tests/ # 测试目录 ├── .env.example # 环境变量示例 ├── requirements.txt # 依赖列表 └── README.md4. Harness层核心实现为Agent套上“缰绳”这是项目的核心。我们将实现一个简单的AgentHarness类它负责管理Agent的整个执行生命周期。4.1 定义执行状态State首先我们需要一个数据结构来跟踪一次Agent执行的全过程。这借鉴了状态机的思想。# app/harness/state.py from pydantic import BaseModel, Field from typing import Any, Dict, List, Optional from enum import Enum class AgentStatus(str, Enum): Agent执行状态枚举 PENDING pending PLANNING planning EXECUTING executing EVALUATING evaluating SUCCESS success FAILED failed NEEDS_HUMAN_INPUT needs_human_input class AgentStep(BaseModel): 记录Agent的每一步操作 step_id: int action: str # 如”generate_plan“, ”call_tool:stock_price“, ”generate_response“ input: Dict[str, Any] output: Dict[str, Any] timestamp: float status: AgentStatus error: Optional[str] None class AgentExecutionContext(BaseModel): Agent执行的上下文贯穿整个Harness流程 session_id: str # 会话唯一标识 user_input: str current_status: AgentStatus AgentStatus.PENDING history: List[AgentStep] Field(default_factorylist) intermediate_data: Dict[str, Any] Field(default_factorydict) # 存放中间结果如检索到的文档 final_output: Optional[str] None error_message: Optional[str] None metadata: Dict[str, Any] Field(default_factorydict) # 扩展元数据4.2 实现Harness编排器AgentHarness类是整个系统的“大脑”它按照预定义的流程Plan-Execute-Evaluate来驱动Agent。# app/harness/agent_harness.py import logging import time from typing import Optional, Callable from app.harness.state import AgentExecutionContext, AgentStatus, AgentStep from app.agents.base_agent import BaseAgent logger logging.getLogger(__name__) class AgentHarness: Agent的“缰绳”核心。 职责管理Agent执行流程维护状态处理异常提供可观测性。 def __init__(self, agent: BaseAgent, max_retries: int 2): self.agent agent self.max_retries max_retries self._current_context: Optional[AgentExecutionContext] None async def run(self, session_id: str, user_input: str) - AgentExecutionContext: 执行一次完整的Agent任务 # 1. 初始化上下文 self._current_context AgentExecutionContext( session_idsession_id, user_inputuser_input, current_statusAgentStatus.PLANNING ) logger.info(f[{session_id}] 开始处理请求: {user_input}) try: # 2. 规划阶段 (Plan) await self._step_wrapper(plan, self.agent.plan) # 3. 执行阶段 (Execute) - 可能包含多轮工具调用 while self._current_context.current_status AgentStatus.EXECUTING: await self._step_wrapper(execute, self.agent.execute) # 4. 评估与精炼阶段 (Evaluate Refine) - 可选 if self._current_context.current_status AgentStatus.EVALUATING: await self._step_wrapper(evaluate, self.agent.evaluate) # 5. 生成最终输出 if self._current_context.current_status AgentStatus.SUCCESS: await self._step_wrapper(finalize, self.agent.finalize) except Exception as e: logger.error(f[{session_id}] Harness执行失败: {e}, exc_infoTrue) self._current_context.current_status AgentStatus.FAILED self._current_context.error_message str(e) # 这里可以触发告警或降级策略 self._fallback_strategy() finally: logger.info(f[{session_id}] 请求处理完成状态: {self._current_context.current_status}) return self._current_context async def _step_wrapper(self, step_name: str, step_func: Callable): 包装每一步执行记录日志和状态 step_id len(self._current_context.history) 1 step AgentStep( step_idstep_id, actionstep_name, input{context: self._current_context.dict()}, output{}, timestamptime.time(), statusself._current_context.current_status ) self._current_context.history.append(step) logger.debug(f[{self._current_context.session_id}] 进入步骤: {step_name}) start_time time.time() try: # 调用Agent的具体步骤函数 next_status await step_func(self._current_context) # 更新上下文状态 self._current_context.current_status next_status step.output {next_status: next_status, **self._current_context.intermediate_data} step.status next_status duration time.time() - start_time logger.debug(f[{self._current_context.session_id}] 步骤 {step_name} 完成耗时: {duration:.2f}s 下一状态: {next_status}) except Exception as e: logger.error(f[{self._current_context.session_id}] 步骤 {step_name} 执行异常: {e}) step.status AgentStatus.FAILED step.error str(e) self._current_context.current_status AgentStatus.FAILED self._current_context.error_message f步骤 {step_name} 失败: {e} raise # 向上抛出由外层run方法捕获 def _fallback_strategy(self): 降级策略当Agent完全失败时返回一个友好的默认响应 if self._current_context: self._current_context.final_output 抱歉系统暂时无法处理您的请求。您可以尝试重新提问或联系人工客服。 logger.warning(f[{self._current_context.session_id}] 已触发降级策略。)4.3 实现基础Agent抽象类所有具体的Agent都应继承这个基类实现标准的生命周期方法。# app/agents/base_agent.py from abc import ABC, abstractmethod from app.harness.state import AgentExecutionContext, AgentStatus import logging logger logging.getLogger(__name__) class BaseAgent(ABC): 所有Agent的抽象基类定义Harness所需的接口 abstractmethod async def plan(self, context: AgentExecutionContext) - AgentStatus: 分析用户输入制定执行计划。 例如判断是否需要检索文档、调用工具还是直接回答。 返回下一个状态。 pass abstractmethod async def execute(self, context: AgentExecutionContext) - AgentStatus: 执行计划。可能包括调用工具、生成文本等。 返回下一个状态如 EXECUTING, EVALUATING, SUCCESS。 pass async def evaluate(self, context: AgentExecutionContext) - AgentStatus: 可选评估执行结果决定是否需要精炼。 默认直接返回 SUCCESS。 return AgentStatus.SUCCESS async def finalize(self, context: AgentExecutionContext) - AgentStatus: 生成最终输出写入 context.final_output。 # 默认实现将 intermediate_data 中的结果转为输出 if response in context.intermediate_data: context.final_output context.intermediate_data[response] return AgentStatus.SUCCESS5. 构建金融问答Agent现在我们基于上述Harness框架实现一个具体的金融问答Agent。5.1 实现工具股票查询与RAG检索首先构建Agent可以使用的“工具”。# app/tools/stock_tool.py import httpx from typing import Optional import os from pydantic import BaseModel, Field class StockQueryInput(BaseModel): symbol: str Field(description股票代码例如AAPL, 000001.SZ) async def get_stock_price(symbol: str) - Optional[float]: 模拟股票价格查询工具。 实际项目中应替换为真实的金融数据API如Alpha Vantage, Yahoo Finance等。 # 这里使用一个模拟API。真实情况需要申请API Key。 # 示例使用一个公开的测试API仅用于演示。 api_url fhttps://api.example-mock-finance.com/quote?symbol{symbol} try: async with httpx.AsyncClient(timeout10.0) as client: resp await client.get(api_url) if resp.status_code 200: data resp.json() # 假设返回格式为 {symbol: AAPL, price: 175.25} return data.get(price) else: return None except Exception as e: print(f查询股票价格失败: {e}) return None# app/services/rag_service.py from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_community.embeddings import DashScopeEmbeddings # 使用DashScope的嵌入模型 # 或 from langchain_openai import OpenAIEmbeddings from langchain.schema import Document import os from typing import List class RAGService: RAG服务封装文档加载、向量化、检索逻辑 def __init__(self, persist_directory: str ./storage/chroma_db): # 初始化嵌入模型 (使用通义千问的嵌入模型需要DASHSCOPE_API_KEY) self.embeddings DashScopeEmbeddings( modeltext-embedding-v1, # DashScope嵌入模型名称 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY) ) self.persist_directory persist_directory self.vectorstore None self._init_vectorstore() def _init_vectorstore(self): 初始化或加载已有的向量库 if os.path.exists(self.persist_directory): self.vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(f已加载已有向量库文档数: {self.vectorstore._collection.count()}) else: self.vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(创建新的向量库。) def ingest_document(self, file_path: str): 摄入PDF文档并存入向量数据库 loader PyPDFLoader(file_path) documents loader.load() # 文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) splits text_splitter.split_documents(documents) # 添加到向量库 self.vectorstore.add_documents(splits) self.vectorstore.persist() print(f文档 {file_path} 已成功摄入新增 {len(splits)} 个片段。) def search(self, query: str, k: int 4) - List[Document]: 检索与查询最相关的文档片段 if not self.vectorstore: return [] docs self.vectorstore.similarity_search(query, kk) return docs# app/tools/rag_tool.py from app.services.rag_service import RAGService from typing import List import os # 全局RAG服务实例 _rag_service None def get_rag_service() - RAGService: global _rag_service if _rag_service is None: _rag_service RAGService() return _rag_service async def retrieve_financial_docs(query: str, top_k: int 3) - List[str]: RAG检索工具。 根据用户问题从已摄入的金融文档中检索相关片段。 service get_rag_service() docs service.search(query, ktop_k) # 将Document对象转换为文本 contexts [doc.page_content for doc in docs] return contexts5.2 实现具体的问答Agent这个Agent将集成规划、工具调用和回答生成。# app/agents/qa_agent.py import re from app.agents.base_agent import BaseAgent from app.harness.state import AgentExecutionContext, AgentStatus from app.tools.stock_tool import get_stock_price from app.tools.rag_tool import retrieve_financial_docs from langchain.chat_models import ChatOpenAI from langchain.schema import SystemMessage, HumanMessage, AIMessage import os from typing import List class FinancialQAAgent(BaseAgent): def __init__(self): # 初始化LLM。这里假设本地部署的Qwen服务兼容OpenAI API格式。 # 如果使用DashScope API需使用 langchain_community.chat_models.tongyi.ChatTongyi api_base os.getenv(LLM_API_BASE, http://localhost:8000/v1) # 本地部署地址 api_key os.getenv(LLM_API_KEY, not-needed) # 本地部署可能不需要key self.llm ChatOpenAI( modelqwen-7b-chat, # 模型名称根据实际调整 openai_api_baseapi_base, openai_api_keyapi_key, temperature0.1, # 低温度保证回答稳定性 max_tokens1024 ) # 定义系统提示词约束Agent行为 self.system_prompt 你是一个专业的金融助手专门回答关于金融市场、公司财报、投资产品和金融术语的问题。 你必须严格遵守以下规则 1. 如果用户询问股票价格包含股票代码如AAPL, 00700.HK等你必须调用get_stock_price工具。 2. 如果用户的问题涉及具体的上市公司、金融产品说明书或财报概念你应该优先使用retrieve_financial_docs工具从知识库中查找信息。 3. 如果知识库中没有相关信息再基于你的通用知识回答并注明“根据通用知识”。 4. 回答需严谨、客观避免给出投资建议。 5. 如果无法确定请直接说不知道不要编造信息。 async def plan(self, context: AgentExecutionContext) - AgentStatus: 分析用户意图决定执行路径 user_input context.user_input.lower() # 简单意图识别实际项目可用更复杂的NLU模型 if re.search(r\b(stock|price|quote|股价|行情|代码)\b, user_input) and re.search(r[A-Z]{1,5}|\d{6}, user_input): context.intermediate_data[intent] query_stock_price context.intermediate_data[symbol] self._extract_symbol(user_input) print(f识别意图: 查询股票价格代码: {context.intermediate_data[symbol]}) elif re.search(r\b(财报|年报|季报|收入|利润|负债|公告|招股书|研报)\b, user_input): context.intermediate_data[intent] query_financial_doc print(f识别意图: 查询金融文档) else: context.intermediate_data[intent] general_qa print(f识别意图: 通用问答) # 进入执行阶段 return AgentStatus.EXECUTING def _extract_symbol(self, text: str) - str: 简单提取股票代码示例逻辑实际需要更健壮 # 匹配类似 AAPL, 000001.SZ, 00700.HK 的格式 match re.search(r([A-Z]{1,5})|(\d{6}\.[A-Z]{2}), text.upper()) return match.group(0) if match else UNKNOWN async def execute(self, context: AgentExecutionContext) - AgentStatus: 根据规划执行具体操作 intent context.intermediate_data.get(intent) if intent query_stock_price: symbol context.intermediate_data.get(symbol, UNKNOWN) price await get_stock_price(symbol) if price: context.intermediate_data[tool_result] f股票 {symbol} 的最新价格为 ${price:.2f} 美元。 else: context.intermediate_data[tool_result] f无法获取股票 {symbol} 的价格信息请检查代码或稍后再试。 # 执行完毕进入评估/最终化阶段 return AgentStatus.EVALUATING elif intent query_financial_doc: query context.user_input relevant_docs await retrieve_financial_docs(query, top_k3) if relevant_docs: context.intermediate_data[retrieved_context] \n\n---\n\n.join(relevant_docs[:2]) # 取前两个片段 context.intermediate_data[has_relevant_doc] True else: context.intermediate_data[has_relevant_doc] False # 无论是否检索到都进入下一步生成回答 return AgentStatus.SUCCESS # 这里简化直接进入最终化 else: # general_qa # 通用问题直接准备生成回答 return AgentStatus.SUCCESS async def finalize(self, context: AgentExecutionContext) - AgentStatus: 整合信息调用LLM生成最终回答 messages [ SystemMessage(contentself.system_prompt), HumanMessage(contentcontext.user_input) ] # 如果有工具执行结果添加到上下文 if tool_result in context.intermediate_data: messages.append(AIMessage(contentf[工具调用结果] {context.intermediate_data[tool_result]})) # 如果有检索到的文档也添加 if context.intermediate_data.get(has_relevant_doc, False): retrieved_text context.intermediate_data.get(retrieved_context, ) if retrieved_text: # 将检索到的知识作为系统提示的一部分或单独消息注入 knowledge_prompt f以下是从公司文档中检索到的相关信息请基于此回答\n{retrieved_text} messages.insert(1, SystemMessage(contentknowledge_prompt)) # 插入到系统提示后 try: response await self.llm.agenerate([messages]) answer response.generations[0][0].text context.final_output answer.strip() print(fLLM生成回答成功长度: {len(answer)}) except Exception as e: print(fLLM调用失败: {e}) context.final_output 抱歉生成回答时出现错误。 return AgentStatus.SUCCESS6. 集成FastAPI提供HTTP服务将我们的Harness和Agent封装成Web服务。# app/main.py from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional import uuid import asyncio from app.harness.agent_harness import AgentHarness from app.agents.qa_agent import FinancialQAAgent from app.harness.state import AgentExecutionContext import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(title金融大模型问答机器人 API, version1.0) # 全局Agent和Harness实例实际生产环境需要考虑并发和生命周期 _agent FinancialQAAgent() _harness AgentHarness(_agent) # 用于存储任务结果的简单内存缓存生产环境应使用Redis等 _task_results {} class QueryRequest(BaseModel): question: str session_id: Optional[str] None # 可选用于多轮对话 class QueryResponse(BaseModel): session_id: str answer: str status: str request_id: str app.post(/query, response_modelQueryResponse) async def query_finance_agent(request: QueryRequest, background_tasks: BackgroundTasks): 主要查询端点。 由于Agent处理可能耗时我们采用异步处理立即返回任务ID。 session_id request.session_id or str(uuid.uuid4()) request_id str(uuid.uuid4()) # 立即返回告知请求已接受 initial_response QueryResponse( session_idsession_id, answer, statusprocessing, request_idrequest_id ) # 将实际处理任务放入后台 background_tasks.add_task( process_query_task, session_idsession_id, questionrequest.question, request_idrequest_id ) return initial_response app.get(/result/{request_id}) async def get_result(request_id: str): 根据request_id获取处理结果 if request_id not in _task_results: raise HTTPException(status_code404, detail任务未找到或仍在处理中) result _task_results[request_id] return result async def process_query_task(session_id: str, question: str, request_id: str): 后台处理任务 try: logger.info(f开始处理任务 {request_id}, 会话: {session_id}, 问题: {question}) # 运行Harness context await _harness.run(session_id, question) # 存储结果 _task_results[request_id] { session_id: session_id, answer: context.final_output or 无回答生成。, status: context.current_status.value, request_id: request_id, detail: { history_steps: len(context.history), error: context.error_message } } logger.info(f任务 {request_id} 处理完成状态: {context.current_status}) # 可选设置结果过期简易清理 await asyncio.sleep(300) # 5分钟后清理 if request_id in _task_results: del _task_results[request_id] except Exception as e: logger.error(f处理任务 {request_id} 时发生未捕获异常: {e}) _task_results[request_id] { session_id: session_id, answer: 系统内部错误请稍后重试。, status: failed, request_id: request_id, detail: {error: str(e)} } # 文档上传接口用于RAG知识库更新 from fastapi import File, UploadFile import shutil import os UPLOAD_DIR ./data/uploads os.makedirs(UPLOAD_DIR, exist_okTrue) app.post(/upload_doc) async def upload_document(file: UploadFile File(...)): 上传PDF文档并添加到RAG知识库 if not file.filename.endswith(.pdf): raise HTTPException(status_code400, detail仅支持PDF文件) file_path os.path.join(UPLOAD_DIR, file.filename) with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) # 这里需要调用RAGService的ingest_document方法 # 为了简化我们假设有一个全局的rag_service实例 from app.services.rag_service import RAGService rag_service RAGService() rag_service.ingest_document(file_path) return {message: f文档 {file.filename} 上传并处理成功, file_path: file_path} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)7. 运行与测试7.1 启动服务确保已安装所有依赖pip install -r requirements.txt设置环境变量创建.env文件DASHSCOPE_API_KEYyour_dashscope_key_here # 如果用通义千问API LLM_API_BASEhttp://localhost:8000/v1 # 本地Qwen服务地址 LLM_API_KEYnot-needed启动FastAPI服务cd financial_agent_harness uvicorn app.main:app --reload --host 0.0.0.0 --port 80007.2 测试API使用curl或httpie测试# 1. 上传文档构建知识库 curl -X POST http://localhost:8000/upload_doc \ -H accept: application/json \ -H Content-Type: multipart/form-data \ -F file/path/to/your/annual_report.pdf # 2. 发送查询请求 curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 腾讯控股00700.HK最近的股价是多少} # 返回示例 # {session_id:..., answer:, status:processing, request_id:req_123} # 3. 使用返回的request_id获取结果 curl -X GET http://localhost:8000/result/req_1238. 常见问题与排查思路在开发和部署过程中你可能会遇到以下问题问题现象可能原因排查思路与解决方案启动服务时报ImportError依赖未安装或版本冲突1. 检查requirements.txt是否正确。2. 使用pip list确认关键包langchain, chromadb, fastapi已安装。3. 创建新的虚拟环境重新安装。调用/query接口长时间无结果Agent处理超时或LLM服务未响应1. 检查后台日志看process_query_task是否报错。2. 确认LLM服务本地Qwen或云端API可访问且正常。3. 在AgentHarness.run方法中添加超时机制。RAG检索返回无关内容文档切分不合理或嵌入模型不匹配1. 调整RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。2. 尝试不同的嵌入模型如text-embedding-v2。3. 在检索后增加一个“重排序”步骤提升精度。Agent识别意图错误规划阶段的意图识别逻辑过于简单1. 引入更专业的意图识别模型如微调的小模型。2. 使用LLM本身进行意图判断增加一个规划LLM调用。3. 完善正则规则覆盖更多用例。工具调用失败如股票API网络问题、API变更、权限错误1. 在工具函数内增加更详细的错误日志和重试逻辑。2. 使用Mock数据或备用数据源作为降级方案。3. 在Harness的_fallback_strategy中提供友好提示。多轮对话上下文丢失未将会话ID与历史记录关联1. 在AgentExecutionContext中持久化历史对话。2. 使用向量数据库或缓存如Redis存储和检索对话历史。3. 在每次请求时将历史对话作为上下文注入LLM。9. 最佳实践与进阶优化建议基于Harness工程思想这个项目还可以从以下维度进行深化和优化9.1 可观测性与监控结构化日志为AgentStep设计更丰富的日志格式便于接入ELK或Loki。指标收集记录每个步骤的耗时、LLM的Token使用量、工具调用成功率等使用Prometheus暴露指标。链路追踪集成OpenTelemetry为每次用户请求生成唯一的Trace ID贯穿所有Harness步骤和工具调用。9.2 性能与稳定性异步化本文已使用async/await确保所有I/O操作LLM调用、工具API、向量检索都是异步的避免阻塞。缓存策略对频繁且结果稳定的查询如某些股票价格、常见金融术语解释进行缓存减少LLM调用和工具调用。限流与熔断在FastAPI层或Harness入口添加限流如slowapi防止滥用。对脆弱的工具调用如外部API实现熔断机制如circuitbreaker。9.3 智能体能力增强复杂任务分解当前Agent的plan阶段较简单。可引入更强大的规划器如LLM本身通过Chain of Thought提示或专门的规划模型将复杂问题分解为子任务图。动态工具管理实现一个工具注册中心Agent可以在运行时发现并选择最合适的工具而不是硬编码。记忆与反思为Agent添加长期记忆使其能从过去的错误中学习反思。可以在AgentStep中记录成功/失败经验并在后续类似任务中作为提示。9.4 生产环境部署配置管理将所有配置模型参数、API密钥、超时时间外置到配置中心如Apollo或环境变量避免硬编码。健康检查为FastAPI服务添加/health端点检查LLM连接、向量数据库状态、工具API连通性。版本管理与回滚对Agent的提示词、工具集、Harness逻辑进行版本控制。支持快速回滚到稳定版本。9.5 安全与合规输入输出过滤在Harness层之前添加一个安全过滤层对用户输入进行敏感词过滤、恶意指令检测Prompt Injection。对模型输出进行内容安全审核。权限控制不同的用户或API密钥可能对应不同的工具使用权限如只有VIP用户能调用实时股票API。在Harness的plan或execute阶段加入权限校验。数据隐私确保上传的PDF文档在向量化过程中敏感信息如个人身份证号、电话号码被适当脱敏。通过以上步骤我们不仅实现了一个功能性的金融问答机器人更实践了一套名为Harness工程的、可扩展、可观测、易维护的AI智能体构建方法。这套方法的核心在于通过标准化的状态管理和流程编排将非确定性的LLM能力转化为可靠的应用服务。你可以将此项目作为模板替换其中的工具、领域知识和LLM快速构建出适用于客服、编程助手、数据分析等不同场景的智能体。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
