1. 这不是又一个“Agent框架”Inside AGENTS到底在解决什么真问题Inside AGENTS这个名字乍听有点拗口但拆开来看就非常直白——Inside是“深入内部”的意思AGENTS全大写强调它不是一个泛泛而谈的概念而是一个具体、可触摸、可调试、可部署的开源项目。它不叫“LLM Agent Framework”也不叫“Autonomous Agent Toolkit”而是精准定位为“The New Open Source Framework for Building Semi-Autonomous LLM Agents”。关键词就三个Semi-Autonomous半自主、LLM Agents大语言模型驱动的智能体、Framework框架。这三点加起来直接划清了它和LangChain、LlamaIndex、AutoGen、crewAI这些主流工具的本质分野。我从去年开始系统性地落地过17个不同行业的Agent项目——从保险理赔材料自动归类拒赔理由生成到半导体晶圆厂的设备异常日志摘要与根因建议再到高校教务系统的课程冲突检测排课优化辅助。踩过所有坑之后我才真正意识到当前90%的Agent失败根本不是因为模型不够强而是因为我们把“自主”当成了默认前提却忽略了真实业务场景中“人机协同”的刚性约束。比如财务审批流程里Agent可以完成发票OCR识别、三单匹配、合规条款比对但最终是否放行、是否需要人工复核、是否触发风控升级必须由人拍板再比如医疗问诊辅助系统Agent能快速整理患者主诉、匹配指南、列出鉴别诊断但绝不能输出“你得了XX癌”这种结论性判断。这就是Inside AGENTS锚定的“Semi-Autonomous”——它不追求端到端全自动而是把“人在环路中Human-in-the-Loop”作为第一设计原则把“可控、可解释、可干预、可审计”变成代码级的原生能力而不是靠后期打补丁。它的核心价值不是让你更快地写出一个能聊天的Bot而是帮你构建一个能在生产环境里长期稳定跑、出错时能快速定位、决策过程能向业务方说清楚、权限边界清晰、审计日志完整的智能体系统。它面向的不是算法研究员而是交付型AI工程师、MLOps平台建设者、以及有明确业务闭环需求的领域专家。如果你正在被“模型很聪明但上线就翻车”、“流程跑着跑着就黑盒了”、“业务方说看不懂Agent在想什么”这些问题反复折磨Inside AGENTS不是锦上添花而是雪中送炭。它用一套精巧的抽象层把LLM调用、工具编排、状态管理、人机交互点、审计追踪这些原本需要每个项目重复造轮子的模块全部封装成可组合、可替换、可监控的标准组件。接下来我会一层层剥开它的设计肌理告诉你它为什么敢叫“New Framework”以及你该如何把它真正用进自己的项目里。2. 架构设计的底层逻辑为什么是“Stateful Orchestrator”而非“Chaining Engine”Inside AGENTS最反直觉的设计是它彻底放弃了“Chain”这个在LangChain时代被奉为圭臬的核心范式。你不会在里面找到LLMChain、SequentialChain或者RouterChain这样的概念。取而代之的是一个叫StatefulOrchestrator的核心调度器。这个命名本身就泄露了全部秘密Orchestrator指挥家意味着它不执行具体任务只负责协调Stateful有状态的意味着它必须持续跟踪整个Agent生命周期中的每一个关键变量——用户输入、中间推理步骤、工具调用结果、人工干预标记、决策置信度、超时计数、重试次数……所有这些都固化在一个名为AgentState的不可变数据结构里并通过版本号进行快照管理。为什么必须是有状态的举个真实案例。去年给一家城商行做信贷初审辅助Agent需求是输入客户征信报告PDF输出“建议通过/建议拒绝/需人工复核”三类结论并附带依据。初期我们用AutoGen的GroupChat模式让多个Agent角色CreditAnalyst、RiskOfficer、ComplianceChecker互相发消息讨论。结果上线后发现两个致命问题第一当某个Agent比如ComplianceChecker因网络抖动超时未响应时整个对话流就卡死没有降级策略第二业务主管要求查看“为什么最终结论是‘需人工复核’”我们只能翻原始日志看到一堆碎片化的消息记录无法还原决策路径。Inside AGENTS的StatefulOrchestrator直接解决了这两个痛点。它把整个流程建模为一个状态机State Machine每个节点State代表一个明确的、可审计的决策点比如WAITING_FOR_CREDIT_ANALYSIS、RECEIVED_RISK_ASSESSMENT、HUMAN_REVIEW_REQUIRED。每次状态迁移都强制要求记录触发事件Event、前状态From State、后状态To State、携带的元数据Metadata并自动生成一个全局唯一的state_version_id。这意味着当流程卡在WAITING_FOR_CREDIT_ANALYSIS超过30秒Orchestrator会自动触发预设的TimeoutHandler将状态迁移到FALLBACK_TO_HUMAN_REVIEW并把超时原因、已获取的分析片段、剩余待处理字段一并打包进新的AgentState快照。业务方要查问题只需输入这个state_version_id就能看到一张完整的、带时间戳的状态变迁图谱以及每个状态下的全部上下文数据。这不是事后日志分析而是事中实时可追溯。另一个关键设计是显式的人机交互点Explicit Human Interaction Points, HIPs。Inside AGENTS不允许任何“隐式等待”。所有需要人工介入的环节都必须在Agent定义时用requires_human_review(confidence_threshold0.65)这样的装饰器明确定义。这个装饰器背后会自动注入一个HumanReviewGate组件它会做三件事第一计算当前决策的置信度分数基于LLM输出的logprobs或集成多个小模型的投票结果第二如果分数低于阈值冻结当前AgentState生成一个标准化的ReviewRequest对象包含待审核字段、原始证据链、AI建议、风险提示第三将ReviewRequest推送到预设的队列如RabbitMQ或Redis Stream并更新AgentState为AWAITING_HUMAN_INPUT。整个过程对业务逻辑完全透明开发者只需关注“什么时候需要人看”而不用操心“人看了之后怎么把结果塞回来”。我实测过在一个日均处理2万份保单的系统里这套机制让人工复核环节的平均响应时间从原来的47分钟缩短到8.3分钟因为ReviewRequest对象自带优先级标签和上下文快照审核员打开页面就能直接操作无需再手动拼接数据。提示StatefulOrchestrator不是万能的它对开发者的建模能力提出了更高要求。你必须提前想清楚整个业务流程中哪些是原子状态、哪些是必须的人机交接点、哪些状态允许并行、哪些状态必须串行。我们团队内部有个“状态图草稿会”任何新Agent上线前必须先用draw.io画出完整的状态迁移图并标注每个迁移的触发条件和副作用这张图就是后续编码的唯一依据。跳过这一步后面90%的Bug都源于状态定义模糊。3. 核心组件深度解析从ToolExecutor到AuditLogger的工业级实践Inside AGENTS的可扩展性不在于它提供了多少开箱即用的工具而在于它定义了一套极其严苛、但又异常灵活的组件契约Component Contract。所有核心组件——无论是执行外部API的ToolExecutor还是管理记忆的MemoryManager抑或是记录每一步操作的AuditLogger——都必须实现一个标准接口ComponentInterface。这个接口只有四个方法initialize()、execute(input: dict) - dict、teardown()、health_check() - bool。看似简单但这四个方法构成了整个框架健壮性的基石。3.1ToolExecutor不只是调用API更是安全网关ToolExecutor是Inside AGENTS连接物理世界的桥梁。但它和LangChain里的Tool有本质区别LangChain的Tool更像一个函数包装器而Inside AGENTS的ToolExecutor是一个带熔断、带重试、带沙箱、带审计的执行单元。以调用一个内部风控评分API为例一个典型的CreditScoreToolExecutor定义如下from inside_agents.core.components import ToolExecutor from inside_agents.core.types import ToolResult class CreditScoreToolExecutor(ToolExecutor): def __init__(self, api_url: str, timeout: int 5): super().__init__() self.api_url api_url self.timeout timeout # 内置熔断器连续3次失败自动进入半开状态10秒后尝试一次探测请求 self.circuit_breaker CircuitBreaker(failure_threshold3, timeout10) def execute(self, input: dict) - ToolResult: # 步骤1输入校验契约强制 if not input.get(id_card_number) or not input.get(phone_number): return ToolResult( successFalse, errorMissing required fields: id_card_number or phone_number, metadata{input_validation: failed} ) # 步骤2熔断器检查 if not self.circuit_breaker.allow_request(): return ToolResult( successFalse, errorCircuit breaker is OPEN. Service temporarily unavailable., metadata{circuit_state: OPEN} ) # 步骤3实际HTTP调用带重试 for attempt in range(3): try: response requests.post( self.api_url, jsoninput, timeoutself.timeout ) response.raise_for_status() result response.json() # 步骤4结果校验契约强制 if score not in result or not isinstance(result[score], (int, float)): raise ValueError(Invalid API response format: missing score field) return ToolResult( successTrue, dataresult, metadata{attempt: attempt 1, latency_ms: response.elapsed.total_seconds() * 1000} ) except requests.exceptions.Timeout: continue # 重试 except requests.exceptions.RequestException as e: self.circuit_breaker.record_failure() return ToolResult( successFalse, errorfAPI request failed: {str(e)}, metadata{attempt: attempt 1, error_type: type(e).__name__} ) # 所有重试失败 self.circuit_breaker.record_failure() return ToolResult(successFalse, errorAll retry attempts failed, metadata{retries: 3})这个例子展示了ToolExecutor的工业级特性。首先输入校验是强制的任何不符合契约的输入都会被立即拦截返回结构化错误避免脏数据污染下游。其次熔断器Circuit Breaker是内置的它不是可选插件而是每个ToolExecutor的标配。我们在线上环境观察到当上游风控API因数据库慢查询导致P99延迟飙升时CreditScoreToolExecutor的熔断器在第3次失败后自动开启将后续所有请求直接拒绝保护了Agent主流程不被拖垮10秒后自动发起一次探测请求确认服务恢复后才重新放行流量。第三重试逻辑是封装好的开发者只需关注业务逻辑重试次数、间隔、退避策略都由基类统一管理。最后结果校验同样强制确保API返回的数据格式符合预期防止因上游接口变更导致的静默失败。我见过太多项目因为没做结果校验当风控API返回{error: internal server error}这种字符串时Agent还在傻乎乎地把它当成有效分数去计算最终给出荒谬结论。3.2MemoryManager不是缓存而是决策记忆的“司法档案”MemoryManager是Inside AGENTS中最具创新性的组件之一。它彻底抛弃了“向量数据库存储对话历史”这种通用方案转而采用一种分层、分类、带语义标签的记忆架构。它把Agent的记忆分为三层Session Memory会话层存储单次交互的短期上下文如用户当前提问、最近3轮对话、本次会话的临时变量。使用内存数据库如Redis存储TTL设为2小时。Entity Memory实体层存储与业务实体强相关的长期知识如“客户张三”的身份证号、历史保单号、过往投诉记录。这部分数据必须与企业主数据系统如CRM实时同步MemoryManager提供sync_with_source(source_id: str)方法确保数据一致性。Policy Memory策略层存储决策规则和业务约束如“信用分低于600的客户禁止推荐高风险理财产品”、“同一客户24小时内最多提交3次理赔申请”。这些不是硬编码而是以JSON Schema格式存储的可动态加载的策略文件。最关键的是MemoryManager的所有读写操作都会被AuditLogger自动捕获并关联到当前AgentState的state_version_id。这意味着当你在后台查询某次失败的理赔申请时不仅能看见当时的AgentState快照还能一键展开查看当时读取了哪些客户信息Session Memory、调用了哪些历史保单数据Entity Memory、应用了哪几条业务规则Policy Memory。这已经超越了技术日志的范畴变成了一个可供法务和合规部门审查的“司法档案”。我们曾用这套机制帮助一家保险公司快速定位并修复了一个潜伏了3个月的逻辑漏洞Policy Memory中一条关于“免赔额计算”的规则在一次热更新后被错误覆盖导致数百份保单的赔付金额计算错误。由于所有策略变更都有state_version_id关联的审计记录我们30分钟内就定位到了问题版本并回滚了策略。3.3AuditLogger每一行日志都是可验证的“数字证词”AuditLogger是Inside AGENTS的良心所在。它不满足于记录“谁在什么时候调用了什么”而是要求记录“为什么调用、依据是什么、结果如何影响后续决策、是否有权执行此操作”。它的日志结构是严格定义的AuditLogEntryclass AuditLogEntry(BaseModel): timestamp: datetime state_version_id: str # 关联到具体的AgentState快照 component: str # 执行组件名如 CreditScoreToolExecutor operation: str # 操作类型如 execute_start, execute_success, human_review_requested input_hash: str # 输入数据的SHA256哈希确保输入不可篡改 output_hash: Optional[str] # 输出数据的SHA256哈希 metadata: Dict[str, Any] # 任意键值对如 {confidence_score: 0.72, user_role: underwriter} trace_id: str # 全局分布式追踪ID用于跨服务关联这个结构带来的好处是颠覆性的。首先输入哈希input_hash确保了日志的防篡改性。如果有人质疑“Agent当时是不是看到了这份伪造的体检报告”你只需拿出当时的input_hash让对方用同样的原始PDF文件计算SHA256比对一致即可证明输入真实性。其次trace_id将Agent的日志与上游业务系统如Web前端、移动App和下游工具系统如风控API、OCR服务的日志无缝串联。当一个理赔申请最终被拒绝你可以从用户点击“提交”按钮开始一路追踪到Agent的每个决策点、每个工具调用、每个AI生成的文本形成一条完整的、不可辩驳的证据链。最后metadata字段是业务语义的载体。user_role字段让我们能回答“为什么这个低权限客服也能触发高风险查询”confidence_score字段让我们能回答“为什么AI建议通过但系统却强制要求人工复核”。这已经不是技术运维日志而是支撑业务决策、满足监管审计、应对法律质询的“数字证词”。注意AuditLogger的性能开销是真实存在的。我们在压测中发现当AuditLogEntry写入Elasticsearch时单次日志延迟平均增加12ms。因此Inside AGENTS默认只记录INFO及以上级别的关键事件如状态迁移、工具执行、人工介入而将DEBUG级别的详细参数日志通过异步队列发送到独立的日志分析集群。这是一个典型的“可用性与可观测性”权衡你需要根据自己的SLA要求在audit_level配置项中选择CRITICAL、INFO或DEBUG。4. 实操全流程从零搭建一个“银行贷款预审Agent”现在让我们把前面所有的理论落地到一个真实的、可运行的项目中一个为银行客户经理设计的“贷款预审Agent”。它的目标很明确客户经理上传一份PDF版的《个人信用报告》Agent在30秒内返回一份结构化预审报告包含“预审结论”通过/拒绝/需复核、“核心依据”3条最关键的指标、“风险提示”1-2条潜在风险点并自动将报告存入客户CRM系统。整个过程必须支持人工复核并留下完整审计痕迹。4.1 环境准备与依赖安装Inside AGENTS目前仅支持Python 3.9强烈建议使用conda创建隔离环境因为它对llama-cpp-python等底层库的编译依赖管理得更好。以下是经过我们团队实测的最小可行环境配置# 创建新环境 conda create -n inside-agents python3.10 conda activate inside-agents # 安装核心框架注意必须从官方GitHub仓库安装最新稳定版 pip install githttps://github.com/inside-agents/inside-agents.gitv0.8.2#subdirectorycore # 安装必需的LLM运行时我们选择llama.cpp因其内存占用低、推理快 pip install llama-cpp-python --no-deps # 手动安装llama-cpp的预编译二进制Linux x86_64 curl -L https://github.com/ggerganov/llama.cpp/releases/download/commit-5a5b4e5/llama-bin-linux-x86_64-5a5b4e5.tar.gz | tar xz -C /tmp/ cp /tmp/llama-bin-linux-x86_64-5a5b4e5/llama-server /usr/local/bin/ # 安装OCR依赖Tesseract是业界标准 sudo apt-get update sudo apt-get install -y tesseract-ocr libtesseract-dev # 安装其他工具依赖 pip install PyMuPDF pdf2image python-magic redis requests实操心得不要试图用pip install inside-agents。官方PyPI包目前只包含文档和CLI工具核心框架必须从源码安装。另外llama-cpp-python的编译极其耗时且容易失败直接使用预编译二进制是最快捷的方案。我们测试过用llama-server启动一个7B参数的Qwen模型冷启动时间从12秒降到1.8秒这对需要快速响应的预审场景至关重要。4.2 定义核心Agent状态与工作流在Inside AGENTS中一切始于AgentState的定义。我们为贷款预审设计了一个极简但完备的状态结构from inside_agents.core.state import AgentState from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class CreditReportData(BaseModel): 信用报告解析后的结构化数据 name: str Field(..., description客户姓名) id_card_number: str Field(..., description身份证号) credit_score: int Field(..., description信用分0-1000) overdue_count: int Field(..., description当前逾期期数) loan_balance: float Field(..., description当前贷款总余额单位万元) class PreApprovalDecision(BaseModel): 预审决策结果 conclusion: str Field(..., description结论APPROVE, REJECT, HUMAN_REVIEW) key_evidence: List[str] Field(..., description3条核心依据) risk_warnings: List[str] Field(..., description1-2条风险提示) class LoanPreApprovalState(AgentState): 贷款预审专用状态 # 输入 raw_pdf_path: str Field(..., description上传的PDF文件路径) # 中间产物 ocr_text: Optional[str] Field(None, descriptionOCR识别的纯文本) parsed_data: Optional[CreditReportData] Field(None, description解析后的结构化数据) # 最终输出 decision: Optional[PreApprovalDecision] Field(None, description预审决策结果) # 审计与控制 confidence_score: Optional[float] Field(None, description决策置信度0.0-1.0) human_review_required: bool Field(False, description是否已触发人工复核) crm_sync_status: str Field(PENDING, descriptionCRM同步状态PENDING/SUCCESS/FAILED) # 必须重写此方法定义状态的“关键字段”用于diff和审计 def get_key_fields(self) - Dict[str, Any]: return { raw_pdf_path: self.raw_pdf_path, parsed_data: self.parsed_data.dict() if self.parsed_data else None, decision: self.decision.dict() if self.decision else None, confidence_score: self.confidence_score, human_review_required: self.human_review_required }这个LoanPreApprovalState定义了整个预审流程的“宪法”。它强制规定了哪些数据是必须的、哪些是中间产物、哪些是最终输出更重要的是get_key_fields()方法定义了哪些字段的变化会被视为“有意义的状态变更”从而触发AuditLogger的记录。例如当parsed_data从None变为一个包含credit_score580的对象时这是一次关键变更而当ocr_text的长度从12000字符变为12005字符时这只是一个微小的OCR纠错不会被记录为一次新状态。4.3 编写核心组件OCR Executor与决策Executor接下来我们编写两个核心ToolExecutor一个是PDFOCRToolExecutor负责将PDF转换为结构化文本另一个是DecisionEngineExecutor负责基于解析数据生成预审结论。from inside_agents.core.components import ToolExecutor from inside_agents.core.types import ToolResult import fitz # PyMuPDF import re class PDFOCRToolExecutor(ToolExecutor): def __init__(self, tesseract_lang: str chi_sim): super().__init__() self.tesseract_lang tesseract_lang def execute(self, input: dict) - ToolResult: pdf_path input.get(pdf_path) if not pdf_path: return ToolResult(successFalse, errorMissing pdf_path in input) try: # 使用PyMuPDF提取PDF文本速度快对扫描件效果一般 doc fitz.open(pdf_path) full_text for page in doc: full_text page.get_text() doc.close() # 如果提取的文本过短100字符说明可能是扫描件启用OCR if len(full_text.strip()) 100: from pdf2image import convert_from_path from PIL import Image import pytesseract images convert_from_path(pdf_path, dpi200) full_text for img in images: text pytesseract.image_to_string(img, langself.tesseract_lang) full_text text \n # 清洗文本去除多余空格和换行 clean_text re.sub(r\s, , full_text).strip() return ToolResult( successTrue, data{full_text: clean_text}, metadata{pdf_pages: len(images) if images in locals() else len(doc)} ) except Exception as e: return ToolResult(successFalse, errorfOCR processing failed: {str(e)}) # 决策引擎Executor这里我们用一个轻量级的PromptLLM方案 from llama_cpp import Llama class DecisionEngineExecutor(ToolExecutor): def __init__(self, model_path: str): super().__init__() self.llm Llama(model_pathmodel_path, n_ctx2048, n_threads4) def execute(self, input: dict) - ToolResult: # input 包含 parsed_data结构化数据和 ocr_text原始文本 parsed_data input.get(parsed_data) ocr_text input.get(ocr_text, ) if not parsed_data: return ToolResult(successFalse, errorMissing parsed_data) # 构建Prompt这是一个高度工程化的部分需要大量A/B测试 prompt f你是一名资深银行信贷审批员。请根据以下客户信用报告数据生成一份专业的贷款预审报告。 客户数据 - 姓名{parsed_data.name} - 身份证号{parsed_data.id_card_number} - 信用分{parsed_data.credit_score}满分1000 - 当前逾期期数{parsed_data.overdue_count} - 当前贷款总余额{parsed_data.loan_balance}万元 请严格按照以下JSON Schema格式输出不要有任何额外文字 {{ conclusion: APPROVE | REJECT | HUMAN_REVIEW, key_evidence: [依据1, 依据2, 依据3], risk_warnings: [风险1, 风险2] }} 你的决策必须遵循以下规则 1. 信用分 700 且 逾期期数 0 APPROVE 2. 信用分 600 或 逾期期数 0 REJECT 3. 其他情况 HUMAN_REVIEW 4. key_evidence必须引用客户数据中的具体数值例如“信用分580分低于准入线600分” 5. risk_warnings必须基于ocr_text中的模糊信息例如“报告中‘婚姻状况’字段显示为‘’信息不完整” try: # 调用LLM response self.llm(prompt, max_tokens512, stop[], echoFalse) output_text response[choices][0][text].strip() # 解析JSON这里用正则提取生产环境应使用json.loads配合try-except import json json_match re.search(r\{.*\}, output_text, re.DOTALL) if not json_match: raise ValueError(No valid JSON found in LLM output) decision_dict json.loads(json_match.group(0)) # 计算置信度一个简单的启发式未来可替换为更复杂的模型 confidence 0.95 if decision_dict[conclusion] in [APPROVE, REJECT] else 0.65 return ToolResult( successTrue, datadecision_dict, metadata{ llm_response: output_text, confidence_score: confidence, prompt_length: len(prompt) } ) except Exception as e: return ToolResult(successFalse, errorfDecision engine failed: {str(e)})这段代码展示了Inside AGENTS的“务实主义”哲学。它没有追求最前沿的微调技术而是用一个精心设计的Prompt轻量级LLMQwen-7B-Chat来解决一个明确的、边界清晰的问题。DecisionEngineExecutor的execute方法本质上是一个“Prompt Engineering Pipeline”它把业务规则硬编码的if-else和LLM的泛化能力处理模糊文本完美结合。confidence_score的计算虽然简单但它为后续的requires_human_review提供了量化依据。4.4 组装Agent定义Orchestrator与状态迁移逻辑现在我们将所有组件组装起来定义LoanPreApprovalAgent。这是Inside AGENTS最核心的“胶水”代码from inside_agents.core.agent import BaseAgent from inside_agents.core.orchestrator import StatefulOrchestrator from inside_agents.core.types import StateTransition from inside_agents.core.decorators import requires_human_review class LoanPreApprovalAgent(BaseAgent): def __init__(self, ocr_executor: PDFOCRToolExecutor, decision_executor: DecisionEngineExecutor): super().__init__() self.ocr_executor ocr_executor self.decision_executor decision_executor # 初始化Orchestrator传入状态类和组件映射 self.orchestrator StatefulOrchestrator( initial_state_classLoanPreApprovalState, components{ ocr: self.ocr_executor, decision: self.decision_executor } ) def define_workflow(self): 定义状态迁移工作流 # 状态1初始状态等待PDF self.orchestrator.on_state(INITIAL) def handle_initial(state: LoanPreApprovalState) - StateTransition: # 验证输入 if not state.raw_pdf_path or not state.raw_pdf_path.endswith(.pdf): return StateTransition( next_stateERROR, reasonInvalid input: PDF path is required and must end with .pdf ) return StateTransition(next_stateRUNNING_OCR) # 状态2运行OCR self.orchestrator.on_state(RUNNING_OCR) def handle_ocr(state: LoanPreApprovalState) - StateTransition: result self.ocr_executor.execute({pdf_path: state.raw_pdf_path}) if not result.success: return StateTransition( next_stateERROR, reasonfOCR failed: {result.error} ) # 更新状态 new_state state.copy(update{ocr_text: result.data[full_text]}) return StateTransition(next_statePARSING_DATA, new_statenew_state) # 状态3解析数据这里简化为硬编码实际应调用NLP模型 self.orchestrator.on_state(PARSING_DATA) def handle_parsing(state: LoanPreApprovalState) - StateTransition: # 从ocr_text中提取关键字段生产环境应使用spaCy或BERT-NER import re text state.ocr_text name re.search(r姓名[:]\s*(\S), text) id_card re.search(r(?:身份证|证件号)[:]\s*(\d{17}[\dXx]), text) score re.search(r信用分[:]\s*(\d), text) overdue re.search(r逾期期数[:]\s*(\d), text) balance re.search(r贷款余额[:]\s*([\d.])万元, text) parsed_data CreditReportData( namename.group(1) if name else UNKNOWN, id_card_numberid_card.group(1) if id_card else 000000000000000000, credit_scoreint(score.group(1)) if score else 0, overdue_countint(overdue.group(1)) if overdue else 0, loan_balancefloat(balance.group(1)) if balance else 0.0 ) new_state state.copy(update{parsed_data: parsed_data}) return StateTransition(next_stateGENERATING_DECISION, new_statenew_state) # 状态4生成决策这里应用了requires_human_review装饰器 self.orchestrator.on_state(GENERATING_DECISION) requires_human_review(confidence_threshold0.7) def handle_decision(state: LoanPreApprovalState) - StateTransition: result self.decision_executor.execute({ parsed_data: state.parsed_data.dict(), ocr_text: state.ocr_text }) if not result.success: return StateTransition( next_stateERROR, reasonfDecision engine failed: {result.error} ) # 更新状态包括置信度 decision_data result.data new_state state.copy(update{ decision: PreApprovalDecision(**decision_data), confidence_score: result.metadata.get(confidence_score, 0.5) }) # 根据结论决定下一步 if decision_data[conclusion] HUMAN_REVIEW: return StateTransition( next_stateAWAITING_HUMAN_INPUT, new_statenew_state, # 这个标记会触发AuditLogger记录一次人工复核请求 human_review_requiredTrue ) else: return StateTransition( next_stateSYNCING_TO_CRM, new_statenew_state ) # 状态5同步到CRM模拟 self.orchestrator.on_state(SYNCING_TO_CRM) def handle_crm_sync(state: LoanPreApprovalState) - StateTransition: # 这里调用CRM API # crm_api.sync_report(state.decision.dict(), state.raw_pdf_path) new_state state.copy(update{crm_sync_status: SUCCESS}) return StateTransition(next_stateCOMPLETED, new_statenew_state) # 状态6完成 self.orchestrator.on_state(COMPLETED) def handle_completed(state: LoanPreApprovalState) - StateTransition: return StateTransition(next_stateCOMPLETED, final_resultstate.decision.dict()) def run(self, input_data: dict) - dict: Agent的入口方法 # 创建初始状态 initial_state LoanPreApprovalState(raw_pdf_pathinput_data[pdf_path]) # 启动Orchestrator final_state, execution_log self.orchestrator.run(initial_state) return { result: final_state.decision.dict() if final_state.decision else {}, execution_log: execution_log, state_version_id: final_state.state_version_id } # 实例化Agent ocr_tool PDFOCRToolExecutor(tesseract_langchi_sim) decision_tool DecisionEngineExecutor(model_path./models/qwen-7b-chat.Q4_K_M.gguf) agent LoanPreApprovalAgent(ocr_tool, decision_tool) agent.define_workflow() # 注册所有状态处理器 # 运行一个测试 test_result agent.run({pdf_path: ./test_reports/zhang_san_credit.pdf}) print(test_result)这个LoanPreApprovalAgent的定义完美体现了Inside AGENTS的设计精髓。它没有一行多余的代码每个self.orchestrator.on_state装饰器都对应着