在AI技术从“大模型狂热”转向“智能体（Agent）落地”的2025年，开发者需要掌握的不仅是模型训练技术，更是如何让模型“自主行动”的系统化能力。本文结合九天菜菜团队最新课程《大模型与Agent高阶开发指南》中的核心方法论，通过一个完整的金融分析Agent案例，展示从任务拆解到多Agent协作的全流程开发实践。

一、Agent开发的核心技术栈

1.1 模型层：QLoRA微调与GRPO强化学习

传统大模型存在知识截止和操作能力缺失两大局限。九天菜菜课程通过QLoRA量化压缩技术将Qwen2.5等模型的显存占用降低80%，同时保持95%以上性能。例如在金融分析场景中，使用QLoRA微调后的Qwen2.5-32B模型可精准解析财报中的关键指标：

python

from transformers import AutoModelForCausalLM, AutoTokenizer

import torch

# QLoRA微调后的模型加载（显存占用仅需12GB）

model = AutoModelForCausalLM.from_pretrained(

"Qwen2.5-32B-QLoRA",

torch_dtype=torch.float16,

device_map="auto"

)

tokenizer = AutoTokenizer.from_pretrained("Qwen2.5-32B")

# 财报解析示例

prompt = """

从以下财报中提取营收增长率、毛利率、净利润率：

"2024年Q2财报显示，营收58.2亿元，同比增长12.3%；

营业成本42.1亿元；净利润8.7亿元。"

"""

inputs = tokenizer(prompt, return_tensors="pt").to("cuda")

outputs = model.generate(**inputs, max_length=200)

print(tokenizer.decode(outputs[0], skip_special_tokens=True))

# 输出：营收增长率12.3%，毛利率27.7%，净利润率14.9%

1.2 工具层：Function Calling与API集成

通过OpenAI兼容的Function Calling机制，Agent可调用外部API完成实时数据获取。以下代码展示如何构建一个股票查询工具：

python

import requests

from typing import TypedDict

class StockData(TypedDict):

code: str

price: float

change: float

def get_stock_data(code: str) -> StockData:

"""调用金融API获取实时行情"""

response = requests.get(f"https://api.finance.com/stock/{code}")

data = response.json()

return {

"code": code,

"price": data["price"],

"change": data["change_percent"]

}

# 工具描述（供LLM识别调用条件）

TOOL_DESCRIPTION = """

{

"name": "stock_query",

"description": "查询A股实时行情，需提供股票代码（如600519.SH）",

"parameters": {

"type": "object",

"properties": {

"code": {"type": "string", "description": "股票代码"}

},

"required": ["code"]

}

}

"""

二、单Agent任务执行：ReAct推理框架实战

以“生成某科技股技术分析报告”为例，展示ReAct框架如何实现“思考-行动-观察”闭环：

python

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.output_parsers import StructuredOutputParser

# 1. 构建ReAct提示词模板

prompt = ChatPromptTemplate.from_messages([

("system", """你是一个金融分析师Agent，使用ReAct框架完成任务。

步骤：

1. 思考：分析当前步骤需要哪些数据

2. 行动：调用stock_query工具或执行Python代码

3. 观察：检查结果是否满足需求

4. 循环直到任务完成"""),

("user", "{input}")

])

# 2. 定义行动解析器

output_parser = StructuredOutputParser.from_yaml("""

thoughts:

description: 思考过程

action:

description: 要执行的动作

subfields:

name:

description: 工具名称

args:

description: 工具参数

""")

# 3. 执行流程

def react_agent(task):

history = []

while True:

# 构建当前上下文

context = "\n".join([f"History: {h}" for h in history[-3:]]) + f"\nCurrent Task: {task}"

# LLM生成下一步行动

llm_output = model.chat(prompt.format_message(prompt, input=context))

parsed = output_parser.parse(llm_output.content)

# 执行行动

if parsed["action"]["name"] == "stock_query":

result = get_stock_data(**parsed["action"]["args"])

observation = f"获取到数据：{result}"

elif parsed["action"]["name"] == "calculate":

# 执行Python计算示例

exec(f"result = {parsed['action']['args']['expression']}")

observation = f"计算结果：{result}"

else:

observation = "未知操作"

history.append(f"{parsed['thoughts']}\nAction: {parsed['action']}\nObservation: {observation}")

# 检查任务是否完成

if "报告生成完成" in parsed["thoughts"]:

break

return "\n".join(history)

# 示例调用

react_agent("生成600519.SH的MACD技术分析报告")

三、多Agent协作：Supervisor架构实现智能BI

在复杂业务场景中，单一Agent难以胜任。九天菜菜课程提出的Supervisor架构通过主从Agent协作实现任务分解与资源调度：

python

from langgraph.graph import StateGraph

from langgraph.graph.state import State

# 1. 定义Agent状态

class BIState(State):

query: str

step: str = "init"

data: dict = {}

report: str = ""

# 2. 构建多Agent工作流

def build_bi_workflow():

workflow = StateGraph(BIState)

