Agent工具调用机制思考

1
2
3
4
5
6
7
8

@tool_registry.register_tool
async def extract_concepts(
    content: str, 
    method: str = "ai_enhanced", 
    max_concepts: int = 10
) -> List[Dict[str, Any]]:
    """从内容中提取核心概念和重要术语"""
    # 实现逻辑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 核心迭代循环
max_iterations = 10
current_messages = messages.copy()

while iteration < max_iterations:
    # 1. Claude决策
    response = ai_client.chat.completions.create(
        messages=current_messages,  # 包含完整历史
        tools=available_tools,
        tool_choice="auto"
    )
    
    # 2. 执行工具调用
    if message.tool_calls:
        for tool_call in message.tool_calls:
            result = await tool_registry.call_tool(
                tool_call.function.name,
                **json.loads(tool_call.function.arguments)
            )
            # 添加工具结果到对话历史
            current_messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })
    else:
        break  # Claude认为任务完成

1
2
3
4
5
6
7
8
9

try:
    tool_result = await tool_registry.call_tool(tool_name, **tool_args)
except Exception as e:
    # 错误信息返回给Claude，让其决策下一步
    tool_results.append({
        "tool_call_id": tool_call.id,
        "role": "tool",
        "content": f"工具调用失败: {str(e)}"
    })

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

async def _ai_create_processing_plan(self, content: str, content_type: str):
    """AI分析内容并制定执行计划"""
    response = await ai_client.chat.completions.create(
        model=self.flash_model,  # 使用便宜的Flash模型
        messages=[{
            "role": "system",
            "content": "你是内容处理策略专家，请返回JSON格式处理方案"
        }]
    )
    
    # 返回结构化执行计划
    return {
        "execution_phases": [
            {"phase": "analysis", "tools": ["basic_content_analyzer"]},
            {"phase": "extraction", "tools": ["concept_extractor"]},
            {"phase": "synthesis", "tools": ["markdown_structurer"]}
        ]
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

async def _execute_processing_plan(self, plan: Dict, content: str):
    results = {"content": content}
    
    for phase_config in plan["execution_phases"]:
        # 按阶段执行工具
        if phase_config.get("execution_type") == "parallel":
            phase_results = await self._execute_tools_parallel(
                phase_config["tools"], results
            )
        else:
            phase_results = await self._execute_tools_sequential(
                phase_config["tools"], results
            )
        
        results.update(phase_results)
    
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

async def process_content_smart(content: str):
    try:
        # 优先：Claude工具调用
        return await process_with_claude_tools(content)
    except Exception as e:
        logger.warning(f"Claude失败: {e}")
        try:
            # 降级：AI编排
            return await process_with_ai_orchestration(content)
        except Exception as e2:
            logger.warning(f"AI编排失败: {e2}")
            # 最终：Fallback方法
            return _create_fallback_structure(content)