Function Call原理

你是一个助手，请调用 ocr_service 处理这张图片。
图片路径是 /uploads/hw_001.jpg
请返回 JSON 格式：{"tool": "ocr_service", "image_path": "/uploads/hw_001.jpg"}

client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "请帮我识别这张作业图片"}],
    functions=[
        {
            "name": "ocr_service",
            "description": "对上传的图片进行光学字符识别，返回文本内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "image_path": {
                        "type": "string",
                        "description": "待识别图像的服务器路径"
                    },
                    "language": {
                        "type": "string",
                        "enum": ["chinese", "english"],
                        "description": "图像中文字的语言，默认为chinese"
                    }
                },
                "required": ["image_path"]
            }
        }
    ],
    function_call="auto"  # auto, none, 或指定函数名
)

{
  "choices": [
    {
      "message": {
        "role": "assistant",
        "function_call": {
          "name": "ocr_service",
          "arguments": "{\"image_path\": \"/uploads/hw_001.jpg\", \"language\": \"chinese\"}"
        }
      }
    }
  ]
}

import json
# 伪代码示例
if response.choices[0].message.get("function_call"):
    func_name = response.choices[0].message.function_call.name
    args = json.loads(response.choices[0].message.function_call.arguments)
    if func_name == "ocr_service":
        result = ocr_service(image_path=args["image_path"], language=args.get("language", "chinese"))
        
        # 将结果作为新消息发回给 LLM
        messages.append({
            "role": "function",
            "name": func_name,
            "content": json.dumps(result)  # 必须是字符串
        })

"description": "调用OCR"

"description": "对上传的学生手写作答图片进行高精度OCR识别，支持中英文数学公式和表格结构还原，适用于作业批改场景。输入为服务器本地路径，输出为纯文本内容。"

"parameters": {
  "type": "object",
  "properties": {
    "topic": {
      "type": "string",
      "description": "知识点主题，例如'三角函数诱导公式'"
    },
    "difficulty": {
      "type": "number",
      "minimum": 1,
      "maximum": 5,
      "description": "题目难度等级，1最简单，5最难"
    },
    "count": {
      "type": "number",
      "default": 5,
      "description": "生成题目的数量"
    },
    "include_solution": {
      "type": "boolean",
      "default": true,
      "description": "是否包含详细解答过程"
    }
  },
  "required": ["topic"]
}

"output_format": {
  "type": "string",
  "enum": ["plaintext", "latex", "markdown"],
  "description": "期望的输出格式"
}

def safe_parse_arguments(raw_args: str):
    try:
        return json.loads(raw_args)
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON in function call arguments: {raw_args}") from e

def validate_ocr_params(params):
    if "image_path" not in params:
        raise ValueError("Missing required parameter: image_path")
    
    if not os.path.exists(params["image_path"]):
        raise FileNotFoundError(f"Image not found: {params['image_path']}")
    
    if "language" in params and params["language"] not in ["chinese", "english"]:
        raise ValueError("Invalid language value")

try:
    result = ocr_service(**args)
except Exception as e:
    # 返回结构化错误信息，便于 LLM 理解
    error_content = {
        "error": "ocr_failed",
        "message": str(e),
        "suggestion": "请检查图片是否清晰，或尝试重新上传"
    }
    messages.append({
        "role": "function",
        "name": "ocr_service",
        "content": json.dumps(error_content)
    })

if user_input.contains("题") and has_image():
    call_ocr_api()
else:
    search_knowledge_base()

# mini_agent.py
import json
import openai
import os
from typing import Dict, Any
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())

# ========================
# 1. 配置 OpenAI 客户端
# ========================
client = openai.OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# ========================
# 2. 定义外部工具函数（Tool Implementation）
# ========================
def ocr_service(image_path: str) -> Dict[str, Any]:
    """
    模拟OCR服务：识别图像中的文本内容
    在真实项目中，这里会调用 PaddleOCR 或云OCR API
    """
    print(f"[工具执行] 正在调用 OCR 识别图片: {image_path}")
    # 模拟返回结果
    if "math" in image_path:
        return {
            "success": True,
            "text": "题目：已知 sin(α) = 3/5，且 α ∈ (π/2, π)，求 cos(α) 的值。\n作答：cos(α) = 4/5"
        }
    else:
        return {"success": False, "error": "无法识别图像内容"}

def explain_concept(concept: str) -> Dict[str, Any]:
    """
    模拟知识讲解服务：返回某个知识点的解释
    """
    print(f"[工具执行] 正在查询知识点: {concept}")
    return {
        "explanation": f"{concept} 是指在第二象限中，正弦为正，余弦为负。常用公式：sin²α + cos²α = 1。"
    }

# ========================
# 3. 注册可用函数 Schema
# ========================
functions = [
    {
        "name": "ocr_service",
        "description": "对上传的学生手写作答图片进行OCR识别，提取题目和答案文本，适用于作业批改场景。",
        "parameters": {
            "type": "object",
            "properties": {
                "image_path": {
                    "type": "string",
                    "description": "待识别图像的服务器路径，例如 /uploads/hw_001.jpg"
                }
            },
            "required": ["image_path"]
        }
    },
    {
        "name": "explain_concept",
        "description": "根据知识点名称，提供详细的中文讲解，包括定义、公式和常见错误。",
        "parameters": {
            "type": "object",
            "properties": {
                "concept": {
                    "type": "string",
                    "description": "要解释的知识点名称，如'三角函数诱导公式'"
                }
            },
            "required": ["concept"]
        }
    }
]

# ========================
# 4. 主执行流程
# ========================
def run_agent(user_input: str):
    messages = [{"role": "user", "content": user_input}]
    while True:
        response = client.chat.completions.create(
            model="qwen-plus",  # 使用通义千问等支持function call的模型
            messages=messages,
            functions=functions,
            function_call="auto"
        )
        print(f'response: {response}')
        message = response.choices[0].message
        # 判断是否需要调用函数
        if message.function_call:
            func_name = message.function_call.name
            try:
                args = json.loads(message.function_call.arguments)
            except json.JSONDecodeError as e:
                print(f"[参数解析失败] {e}")
                break
            print(f"[决策] LLM 决定调用函数: {func_name}，参数: {args}")
            # 执行对应函数
            if func_name == "ocr_service":
                result = ocr_service(**args)
            elif func_name == "explain_concept":
                result = explain_concept(**args)
            else:
                result = {"error": f"未知函数: {func_name}"}
            # 将结果以 function 角色回传给 LLM
            print(f'message111: {message.model_dump()}')
            messages.append(message.model_dump())
            messages.append({
                "role": "function",
                "name": func_name,
                "content": json.dumps(result)
            })
            # 继续循环，让 LLM 基于结果做下一步决策
            continue
        else:
            # LLM 返回最终回复，结束循环
            print(f"\n[最终回复]\n{message.content}")
            break

# ========================
# 5. 运行测试
# ========================
if __name__ == "__main__":
    test_input = "这是我写的数学题，请帮我看看做得对吗？图片路径是 /uploads/math_hw_001.jpg"
    run_agent(test_input)