小龙虾养殖大模型全栈实战：从多模态分级到Agent投喂决策

import base64
import json
from openai import OpenAI  # 兼容vLLM/OpenAI接口

client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")

def grade_crayfish(image_path: str) -> dict:
    with open(image_path, "rb") as f:
        img_b64 = base64.b64encode(f.read()).decode()
    
    prompt = """你是小龙虾品鉴专家。请根据以下标准严格评分：
    - 色泽(A/B/C): A(鲜红透亮), B(暗红), C(发黑)
    - 鳃部清洁度(1-10): 10为洁白无瑕
    - 活跃度: 尾部蜷缩力度
    必须以JSON格式返回：{"grade": "A/B/C", "gill_score": int, "activity": str, "reason": "简短依据"}"""
    
    response = client.chat.completions.create(
        model="Qwen/Qwen-VL-Chat",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}}
            ]
        }],
        temperature=0.1  # 低温度保证判定稳定性
    )
    # 提取JSON（处理模型可能输出Markdown包裹的情况）
    raw = response.choices[0].message.content
    json_str = raw.replace("```json", "").replace("```", "").strip()
    return json.loads(json_str)

# 测试
result = grade_crayfish("crayfish_sample.jpg")
print(f"分级结果：{result}")
# 输出示例：{"grade": "B", "gill_score": 6, "activity": "中等", "reason": "虾壳暗红，鳃部有轻微附着物"}

from unsloth import FastLanguageModel
import torch
from trl import SFTTrainer
from transformers import TrainingArguments
from datasets import load_dataset

# 1. 加载4bit量化模型
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Qwen2.5-1.5B",
    max_seq_length=2048,
    load_in_4bit=True,
)

# 2. 添加LoRA适配器（秩=16，仅训练0.5%参数）
model = FastLanguageModel.get_peft_model(
    model,
    r=16,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    lora_alpha=16,
    lora_dropout=0.1,
)

# 3. 加载养殖问答数据集（格式：{"instruction": "...", "output": "..."}）
dataset = load_dataset("json", data_files="crayfish_qa.json")["train"]

def format_prompt(example):
    return {"text": f"<|user|>\n{example['instruction']}\n<|assistant|>\n{example['output']}"}
dataset = dataset.map(format_prompt)

# 4. 启动训练（单卡RTX 3060约需2小时）
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    args=TrainingArguments(
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        warmup_steps=100,
        max_steps=800,
        learning_rate=2e-4,
        fp16=True,
        logging_steps=10,
        output_dir="./crayfish_lora",
    ),
)
trainer.train()
model.save_pretrained("crayfish_advisor")

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import random

# 模拟传感器数据
class EnvState(TypedDict):
    temperature: float
    ph: float
    dissolved_oxygen: float
    feed_remaining: int  # 剩余饲料克数
    logs: List[str]

# 工具函数定义
def check_sensor() -> dict:
    # 实际项目接入Modbus协议读取
    return {"temp": 29.5, "ph": 8.1, "do": 3.2}  # 溶氧低于4.0警戒

def query_vector_db(query: str) -> str:
    # 从Chroma检索相似历史处理方案
    return "历史记录：当溶氧3.0时，开启增氧机2小时可回升至5.0"

def decide_action(state: EnvState) -> str:
    # 注入微调后的养殖顾问模型
    advisor = FastLanguageModel.from_pretrained("crayfish_advisor", device_map="cuda")
    prompt = f"""当前水温{state['temperature']}℃，pH{state['ph']}，溶氧{state['dissolved_oxygen']}mg/L。
    {query_vector_db("低溶氧处理")}
    请决策：1)开启增氧机 2)投放饲料 3)换水 4)观察。输出纯文本动作。"""
    response = advisor.generate(prompt, max_new_tokens=64)
    return response.strip()

# 构建LangGraph循环
workflow = StateGraph(EnvState)
workflow.add_node("sensor_read", lambda s: {**s, **check_sensor()})
workflow.add_node("ai_decision", decide_action)
workflow.add_node("executor", lambda s: print(f"执行动作：{s['action']}") or s)

workflow.set_entry_point("sensor_read")
workflow.add_edge("sensor_read", "ai_decision")
workflow.add_edge("ai_decision", "executor")
workflow.add_edge("executor", END)
app = workflow.compile()

# 定时运行（每30分钟触发一次）
initial_state = {"temperature": 0.0, "ph": 0.0, "dissolved_oxygen": 0.0, "feed_remaining": 5000, "logs": []}
app.invoke(initial_state)