# 主Agent（任务分解）

def supervisor_agent(state):

if state.step == "init":

return {

"step": "data_query",

"subtasks": [

{"type": "sql", "query": f"SELECT * FROM stocks WHERE code='{state.query}'"},

{"type": "tech_analysis", "days": 30}

]

}

# SQL查询Agent

def sql_agent(state):

# 实际中连接数据库执行查询

return {"step": "tech_analysis", "raw_data": {...}}

# 技术分析Agent

def tech_agent(state):

# 计算MACD、RSI等指标

return {"step": "report_gen", "analysis": {...}}

# 构建节点

workflow.add_node("supervisor", supervisor_agent)

workflow.add_node("sql", sql_agent)

workflow.add_node("tech", tech_agent)

# 定义流程

workflow.set_entry_point("supervisor")

workflow.add_edge("supervisor", "sql", conditions=lambda s: s.step == "data_query")

workflow.add_edge("sql", "tech", conditions=lambda s: s.step == "tech_analysis")

workflow.add_edge("tech", "supervisor", conditions=lambda s: s.step == "report_gen")

return workflow.compile()

# 3. 执行示例

bi_agent = build_bi_workflow()

state = bi_agent.invoke({"query": "600519.SH"})

print(state["report"])

四、企业级部署关键技术

4.1 状态管理与记忆

使用HippoRAG记忆模块实现长期记忆：

python

from hippograph import HippoRAG

# 初始化记忆库

memory = HippoRAG(

embedding_model="text-embedding-3-small",

persist_directory="./memory_db"

)

# 存储历史案例

memory.insert("2024年600519.SH分析案例", {

"context": "白酒行业龙头股分析",

"findings": "Q2营收超预期，机构持仓增加"

})

# 相似案例检索

def get_similar_cases(query):

return memory.query(query, top_k=3)

4.2 安全管控机制

实现红队测试框架与输出验证：

python

import re

from typing import List

class SafetyValidator:

def __init__(self):

self.patterns = [

r"泄露.*个人信息",

r"提供.*投资建议", # 金融场景需合规

r"https?://[^ ]+" # 防止XSS攻击

]

def validate(self, text: str) -> List[str]:

violations = []

for pattern in self.patterns:

if re.search(pattern, text):

violations.append(f"违反安全规则：{pattern}")

return violations

# 使用示例

validator = SafetyValidator()

violations = validator.validate("推荐买入600519.SH，目标价200元")

if violations:

raise ValueError("输出包含违规内容")

五、行业落地案例解析

5.1 金融领域：智能投研Agent

通过GRPO强化学习算法训练的Agent可自主完成：

1. 调用Wind/同花顺API获取数据
2. 使用Python计算技术指标
3. 生成包含风险评估的完整报告

python

# GRPO微调示例（简化版）

from transformers import Trainer, TrainingArguments

def compute_grpo_loss(model, samples):

# 实现基于人类反馈的强化学习损失计算

# 实际中需对接偏好数据集

pass

trainer = Trainer(

model=model,

args=TrainingArguments(

output_dir="./grpo_output",

per_device_train_batch_size=2,

num_train_epochs=3

),

compute_metrics=compute_grpo_loss

)

trainer.train()

5.2 医疗领域：CT影像诊断系统

结合视觉代理+决策代理的多模态架构：

python

# 视觉代理（病灶识别）

from transformers import AutoImageProcessor, AutoModelForImageSegmentation

processor = AutoImageProcessor.from_pretrained("med-segment-base")

model = AutoModelForImageSegmentation.from_pretrained("med-segment-base")

def detect_lesions(image_path):

inputs = processor(images=image_path, return_tensors="pt")

outputs = model(**inputs)

# 返回病灶位置与类型

return postprocess(outputs)

# 决策代理（生成诊断建议）

def generate_diagnosis(lesions):

prompt = f"""根据以下CT影像分析结果生成诊断建议：

{lesions}

要求：参考最新医学指南，提供3种可能疾病及概率"""

return model.chat(prompt).content

六、未来技术趋势展望

九天菜菜课程指出，2025年后的Agent技术将呈现三大方向：

1. 多模态融合：集成语音、图像、传感器数据，实现更自然的人机交互
2. 自主进化：通过持续学习机制，使Agent具备“终身学习”能力
3. 伦理与安全：构建责任归属框架与隐私保护机制，确保技术可控性

结语

从QLoRA微调到多Agent协作，从金融投研到医疗诊断，本文展示的案例证明：未来的AI竞争将是Agent生态的竞争。开发者通过掌握九天菜菜课程中的方法论，可快速构建具备自主决策能力的智能系统。正如课程主讲人所述：“2025年的Agent开发，不是写几个脚本，而是设计能够持续进化的数字生命体。”

立即行动建议：

1. 复现本文中的股票分析Agent代码
2. 在LangChain中尝试构建Supervisor架构
3. 参与九天菜菜课程中的金融/医疗行业实战项目

技术演进永不停歇，但掌握核心方法论的开发者，将始终站在浪潮之巅。